Databricks Zerobus와 Lakebase를 활용한 글로벌 실시간 IoT 데이터 스트리밍 데모

CREATE TABLE catalog_name.schema_name.table_name (
  ID STRING NOT NULL, -- Unique record ID (UUID)
  CLIENT_ID STRING, -- 12-char readable client identifier
  TIMESTAMP TIMESTAMP NOT NULL, -- When sensor reading was taken
  acceleration_x DOUBLE, -- Linear acceleration (m/s2)
  ...
  latitude DOUBLE, -- GPS latitude in degrees
  longitude DOUBLE -- GPS longitude in degrees
) USING DELTA;

from lakebase_foreachwriter import LakebaseForeachWriter

writer = LakebaseForeachWriter(
    username="your-username",
    password="your-password",
    table="your_target_table",
    df=your_dataframe,
    host="your-lakebase-dns.databricks.com"
)

query = (
    streaming_df.writeStream
    .foreach(writer)
    .start()
)