0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@taka_yayoi(Takaaki Yayoi)inデータブリックス・ジャパン株式会社

Lakeflow宣言型パイプラインとネイティブ空間型でジオフェンス到着検知を構築し、foliumで可視化する

0
Posted at

はじめに

Databricksのネイティブ空間型 (GEOMETRY / GEOGRAPHY) と組み込みの空間関数を使うと、外部ライブラリなしで地理空間ワークフローを実行できるようになっています。今回は、Lakeflow Spark宣言型パイプライン (SDP) のチュートリアルを動かしてみます。

チュートリアル: ネイティブ空間タイプを使用して地理空間パイプラインを構築する

このチュートリアルでは、以下を構築します。

  • GPSデータをAuto Loaderでブロンズのストリーミングテーブルに増分取り込み
  • 緯度・経度をネイティブのGEOMETRYポイントに変換するシルバーのストリーミングテーブル
  • WKTポリゴンからウェアハウスのジオフェンスを保持するマテリアライズドビュー
  • ST_Containsによる空間結合で「どのデバイスがどのジオフェンスに入ったか」を捕捉する到着テーブル

ブロンズ (生GPS) → シルバー (ジオメトリ化したポイント) → ゴールド (ジオフェンスと到着イベント) というメダリオンアーキテクチャの構成です。

さらに本記事の後半では、チュートリアルには含まれていないfoliumによる可視化を追加します。空間結合の結果をテーブルの件数だけで眺めていても本質が掴めないので、地図に乗せます。これが思いのほか面白い発見につながりました。

関連ドキュメント:

ステップ1: パイプラインの作成

サイドバーの新規 > ETLパイプラインからパイプラインを作成します。名前をSpatial pipeline tutorialなどに変更し、パイプライン名の右側で書き込み権限のあるカタログとスキーマをデフォルトとして選択します。以降のコードの<catalog>と<schema>は、ここで選択した値に置き換えてください。

Screenshot 2026-07-06 at 10.31.49.png

ステップ2: サンプルデータの生成

GPSピン (JSON) とウェアハウスジオフェンス (WKTポリゴンを含むJSON) のサンプルデータをUnity Catalogボリュームに生成します。アセットブラウザの追加 > ExplorationでSetup spatial dataという名前のPythonノートブックを作成し、以下を実行します。

from pyspark.sql import functions as F

catalog = "<catalog>"   # 例: "main"
schema = "<schema>"    # 例: "default"

spark.sql(f"USE CATALOG `{catalog}`")
spark.sql(f"USE SCHEMA `{schema}`")
spark.sql(f"CREATE VOLUME IF NOT EXISTS `{catalog}`.`{schema}`.`raw_data`")
volume_base = f"/Volumes/{catalog}/{schema}/raw_data"

# GPS: 2つのウェアハウスジオフェンスと重なるボックス内に5000行を生成 (LAエリア)
gps_path = f"{volume_base}/gps"
df_gps = (
    spark.range(0, 5000)
    .repartition(10)
    .select(
        F.format_string("device_%d", F.col("id").cast("long")).alias("device_id"),
        F.current_timestamp().alias("timestamp"),
        (-118.3 + F.rand() * 0.2).alias("longitude"),   # -118.3 〜 -118.1
        (34.0 + F.rand() * 0.2).alias("latitude"),     # 34.0 〜 34.2
    )
)
df_gps.write.format("json").mode("overwrite").save(gps_path)
print(f"Wrote 5000 GPS rows to {gps_path}")

# ジオフェンス: 同じリージョン内の2つのウェアハウスポリゴン (WKT)
geofences_path = f"{volume_base}/geofences"
geofences_data = [
    ("Warehouse_A", "POLYGON ((-118.35 34.02, -118.25 34.02, -118.25 34.08, -118.35 34.08, -118.35 34.02))"),
    ("Warehouse_B", "POLYGON ((-118.20 34.05, -118.12 34.05, -118.12 34.12, -118.20 34.12, -118.20 34.05))"),
]
df_geo = spark.createDataFrame(geofences_data, ["warehouse_name", "boundary_wkt"])
df_geo.write.format("json").mode("overwrite").save(geofences_path)
print(f"Wrote {len(geofences_data)} geofences to {geofences_path}")

ロサンゼルス周辺の経度-118.3〜-118.1、緯度34.0〜34.2のバウンディングボックスに一様乱数で5,000点のGPSピンを生成し、2つの矩形ジオフェンスをWKTで定義しています。このバウンディングボックスとポリゴンの位置関係が、後の可視化での発見の伏線になります。

ステップ3: ブロンズ: Auto LoaderによるGPSデータの取り込み

ボリューム上の生のGPS JSONを、Auto Loaderでブロンズのストリーミングテーブルに増分取り込みします。アセットブラウザの追加 > Transformationでgps_bronzeを作成します (言語はSQLまたはPythonを選択できます。本記事ではSQLで進めます)。

CREATE OR REFRESH STREAMING TABLE gps_bronze
COMMENT "Raw GPS pings ingested from volume using Auto Loader";

CREATE FLOW gps_bronze_ingest_flow AS
INSERT INTO gps_bronze BY NAME
SELECT *
FROM STREAM read_files(
  "/Volumes/<catalog>/<schema>/raw_data/gps",
  format => "json",
  inferColumnTypes => "true"
)

ステップ4: シルバー: ネイティブGEOMETRYポイントへの変換

ブロンズテーブルを読み込み、ST_Point(longitude, latitude)でGEOMETRY型のカラムを追加するストリーミングテーブルraw_gps_silverを作成します。

CREATE OR REFRESH STREAMING TABLE raw_gps_silver
COMMENT "GPS pings with native geometry point for spatial joins";

CREATE FLOW raw_gps_silver_flow AS
INSERT INTO raw_gps_silver BY NAME
SELECT
  device_id,
  timestamp,
  longitude,
  latitude,
  ST_Point(longitude, latitude) AS point_geom
FROM STREAM(gps_bronze)

倍精度の経度・緯度カラムから、SQLの組み込み関数だけでネイティブなジオメトリポイントを生成できます。外部ライブラリのUDFを噛ませる必要はありません。

ステップ5: ゴールド: WKTからジオフェンスのマテリアライズドビュー

ボリュームのジオフェンスJSONを読み込み、ST_GeomFromWKTでWKT文字列をGEOMETRYカラムに変換するマテリアライズドビューwarehouse_geofences_goldを作成します。

CREATE OR REPLACE MATERIALIZED VIEW warehouse_geofences_gold AS
SELECT
  warehouse_name,
  ST_GeomFromWKT(boundary_wkt) AS boundary_geom
FROM read_files(
  "/Volumes/<catalog>/<schema>/raw_data/geofences",
  format => "json"
)

ステップ6: 空間結合による到着テーブルの作成

いよいよ本丸です。シルバーのGPSポイントとジオフェンスをST_Contains(boundary_geom, point_geom)で結合し、「デバイスがウェアハウスポリゴンの内側にいる」行だけを抽出するマテリアライズドビューwarehouse_arrivalsを作成します。

CREATE OR REPLACE MATERIALIZED VIEW warehouse_arrivals AS
SELECT
  g.device_id,
  g.timestamp,
  w.warehouse_name
FROM raw_gps_silver g
JOIN warehouse_geofences_gold w
  ON ST_Contains(w.boundary_geom, g.point_geom)

結合条件が等値結合ではなく空間述語になっているのがポイントです。パイプラインを実行すると、グラフにgps_bronze、raw_gps_silver、warehouse_geofences_gold、warehouse_arrivalsの4つのデータセットが表示されます。

Screenshot 2026-07-06 at 10.32.53.png

空間結合の検証

ノートブックまたはSQLエディタで到着件数を確認します。

SELECT warehouse_name, COUNT(*) AS arrival_count
FROM warehouse_arrivals
GROUP BY warehouse_name
ORDER BY warehouse_name;

Screenshot 2026-07-06 at 10.33.27.png

Warehouse_AとWarehouse_Bの両方で非ゼロの件数が返ってくれば、空間結合は機能しています。ここで結果をよく見ると、Warehouse_Bの到着件数がWarehouse_Aを大きく上回っていることに気づきます。ポリゴン定義を見返すと、Warehouse_Aは経度0.10度 × 緯度0.06度、Warehouse_Bは経度0.08度 × 緯度0.07度で、面積はほぼ同等です。なぜ差が出るのか。件数だけ見ていても分かりません。地図に乗せましょう。

foliumによる可視化

GEOMETRY型をPythonの可視化ライブラリに渡すには

ここで1つ、ネイティブ空間型ならではのポイントがあります。GEOMETRY型のカラムを含むDataFrameをそのままtoPandas()するとArrow変換でエラーになるため、SQL側でST_AsGeoJSONやST_AsWKTを使って文字列に変換してからPythonに渡す必要があります。ネイティブ空間型 × Python可視化の定番パターンとして覚えておくと良いです。

もう1つ、warehouse_arrivalsには座標カラムが残っていない (device_id、timestamp、warehouse_nameのみ) ので、可視化用にはパイプラインと同じST_ContainsをLEFT JOINで再実行し、どのジオフェンスにも入らなかった「圏外」の点も残すクエリを組みます。パイプラインの結合ロジックそのものを地図上で検証する形になります。

foliumのインストール

%pip install folium
dbutils.library.restartPython()

データの取得

import json
import pandas as pd

catalog = "<catalog>"   # パイプラインと同じ値
schema = "<schema>"

spark.sql(f"USE CATALOG `{catalog}`")
spark.sql(f"USE SCHEMA `{schema}`")

# ジオフェンス: GEOMETRY型はそのままtoPandas()できないので ST_AsGeoJSON で文字列化
geofences_pdf = spark.sql("""
    SELECT
      warehouse_name,
      ST_AsGeoJSON(boundary_geom) AS boundary_geojson
    FROM warehouse_geofences_gold
""").toPandas()

# GPSポイント: パイプラインと同じ空間結合をLEFT JOINで実行し、圏外の点も残す
# (warehouse_name が NULL = どのジオフェンスにも入っていない = warehouse_arrivals に載らない点)
points_pdf = spark.sql("""
    SELECT
      g.device_id,
      g.longitude,
      g.latitude,
      w.warehouse_name
    FROM raw_gps_silver g
    LEFT JOIN warehouse_geofences_gold w
      ON ST_Contains(w.boundary_geom, g.point_geom)
""").toPandas()

print(f"points: {len(points_pdf)}, arrivals: {points_pdf['warehouse_name'].notna().sum()}")

地図の描画

import folium

colors = {"Warehouse_A": "#1baf7a", "Warehouse_B": "#4a3aa7"}
OUTSIDE_COLOR = "#888888"

center = [points_pdf["latitude"].mean(), points_pdf["longitude"].mean()]
m = folium.Map(location=center, zoom_start=11, tiles="cartodbpositron")

# ジオフェンスポリゴン
for _, row in geofences_pdf.iterrows():
    color = colors.get(row["warehouse_name"], "#378add")
    folium.GeoJson(
        json.loads(row["boundary_geojson"]),
        name=row["warehouse_name"],
        style_function=lambda f, c=color: {
            "color": c,
            "weight": 2,
            "fillColor": c,
            "fillOpacity": 0.15,
        },
        tooltip=row["warehouse_name"],
    ).add_to(m)

# GPSポイント (5000点そのままでも描けるが、重い場合はサンプリング)
plot_pdf = points_pdf.sample(n=min(len(points_pdf), 3000), random_state=42)

for _, row in plot_pdf.iterrows():
    wname = row["warehouse_name"] if pd.notna(row["warehouse_name"]) else None
    folium.CircleMarker(
        location=[row["latitude"], row["longitude"]],
        radius=2.5,
        stroke=False,
        fill=True,
        fill_color=colors.get(wname, OUTSIDE_COLOR),
        fill_opacity=0.85 if wname else 0.25,
        tooltip=f"{row['device_id']}: {wname or '圏外'}",
    ).add_to(m)

# 凡例
legend_html = """
<div style="position: fixed; bottom: 24px; left: 24px; z-index: 9999;
     background: white; padding: 10px 14px; border: 1px solid #ccc;
     border-radius: 6px; font-size: 13px;">
  <div><span style="color:#1baf7a;">&#9679;</span> Warehouse_A に到着</div>
  <div><span style="color:#4a3aa7;">&#9679;</span> Warehouse_B に到着</div>
  <div><span style="color:#888888;">&#9679;</span> ジオフェンス外</div>
</div>
"""
m.get_root().html.add_child(folium.Element(legend_html))

displayHTML(m._repr_html_())

実装上の細かい注意点です。

  • ループ内でstyle_functionのlambdaを作るときはc=colorとデフォルト引数でキャプチャしないと、全ポリゴンが最後の色になります (Pythonの古典的な罠)
  • サーバレスノートブックでは最後の行をmだけにしても描画されないことがあるので、displayHTML(m._repr_html_())が確実です

結果

Screenshot 2026-07-06 at 10.26.41.png

一目瞭然でした。Warehouse_A (緑) のポリゴンの西半分に、点が1つも入っていません。

種明かしをすると、GPSポイントの生成範囲は経度-118.3からなのに対し、Warehouse_Aのポリゴンは-118.35まで西に伸びています。つまりポリゴンの約半分がデータの生成範囲外に落ちているのです。ポリゴンの面積自体はA・Bでほぼ同等でも、GPSポイントが存在しうる実効面積では、データ範囲に完全に収まっているWarehouse_Bが上回る。到着件数の差の正体はこれでした。

集計クエリの結果だけ見て「Bの方がAより到着が多い」と報告して終わっていたら、この構造には気づけませんでした。空間結合の結果は地図に乗せて初めて正しく解釈できる、という教訓です。

まとめ

  • Lakeflow Spark宣言型パイプラインとネイティブ空間型を使うと、外部ライブラリなしでAuto Loader取り込み → ST_Pointによるジオメトリ化 → ST_Containsによる空間結合、というジオフェンス到着検知パイプラインを宣言的に構築できます
  • GEOMETRY型カラムはST_AsGeoJSONで文字列化してからtoPandas()するのがPython可視化の定番パターンです
  • 空間結合の結果は集計値だけでなく地図で確認しましょう。ポリゴンとデータのバウンディングボックスの位置関係のような、数字からは見えない構造が浮かび上がります

次のステップとしては、5,000点をH3関数でヘックスに集約した密度マップや、データ生成時のタイムスタンプにランダムオフセットを加えて到着の時系列分析に発展させる、といった方向が考えられます (チュートリアルのサンプルデータはtimestampが全行current_timestamp()のため、そのままでは時系列の可視化は意味を成しません)。

はじめてのDatabricks

はじめてのDatabricks

Databricks無料トライアル

Databricks無料トライアル

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?