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

GPS データを取り込み、座標をネイティブな空間型に変換し、ウェアハウスのジオフェンスに対して結合することで到着を追跡するパイプラインを作成して展開する方法について、Lakeflow Spark 宣言型パイプライン (SDP) と Auto Loader を用いたデータオーケストレーションの手法を学びます。このチュートリアルでは、Databricks ネイティブ空間型 (GEOMETRY、 GEOGRAPHY) と、 ST_Point、 ST_GeomFromWKT、 ST_Containsなどの組み込みの空間関数を使用するため、外部ライブラリなしで大規模に地理空間ワークフローを実行できます。

このチュートリアルでは、次のことを行います。

パイプラインを作成し、Unity カタログボリュームでサンプル GPS およびジオフェンスデータを生成します。
自動ローダーを使用して生の GPS ping をブロンズストリーミングテーブルにインクリメンタルに取り込みます。
緯度と経度をネイティブ GEOMETRY ポイントに変換するシルバーストリーミングテーブルを作成します。
WKT ポリゴンから倉庫ジオフェンスの具体化されたビューを作成します。
空間結合を実行して、倉庫到着 (どのデバイスがどのジオフェンスに入ったか) のテーブルを生成します。

その結果、メダリオンスタイルのパイプラインとして、ブロンズ (生の GPS)、シルバー (ジオメトリとしてのポイント)、ゴールド (ジオフェンスと到着イベント) が生成されます。詳細については、「 medallion lakehouse のアーキテクチャとは」を参照してください。

必要条件

このチュートリアルを完了するには、次の要件を満たす必要があります。

Azure Databricks ワークスペースにログインします。
ワークスペースに対して Unity カタログを有効にします。
サーバーレス Lakeflow Spark 宣言パイプラインを使用する場合は、アカウントでサーバーレスコンピューティングを有効にします。サーバーレスコンピューティングが有効になっていない場合、手順はワークスペースの既定のコンピューティングで動作します。
コンピューティングリソースを作成したり、コンピューティングリソースにアクセスしたりするためのアクセス許可を持っている。
カタログに新しいスキーマを作成するためのアクセス許可を持っている。必要なアクセス許可は USE CATALOG と CREATE SCHEMA です。
既存のスキーマに新しいボリュームを作成するためのアクセス許可を持っている。必要なアクセス許可は USE SCHEMA と CREATE VOLUME です。
ネイティブ空間型と空間関数をサポートするランタイムを使用します。

手順 1: パイプラインを作成する

新しい ETL パイプラインを作成し、テーブルの既定のカタログとスキーマを設定します。

ワークスペースの左上隅でをクリックして新規作成します。
[ ETL パイプライン] をクリックします。
パイプラインのタイトルを Spatial pipeline tutorial または希望する名前に変更します。
タイトルの下で、書き込みアクセス許可があるカタログとスキーマを選択します。

このカタログとスキーマは、コードでカタログまたはスキーマを指定しない場合に既定で使用されます。次の手順の <catalog> と <schema> を、ここで選択した値に置き換えます。
[詳細設定] オプションで、[空のファイルで開始] を選択します。
コードのフォルダーを選択します。 [参照] を選択してフォルダーを選択できます。バージョン管理には Git フォルダーを使用できます。
最初のファイルの言語として Python または SQL を選択します。後で他の言語でファイルを追加できます。
[ 選択 ] をクリックしてパイプラインを作成し、Lakeflow パイプラインエディターを開きます。

これで、既定のカタログとスキーマを含む空のパイプラインが作成されました。次に、GPS とジオフェンスのサンプルデータを作成します。

手順 2: GPS とジオフェンスのサンプルデータを作成する

この手順では、生の GPS ping (JSON) とウェアハウスジオフェンス (WKT ポリゴンを含む JSON) のサンプルデータをボリュームに生成します。 GPS ポイントは、2 つの倉庫ポリゴンと重なる境界ボックスで生成されるため、後の手順で空間結合を行うと到着行が返されます。ボリュームまたはテーブルに独自のデータが既にある場合は、この手順をスキップできます。

Lakeflow Pipelines エディターのアセットブラウザーで、をクリックし、追加、次に探索を選択します。
[名前] を [Setup spatial data] に設定し、[Python] を選択し、既定の保存先フォルダーのままにします。
Create をクリックしてください。

新しいノートブックに、次のコードを貼り付けます。 <catalog>と<schema>を、手順 1 で設定した既定のカタログとスキーマに置き換えます。

ノートブックで次のコードを使用して、GPS データとジオフェンスデータを生成します。

from pyspark.sql import functions as F

catalog = "<catalog>"   # for example, "main"
schema = "<schema>"    # for example, "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: 5000 rows in a box that overlaps both warehouse geofences (LA area)
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 to -118.1
        (34.0 + F.rand() * 0.2).alias("latitude"),     # 34.0 to 34.2
    )
)
df_gps.write.format("json").mode("overwrite").save(gps_path)
print(f"Wrote 5000 GPS rows to {gps_path}")

# Geofences: two warehouse polygons (WKT) in the same region
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}")

ノートブックのセルを実行します (Shift + Enter キーを押します)。

実行が完了すると、ボリュームに gps (生 ping) と geofences (WKT の多角形) が含まれます。次の手順では、GPS データをブロンズテーブルに取り込みます。

手順 3: ブロンズストリーミングテーブルに GPS データを取り込む

自動ローダーを使用してボリュームから生の GPS JSON を増分的に取り込み、ブロンズストリーミングテーブルに書き込みます。

資産ブラウザーで、をクリックし、次に[追加]、それから[変換]の順に選択します。
[名前] を [gps_bronzeに設定し、SQL または Python を選択して、[作成] をクリックします。

ファイルの内容を次のように置き換えます (言語に一致するタブを使用してください)。 <catalog>と<schema>を既定のカタログとスキーマに置き換えます。

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

Python

from pyspark import pipelines as dp

path = "/Volumes/<catalog>/<schema>/raw_data/gps"

dp.create_streaming_table(
  name="gps_bronze",
  comment="Raw GPS pings ingested from volume using Auto Loader",
)

@dp.append_flow(target="gps_bronze", name="gps_bronze_ingest_flow")
def gps_bronze_ingest_flow():
    return (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .load(path)
    )

[ ファイルを実行 するか 、パイプラインを実行 して更新プログラムを実行します。

更新が完了すると、パイプライングラフに gps_bronze テーブルが表示されます。次に、座標をネイティブジオメトリポイントに変換するシルバーテーブルを追加します。

手順 4: ジオメトリポイントを含むシルバーストリーミングテーブルを追加する

ブロンズテーブルから読み取り、GEOMETRYを使用してST_Point(longitude, latitude)列を追加するストリーミングテーブルを作成します。

資産ブラウザーで、をクリックし、次に[追加]、それから[変換]の順に選択します。
[名前] を [raw_gps_silverに設定し、SQL または Python を選択して、[作成] をクリックします。

次のコードを新しいファイルに貼り付けます。

SQL

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)

Python

from pyspark import pipelines as dp
from pyspark.sql import functions as F

dp.create_streaming_table(
  name="raw_gps_silver",
  comment="GPS pings with native geometry point for spatial joins",
)

@dp.append_flow(target="raw_gps_silver", name="raw_gps_silver_flow")
def raw_gps_silver_flow():
    return (
        spark.readStream.table("gps_bronze")
        .select(
            "device_id",
            "timestamp",
            "longitude",
            "latitude",
            F.expr("ST_Point(longitude, latitude)").alias("point_geom"),
        )
    )

[ ファイルを実行 するか 、パイプラインを実行します。

パイプライングラフに gps_bronze と raw_gps_silverが表示されるようになりました。次に、倉庫ジオフェンスを具体化されたビューとして追加します。

手順 5: 倉庫ジオフェンスのゴールドテーブルを作成する

ボリュームからジオフェンスを読み取り、GEOMETRYを使用して WKT 列をST_GeomFromWKT列に変換する具体化されたビューを作成します。

資産ブラウザーで、[プラス] アイコンをクリックします。 [追加]、[変換]の順に選択します。
[名前] を [warehouse_geofences_goldに設定し、SQL または Python を選択して、[作成] をクリックします。

次のコードを貼り付けます。 <catalog>と<schema>を既定のカタログとスキーマに置き換えます。

SQL

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

Python

from pyspark import pipelines as dp
from pyspark.sql import functions as F

path = "/Volumes/<catalog>/<schema>/raw_data/geofences"

@dp.table(name="warehouse_geofences_gold", comment="Warehouse geofence polygons as geometry")
def warehouse_geofences_gold():
    return (
        spark.read.format("json").load(path).select(
            "warehouse_name",
            F.expr("ST_GeomFromWKT(boundary_wkt)").alias("boundary_geom"),
        )
    )

[ ファイルを実行 するか 、パイプラインを実行します。

パイプラインにジオフェンステーブルが含まれるようになりました。次に、倉庫への到着を計算するために空間結合を追加します。

手順 6: 空間結合を使用して倉庫到着テーブルを作成する

ST_Contains(boundary_geom, point_geom)を使用してシルバー GPS ポイントをジオフェンスに結合する具体化されたビューを追加して、デバイスが倉庫のポリゴン内にあるかどうかを判断します。

資産ブラウザーで、をクリックし、次に[追加]、それから[変換]の順に選択します。
[名前] を [warehouse_arrivalsに設定し、SQL または Python を選択して、[作成] をクリックします。

次のコードを貼り付けます。

SQL

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)

Python

from pyspark import pipelines as dp
from pyspark.sql import functions as F

@dp.table(name="warehouse_arrivals", comment="Devices that have entered a warehouse geofence")
def warehouse_arrivals():
    g = spark.read.table("raw_gps_silver")
    w = spark.read.table("warehouse_geofences_gold")
    return (
        g.alias("g")
        .join(w.alias("w"), F.expr("ST_Contains(w.boundary_geom, g.point_geom)"))
        .select(
            F.col("g.device_id").alias("device_id"),
            F.col("g.timestamp").alias("timestamp"),
            F.col("w.warehouse_name").alias("warehouse_name"),
        )
    )

[ ファイルを実行 するか 、パイプラインを実行します。

更新が完了すると、パイプライングラフには、 gps_bronze、 raw_gps_silver、 warehouse_geofences_gold、 warehouse_arrivalsの 4 つのデータセットがすべて表示されます。

空間結合を確認する

空間結合によって行が生成されたことを確認します。ジオフェンス内にあるシルバーテーブルのポイントが warehouse_arrivalsに表示されます。ノートブックまたは SQL エディターで次のいずれかを実行します (パイプラインターゲットと同じカタログとスキーマを使用します)。

倉庫別の到着数 (SQL):

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

Warehouse_AとWarehouse_Bの 0 以外のカウントが表示されます (サンプル GPS データは両方のポリゴンに重なります)。サンプル行を検査するには:

SELECT device_id, timestamp, warehouse_name
FROM warehouse_arrivals
ORDER BY timestamp DESC
LIMIT 10;

Python (ノートブック) での同じチェック:

# Count by warehouse
display(spark.table("warehouse_arrivals").groupBy("warehouse_name").count().orderBy("warehouse_name"))

# Sample rows
display(spark.table("warehouse_arrivals").orderBy("timestamp", ascending=False).limit(10))

warehouse_arrivalsに行が表示された場合、ST_Contains(boundary_geom, point_geom)結合は正常に動作しています。

手順 7: パイプラインをスケジュールする (省略可能)

新しい GPS データがボリュームに格納された時点でパイプラインを最新の状態に保つには、スケジュールに従ってパイプラインを実行するジョブを作成します。

エディターの上部にある [ スケジュール ] ボタンを選択します。
[スケジュール] ダイアログが表示されたら、[スケジュールの追加] を選択します。
必要に応じて、ジョブに名前を付けます。
既定では、スケジュールは 1 日に 1 回実行されます。これを受け入れるか、独自に設定することができます。 [詳細設定] を選択すると、特定の時刻を設定できます。その他のオプションでは、実行通知を追加できます。
[ 作成] を選択してスケジュールを適用します。

ジョブの実行の詳細については、Lakeflow ジョブの監視と可観測性を参照してください。

その他のリソース

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-03-15

次の方法で共有

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

必要条件

手順 1: パイプラインを作成する

手順 2: GPS とジオフェンスのサンプル データを作成する

手順 3: ブロンズ ストリーミング テーブルに GPS データを取り込む

SQL

Python

手順 4: ジオメトリ ポイントを含むシルバー ストリーミング テーブルを追加する

SQL

Python

手順 5: 倉庫ジオフェンスのゴールド テーブルを作成する

SQL

Python

手順 6: 空間結合を使用して倉庫到着テーブルを作成する

SQL

Python

空間結合を確認する

手順 7: パイプラインをスケジュールする (省略可能)

その他のリソース

フィードバック

その他のリソース

手順 2: GPS とジオフェンスのサンプルデータを作成する

手順 3: ブロンズストリーミングテーブルに GPS データを取り込む

手順 4: ジオメトリポイントを含むシルバーストリーミングテーブルを追加する

手順 5: 倉庫ジオフェンスのゴールドテーブルを作成する