使用 IRIS Cloud Document 和 Databricks 绘制 Rivian 地理位置图-Python教程-PHP中文网

Rivian GeoLocation Plotting with IRIS Cloud Document and Databricks

使用 InterSystems Cloud Document 和 Databricks 绘制来自我的 Rivian R1S 的密歇根州 gnSSLlocation 数据

如果您一直在寻找文档数据库的用例，我意识到我最喜欢的非常简单的一个是能够使用sql查询一堆JSON，与我的其他数据一起没有真正做太多事情。这是通过强大的多模型 InterSystems 数据平台实现的梦想，并在一个简单的笔记本中显示，以可视化我的 Rivian R1S 为 DeezWatts（Rivian 数据冒险）发出的地理位置数据。

因此，这是两步方法，使用 JDBC 文档驱动程序摄取到并可视化来自

InterSystems 云文档。

InterSystems 云文档部署

首先，我在 InterSystems 云服务门户上启动了一个小型云文档部署，并启用了侦听器。

Rivian GeoLocation Plotting with IRIS Cloud Document and Databricks

我下载了 ssl 证书，并获取了 JDBC 驱动程序和随附的文档驱动程序。

摄入

对于摄取，我想掌握如何从文件系统中提取 JSON 文档并将其作为集合保存在侦听器上的文档数据库中，为此我编写了一个独立的 java 应用程序。这更实用，因为数据存入笔记本后，所有乐趣都发生在笔记本中。

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

登录后复制

上面的内容与JAVA垃圾箱非常接近，但是有效，我们可以在部署中的集合浏览器中看到集合。

Rivian GeoLocation Plotting with IRIS Cloud Document and Databricks

数据块

现在这需要一些 Databricks 设置，但与 pyspark 一起工作以获得有趣的部分是非常值得的。

我将两个 InterSystems 驱动程序添加到集群中，并将证书放入 import_cloudsql_certficiate.sh 集群初始化脚本中，以便将其添加到密钥库中。

Rivian GeoLocation Plotting with IRIS Cloud Document and Databricks

为了完整起见，集群运行的是 Databricks 16、Spark 3.5.0 和 Scala 2.12

Rivian GeoLocation Plotting with IRIS Cloud Document and Databricks

可视化

所以我们应该设置运行 PySpark 作业并绘制我的鞭子在我将拖入的数据子集中的位置。

我们使用 geopandas 和地理数据集来进行直接的绘图。

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

登录后复制

现在，这需要一点时间来适应，但这里是使用 JSON 路径语法和 JSON_TABLE 对 InterSystems Cloud Document 的查询。

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

登录后复制

我确实找到了一个网站，可以非常简单地创建 json 路径 @ jsonpath.com。

Rivian GeoLocation Plotting with IRIS Cloud Document and Databricks

接下来我们设置与 IRIS 文档数据库部署的连接并将其读入数据帧。

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

登录后复制

接下来我们从地理数据集中获取一张可用的地图，sdoh 非常适合美国的通用使用。

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

登录后复制

现在最酷的部分是，我们想要放大我们想要包含 R1S 行驶过的地理位置点的位置，为此我们需要一个密歇根州的边界框。

为此，我使用了 Keene 的一个非常灵活的工具来绘制地理围栏边界框，它给了我坐标数组！

Rivian GeoLocation Plotting with IRIS Cloud Document and Databricks

现在我们有了边界框的坐标数组，我们需要将它们放入 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"
)

登录后复制

现在，让我们绘制 Rivian R1S 的轨迹！这将用于大约 10,000 条记录（我使用上面的 top 语句来限制结果）

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

登录后复制

我们就有了...底特律、特拉弗斯城、银湖沙丘、荷兰、鲻鱼湖、因特拉肯...纯正的密歇根、Rivian 风格。

Rivian GeoLocation Plotting with IRIS Cloud Document and Databricks