Rivian GeoLocation Plotting mit IRIS Cloud Document und Databricks

Plot der gnSSLocation-Daten von meinem Rivian R1S in ganz Michigan mit InterSystems Cloud Document und Databricks

Wenn Sie nach einem Anwendungsfall für eine Dokumentendatenbank gesucht haben, bin ich zu der Erkenntnis gelangt, dass mein absolut einfacher Lieblingsfall die Möglichkeit ist, einen Stapel JSON direkt neben meinen anderen Daten mit SQL abzufragen ohne wirklich viel zu tun. Das ist der Traum, der mit der leistungsstarken Multi Model InterSystems Data Platform verwirklicht wurde und hier in einem einfachen Notizbuch gezeigt wird, um meine geografischen Standortdaten zu visualisieren, die mein Rivian R1S für DeezWatts (ein Rivian-Datenabenteuer) aussendet.

Hier ist also der zweistufige Ansatz: Aufnahme zu und Visualisierung aus InterSystems Cloud Document unter Verwendung des JDBC-Dokumenttreibers.

InterSystems Cloud Document Deployment

Zunächst habe ich eine kleine Cloud-Dokumentbereitstellung auf dem InterSystems Cloud Services Portal mit aktiviertem Listener gestartet.

Ich habe das SSL-Zertifikat heruntergeladen und mir auch die Treiber für JDBC und den dazugehörigen Dokumententreiber besorgt.

Einnahme

Für die Aufnahme wollte ich in den Griff bekommen, wie man ein JSON-Dokument aus dem Dateisystem holt und es als Sammlung in der Dokumentdatenbank über den Listener beibehält. Dafür habe ich eine eigenständige Java-App geschrieben. Das war nützlicher, da der Spaß erst im Notizbuch stattfand, nachdem die Daten dort gespeichert waren.
 

 
RivianDocDB.java

package databricks_rivian_irisdocdb;

import java.sql.SQLException;
import com.intersystems.document.*;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.*;
import java.io.IOException;
import java.io.InputStream;
import java.io.File;
import java.io.FileInputStream;
import org.apache.commons.io.IOUtils;

import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.stream.Stream;

public <span>class RivianDocDb </span>{
  <span>public static void main(String[] args) </span>{

    String directoryPath =
"/home/sween/Desktop/POP2/DEEZWATTS/rivian-iris-docdb/databricks_rivian_irisdocdb/in/json/";

    DataSource datasrc = DataSource.createDataSource();
    datasrc.setPortNumber(443);
    datasrc.setServerName("k8s-05868f04-a88b7ecb-5c5e41660d-404345a22ba1370c.elb.us-east-1.amazonaws.com");
    datasrc.setDatabaseName("USER");
    datasrc.setUser("SQLAdmin");
    datasrc.setPassword("REDACTED");

    try {
      datasrc.setConnectionSecurityLevel(10);
    } catch (SQLException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }

    System.out.println("\nCreated datasrc\n");
    System.out.println(datasrc);
    datasrc.preStart(2);
    System.out.println("\nDataSource size =" + datasrc.getSize());

    // creates the collection if it dont exist
    Collection collectedDocs =
Collection.getCollection(datasrc,"deezwatts2");

    try (Stream<Path> paths = Files.list(Paths.get(directoryPath))) {
        paths.filter(Files::isRegularFile)
             .forEach(path -> {
                 File file = path.toFile();
             });
    } catch (IOException e) {
        e.printStackTrace();
    }

    File directory = new File(directoryPath);
    if (directory.isDirectory()) {
        File[] files = directory.listFiles();
        if (files != null) {
            for (File file : files) {
                if (file.isFile()) {

                    try (InputStream is = new
FileInputStream("/home/sween/Desktop/POP2/DEEZWATTS/rivian-iris-docdb/databricks_rivian_irisdocdb/in/json/"
+ file.getName())) {
                      String jsonTxt = IOUtils.toString(is, "UTF-8");
                      
                      Document doc2 = JSONObject.fromJSONString(jsonTxt);
                      // top level key is whip2
                      Document doc3 = new JSONObject().put("whip2",doc2);

                      collectedDocs.insert(doc3);
                    } catch (IOException e) {
                      // TODO Auto-generated catch block
                      e.printStackTrace();
                    }

                }
            }
        }
    }


    long size = collectedDocs.size();
    System.out.println(Long.toString(size));
    System.out.println("\nIngested Documents =" + datasrc.getSize());

 

Das Obige kommt dem JAVA-Papierkorb ziemlich nahe, hat aber funktioniert, wir können die Sammlung im Sammlungsbrowser in der Bereitstellung sehen.

Databricks

Dies erfordert ein wenig Databricks-Einrichtung, aber es lohnt sich auf jeden Fall, für den unterhaltsamen Teil mit Pyspark zu arbeiten.

Ich habe die beiden InterSystems-Treiber zum Cluster hinzugefügt und das Zertifikat in das Cluster-Init-Skript import_cloudsql_certficiate.sh eingefügt, damit es dem Keystore hinzugefügt wird.

Der Vollständigkeit halber: Auf dem Cluster werden Databricks 16, Spark 3.5.0 und Scala 2.12 ausgeführt

Visualisierung

Wir sollten also bereit sein, einen PySpark-Job auszuführen und zu zeichnen, wo sich mein Whip in der Teilmenge der Daten befindet, die ich hineinziehen werde.

Wir verwenden Geopandas und Geodatensätze für einen unkomplizierten Ansatz beim Plotten.

package databricks_rivian_irisdocdb;

import java.sql.SQLException;
import com.intersystems.document.*;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.*;
import java.io.IOException;
import java.io.InputStream;
import java.io.File;
import java.io.FileInputStream;
import org.apache.commons.io.IOUtils;

import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.stream.Stream;

public <span>class RivianDocDb </span>{
  <span>public static void main(String[] args) </span>{

    String directoryPath =
"/home/sween/Desktop/POP2/DEEZWATTS/rivian-iris-docdb/databricks_rivian_irisdocdb/in/json/";

    DataSource datasrc = DataSource.createDataSource();
    datasrc.setPortNumber(443);
    datasrc.setServerName("k8s-05868f04-a88b7ecb-5c5e41660d-404345a22ba1370c.elb.us-east-1.amazonaws.com");
    datasrc.setDatabaseName("USER");
    datasrc.setUser("SQLAdmin");
    datasrc.setPassword("REDACTED");

    try {
      datasrc.setConnectionSecurityLevel(10);
    } catch (SQLException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }

    System.out.println("\nCreated datasrc\n");
    System.out.println(datasrc);
    datasrc.preStart(2);
    System.out.println("\nDataSource size =" + datasrc.getSize());

    // creates the collection if it dont exist
    Collection collectedDocs =
Collection.getCollection(datasrc,"deezwatts2");

    try (Stream<Path> paths = Files.list(Paths.get(directoryPath))) {
        paths.filter(Files::isRegularFile)
             .forEach(path -> {
                 File file = path.toFile();
             });
    } catch (IOException e) {
        e.printStackTrace();
    }

    File directory = new File(directoryPath);
    if (directory.isDirectory()) {
        File[] files = directory.listFiles();
        if (files != null) {
            for (File file : files) {
                if (file.isFile()) {

                    try (InputStream is = new
FileInputStream("/home/sween/Desktop/POP2/DEEZWATTS/rivian-iris-docdb/databricks_rivian_irisdocdb/in/json/"
+ file.getName())) {
                      String jsonTxt = IOUtils.toString(is, "UTF-8");
                      
                      Document doc2 = JSONObject.fromJSONString(jsonTxt);
                      // top level key is whip2
                      Document doc3 = new JSONObject().put("whip2",doc2);

                      collectedDocs.insert(doc3);
                    } catch (IOException e) {
                      // TODO Auto-generated catch block
                      e.printStackTrace();
                    }

                }
            }
        }
    }


    long size = collectedDocs.size();
    System.out.println(Long.toString(size));
    System.out.println("\nIngested Documents =" + datasrc.getSize());

Es ist etwas gewöhnungsbedürftig, aber hier ist die Abfrage an InterSystems Cloud Document unter Verwendung der JSON-Pfadsyntax und JSON_TABLE.

import geopandas as gpd
import geodatasets
from shapely.geometry import Polygon

 

Ich habe es geschafft, eine Website zu finden, die es kinderleicht macht, den JSON-Pfad zu erstellen @ jsonpath.com.

Als nächstes richten wir die Verbindung zum IRIS Document Database Deployment ein und lesen es in einen Datenrahmen ein.

dbtablequery = f"(SELECT TOP 1000 lat,longitude FROM JSON_TABLE(deezwatts2 FORMAT COLLECTION, '$' COLUMNS (lat VARCHAR(20) path '$.whip2.data.vehicleState.gnssLocation.latitude', longitude VARCHAR(20) path '$.whip2.data.vehicleState.gnssLocation.longitude' ))) AS temp_table;"

Als nächstes holen wir uns eine verfügbare Karte aus Geodatensätzen. Die SDOH-Karte eignet sich hervorragend für die allgemeine Verwendung der Vereinigten Staaten.
 

# Read data from InterSystems Document Database via query above
df = (spark.read.format("jdbc") \
  .option("url", "jdbc:IRIS://k8s-05868f04-a88b7ecb-5c5e41660d-404345a22ba1370c.elb.us-east-1.amazonaws.com:443/USER") \
  .option("jars", "/Volumes/cloudsql/iris/irisvolume/intersystems-document-1.0.1.jar") \
  .option("driver", "com.intersystems.jdbc.IRISDriver") \
  .option("dbtable", dbtablequery) \
  .option("sql", "SELECT * FROM temp_table;") \
  .option("user", "SQLAdmin") \
  .option("password", "REDACTED") \
  .option("connection security level","10") \
  .option("sslConnection","true") \
  .load())

Jetzt kommt der coole Teil: Wir wollen die Stelle vergrößern, an der wir die geografischen Standortpunkte der Orte, an denen der R1S gefahren ist, einschließen wollen. Dazu benötigen wir einen Begrenzungsrahmen für den Bundesstaat Michigan.

Hierfür habe ich ein wirklich raffiniertes Werkzeug von Keene verwendet, um den Geo-Zaun-Begrenzungsrahmen zu zeichnen, und es liefert mir das Koordinaten-Array!

Da wir nun das Koordinaten-Array des Begrenzungsrahmens haben, müssen wir sie in ein Polygon-Objekt einfügen.

# sdoh map is fantastic with bounding boxes
michigan = gpd.read_file(geodatasets.get_path("geoda.us_sdoh"))

gdf = gpd.GeoDataFrame(
    df.toPandas(), 
    geometry=gpd.points_from_xy(df.toPandas()['longitude'].astype(float), df.toPandas()['lat'].astype(float)), 
    crs=michigan.crs #"EPSG:4326"
)

 

Lasst uns nun die Spur des Rivian R1S planen! Dies gilt für etwa 10.000 Datensätze (ich habe oben eine Top-Anweisung verwendet, um die Ergebnisse einzuschränken)
 

polygon = Polygon([
      (
        -87.286377,
        45.9664245
      ),
      (
        -81.6503906,
        45.8134865
      ),
      (
        -82.3864746,
        42.1063737
      ),
      (
        -84.7814941,
        41.3520721
      ),
      (
        -87.253418,
        42.5045029
      ),
      (
        -87.5610352,
        45.8823607
      )
    ])

 

Und da haben wir es... Detroit, Traverse City, Silver Lake Sand Dunes, Holland, Mullet Lake, Interlachen... Reiner Michigan-Rivian-Stil.

Das obige ist der detaillierte Inhalt vonRivian GeoLocation Plotting mit IRIS Cloud Document und Databricks. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!