Traçage de géolocalisation Rivian avec IRIS Cloud Document et Databricks

Traçage des données gnSSLocation de mon Rivian R1S à travers le Michigan avec InterSystems Cloud Document et Databricks

Si vous recherchiez un cas d'utilisation pour une base de données de documents, j'ai réalisé que mon cas d'utilisation le plus simple préféré est la possibilité d'interroger une pile de JSON, juste à côté de mes autres données avec sql sans vraiment faire grand chose. Quel est le rêve réalisé à partir de la puissante plate-forme de données multi-modèles InterSystems, et présenté ici dans un simple cahier pour visualiser mes données de géolocalisation que mon Rivian R1S émet pour DeezWatts (A Rivian Data Adventure).

Voici donc l'approche en 2 étapes, Ingestion vers et Visualisation à partir de InterSystems Cloud Document, à l'aide du pilote de document JDBC.

Déploiement de documents Cloud InterSystems

Pour commencer, j'ai lancé un petit déploiement Cloud Document sur le portail InterSystems Cloud Services, avec un écouteur activé.

J'ai téléchargé le certificat SSL et récupéré les pilotes pour JDBC et le pilote du document qui l'accompagne.

Ingestion

Pour l'ingestion, je voulais comprendre comment extraire un document JSON du système de fichiers et le conserver en tant que collection dans la base de données de documents sur l'écouteur, pour cela j'ai écrit une application Java autonome. C'était plus utile car tout le plaisir se déroulait dans le bloc-notes une fois que les données y étaient.
 

 
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());</path>

 

Ce qui précède est assez proche de la corbeille JAVA, mais a fonctionné, nous pouvons voir la collection dans le navigateur de collection lors du déploiement.

Databricks

Maintenant, cela nécessite un peu de configuration de Databricks, mais cela vaut la peine de travailler avec pyspark pour la partie amusante.

J'ai ajouté les deux pilotes InterSystems au cluster et mis le certificat dans le script d'initialisation du cluster import_cloudsql_certficiate.sh pour qu'il soit ajouté au magasin de clés.

Pour être complet, le cluster exécute Databricks 16, Spark 3.5.0 et Scala 2.12

Visualisation

Nous devrions donc être prêts à exécuter une tâche PySpark et à tracer où se trouve mon fouet dans le sous-ensemble de données que je vais glisser.

Nous utilisons des géopandas et des jeux de données géographiques pour une approche simple du traçage.

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());</path>

Maintenant, cela prend un peu de temps pour s'y habituer, mais voici la requête vers InterSystems Cloud Document en utilisant la syntaxe des chemins JSON et JSON_TABLE.

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

 

J'ai réussi à trouver un site qui simplifie grandement la création du chemin json @ jsonpath.com.

Ensuite, nous configurons la connexion au déploiement de la base de données de documents IRIS et la lisons dans une trame de données.

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;"

Ensuite, nous récupérons une carte disponible à partir des jeux de données géographiques, celle de sdoh est idéale pour une utilisation générique des États-Unis.
 

# 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())

Maintenant, la partie intéressante, nous voulons zoomer sur l'endroit où nous voulons contenir les points de géolocalisation de l'endroit où le R1S a conduit, pour cela, nous avons besoin d'un cadre de délimitation pour l'état du Michigan.

Pour cela, j'ai utilisé un outil vraiment astucieux de Keene pour dessiner le cadre de délimitation de la géo-clôture et il me donne le tableau de coordonnées !

Maintenant que nous avons le tableau de coordonnées du cadre de sélection, nous devons les placer dans un objet Polygon.

# 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"
)

 

Maintenant, traçons la trace du Rivian R1S ! Cela concernera environ 10 000 enregistrements (j'ai utilisé une déclaration supérieure ci-dessus pour limiter les résultats)
 

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
      )
    ])

 

Et voilà... Detroit, Traverse City, Silver Lake Sand Dunes, Holland, Mullet Lake, Interlachen... Pur Michigan, style Rivian.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!