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

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

Si vous cherchez un cas d'utilisation pour une Document Database, j'ai compris que mon cas préféré, très simple, est la possibilité d'interroger un paquet de JSON, juste à côté de mes autres données avec sql sans vraiment faire grand-chose. C'est un rêve réalisé à partir de la puissante plateforme de données multi-modèles d'InterSystems, et montré ici dans un simple carnet pour visualiser mes données de localisation géographique que mon Rivian R1S émet pour DeezWatts ( Une aventure de données Rivian ).

Voici donc l'approche en 2 étapes, à savoir l'ingestion à et la visualisation de InterSystems Cloud Document, à l'aide du pilote document JDBC.

Déploiement de documents dans InterSystems Cloud

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

J'ai téléchargé le certificat ssl, et j'ai récupéré les drivers pour JDBC ainsi que le pilote de document qui l'accompagne.

Ingestion

Pour l'ingestion, j'ai voulu comprendre comment extraire un document JSON du système de fichiers et le faire persister en tant que collection dans la base de données de documents via le listener, pour cela j'ai écrit une application Java autonome. Cette approche était plus utilitaire, car tout ce qui était amusant se passait dans le carnet une fois que les données y étaient placées.
 

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 class RivianDocDb {
  public static void main(String[] args) {

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

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

Databricks

Cela demande un peu d'installation de Databricks, mais cela vaut la peine de travailler avec pyspark la partie amusante.

J'ai ajouté les deux pilotes InterSystems au cluster, et j'ai placé le certificat dans le script d'initialisation du cluster import_cloudsql_certficiate.sh afin qu'il soit ajouté au keystore.

Pour être complet, le cluster utilise Databricks 16, Spark 3.5.0 et Scala 2.12

Visualization

Nous devrions donc être prêts à exécuter une tâche PySpark et à tracer l'endroit où mon fouet a été dans le sous-ensemble de données que je traîne.

Nous utilisons l'ensemble de données de base géo geopandas and geodatasets pour une approche directe de la représentation graphique.

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

Il faut un peu de temps pour s'y habituer, mais voici la requête adressée à InterSystems Cloud Document en utilisant la syntaxe des chemins JSON et JSON_TABLE.

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

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

Ensuite, nous établissons la connexion avec le déploiement de la base de données documentaire IRIS Document Database et nous lisons les données dans un cadre de données.

# Read data from InterSystems Lecture des données de la base de données documentaire InterSystems Document Database via la requête ci-dessus
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())

Ensuite, nous prenons une carte disponible à partir de jeux de données géographiques, le jeu sdoh est parfait pour une utilisation générique des États-Unis.
 

# La carte sdoh est fantastique grâce aux boîtes de délimitation
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, la partie la plus intéressante, nous voulons zoomer sur l'endroit où nous voulons placer les points de géolocalisation de l'endroit où le R1S a roulé, pour cela nous avons besoin d'une boîte de délimitation pour l'état du Michigan.

J'ai utilisé un outil très astucieux de Keene pour dessiner la boîte de délimitation de la géo clôture et cela me donne le tableau de coordonnées!

Maintenant le tableau de coordonnées de la boîte de délimitation est en notre possession, il nous faut l'insérer dans un objet Polygone.

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

Maintenant, traçons la piste du Rivian R1S! Il s'agit d'environ 10 000 enregistrements (j'ai utilisé l'instruction top ci-dessus pour limiter les résultats)
 

ax = michigan.clip(polygon).plot(color="lightblue", alpha=0.5,linewidth=0.8, edgecolor='gray')
ax.axis('off')
ax.annotate("Data: Rivian R1S Telemetry Data via InterSystems Document Database", xy=(0.01, .085), xycoords='figure fraction', fontsize=14, color='#555555')

gdf.plot(ax=ax, color="red", markersize=1.50, alpha=0.5, figsize=(200,200))

Et voilà, nous l'avons... Détroit, Traverse City, Silver Lake Sand Dunes, Holland, Mullet Lake, Interlachen... Le Michigan à l'état pur, à la Rivian.