ローカルから疑似工場センサーデータを流してDatabricks Zerobus Ingestを試してみた

https://<workspace-id>.zerobus.<region>.cloud.databricks.com

https://dbc-xxxxxxxx-yyyy.cloud.databricks.com/?o=1234567890123456

DATABRICKS_WORKSPACE_URL=https://dbc-xxxxxxxx-yyyy.cloud.databricks.com
WORKSPACE_ID=1234567890123456

https://1234567890123456.zerobus.us-west-2.cloud.databricks.com

{"error_code":"INVALID_PARAMETER_VALUE","message":"Unsupported table kind. Tables created in default storage are not supported. Error Code: 4024, Error State: 0."}

CREATE CATALOG IF NOT EXISTS sagara_zerobus
MANAGED LOCATION 's3://databricks-bucket-sagara/sagara-free-edition';

CREATE SCHEMA IF NOT EXISTS sagara_zerobus.iot;

CREATE TABLE IF NOT EXISTS sagara_zerobus.iot.factory_telemetry_zerobus (
  event_id STRING,
  line_id STRING,
  device_id STRING,
  seq BIGINT,
  observed_at_ms BIGINT,
  sent_at_ms BIGINT,
  temp_c DOUBLE,
  humidity_pct DOUBLE,
  motor_rpm INT,
  power_kw DOUBLE,
  status STRING,
  source STRING,
  producer_host STRING
);

DESCRIBE EXTENDED sagara_zerobus.iot.factory_telemetry_zerobus;

Settings
  → Identity and access
    → Service principals
      → Manage
      → Add service principal

GRANT USE CATALOG ON CATALOG sagara_zerobus
TO `<SERVICE_PRINCIPAL_APPLICATION_ID>`;

GRANT USE SCHEMA ON SCHEMA sagara_zerobus.iot
TO `<SERVICE_PRINCIPAL_APPLICATION_ID>`;

GRANT MODIFY, SELECT
ON TABLE sagara_zerobus.iot.factory_telemetry_zerobus
TO `<SERVICE_PRINCIPAL_APPLICATION_ID>`;

sudo apt update
sudo apt install -y curl jq

uv init zerobus-ingest-demo --python 3.12
cd zerobus-ingest-demo

zerobus-ingest-demo/
  ├─ .python-version
  ├─ README.md
  ├─ main.py
  └─ pyproject.toml

uv add databricks-zerobus-ingest-sdk python-dotenv

uv run python --version
uv run python -c "import zerobus; print('zerobus sdk import ok')"

# Databricks workspace
DATABRICKS_WORKSPACE_URL=https://dbc-xxxxxxxx-yyyy.cloud.databricks.com
WORKSPACE_ID=1234567890123456
DATABRICKS_REGION=us-west-2

# Zerobus endpoint
ZEROBUS_ENDPOINT=https://1234567890123456.zerobus.us-west-2.cloud.databricks.com

# Service principal
DATABRICKS_CLIENT_ID=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
DATABRICKS_CLIENT_SECRET=xxxxxxxxxxxxxxxxxxxxxxxx

# Unity Catalog target
CATALOG=sagara_zerobus
SCHEMA=iot
TABLE=factory_telemetry_zerobus
TABLE_NAME=sagara_zerobus.iot.factory_telemetry_zerobus

chmod 600 .env

# secrets
.env

#!/usr/bin/env bash
set -euo pipefail

source .env

authorization_details=$(cat <<JSON
[{
  "type": "unity_catalog_privileges",
  "privileges": ["USE CATALOG"],
  "object_type": "CATALOG",
  "object_full_path": "$CATALOG"
},
{
  "type": "unity_catalog_privileges",
  "privileges": ["USE SCHEMA"],
  "object_type": "SCHEMA",
  "object_full_path": "$CATALOG.$SCHEMA"
},
{
  "type": "unity_catalog_privileges",
  "privileges": ["SELECT", "MODIFY"],
  "object_type": "TABLE",
  "object_full_path": "$CATALOG.$SCHEMA.$TABLE"
}]
JSON
)

echo "Getting OAuth token..."

OAUTH_RESPONSE=$(curl -s -X POST \
  -u "$DATABRICKS_CLIENT_ID:$DATABRICKS_CLIENT_SECRET" \
  -d "grant_type=client_credentials" \
  -d "scope=all-apis" \
  -d "resource=api://databricks/workspaces/$WORKSPACE_ID/zerobusDirectWriteApi" \
  --data-urlencode "authorization_details=$authorization_details" \
  "$DATABRICKS_WORKSPACE_URL/oidc/v1/token")

OAUTH_TOKEN=$(echo "$OAUTH_RESPONSE" | jq -r '.access_token')

if [ "$OAUTH_TOKEN" = "null" ] || [ -z "$OAUTH_TOKEN" ]; then
  echo "Failed to get OAuth token."
  echo "$OAUTH_RESPONSE" | jq .
  exit 1
fi

now_ms=$(date +%s%3N)

event_id=$(uv run python - <<'PY'
import uuid
print(uuid.uuid4())
PY
)

payload=$(cat <<JSON
[{
  "event_id": "$event_id",
  "line_id": "line-a",
  "device_id": "wsl-sensor-001",
  "seq": 1,
  "observed_at_ms": $now_ms,
  "sent_at_ms": $now_ms,
  "temp_c": 31.5,
  "humidity_pct": 62.0,
  "motor_rpm": 1450,
  "power_kw": 3.2,
  "status": "normal",
  "source": "wsl2-rest-uv",
  "producer_host": "$(hostname)"
}]
JSON
)

echo "Posting record to Zerobus Ingest..."
echo "Target: $CATALOG.$SCHEMA.$TABLE"

curl -i -X POST \
  "$ZEROBUS_ENDPOINT/zerobus/v1/tables/$CATALOG.$SCHEMA.$TABLE/insert" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OAUTH_TOKEN" \
  -d "$payload"

echo

chmod +x rest_insert_once.sh

./rest_insert_once.sh

SELECT *
FROM sagara_zerobus.iot.factory_telemetry_zerobus
ORDER BY sent_at_ms DESC
LIMIT 10;

uv run python producer_grpc.py \
  --devices 5 \
  --seconds 60 \
  --interval 1.0 \
  --wait-each

uv run python producer_grpc.py \
  --devices 5 \
  --seconds 120 \
  --interval 1.0 \
  --wait-each

SELECT
  COUNT(*) AS records,
  MAX(sent_at_ms) AS max_sent_at_ms
FROM sagara_zerobus.iot.factory_telemetry_zerobus;

CREATE OR REPLACE VIEW sagara_zerobus.iot.factory_telemetry_zerobus_dedup AS
WITH ranked AS (
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY event_id
      ORDER BY sent_at_ms DESC
    ) AS rn
  FROM sagara_zerobus.iot.factory_telemetry_zerobus
)
SELECT
  event_id,
  line_id,
  device_id,
  seq,
  observed_at_ms,
  sent_at_ms,
  temp_c,
  humidity_pct,
  motor_rpm,
  power_kw,
  status,
  source,
  producer_host
FROM ranked
WHERE rn = 1;

SELECT
  line_id,
  device_id,
  COUNT(*) AS high_temp_events,
  MAX(temp_c) AS max_temp_c,
  AVG(temp_c) AS avg_temp_c
FROM sagara_zerobus.iot.factory_telemetry_zerobus_dedup
WHERE status IN ('warning', 'critical')
GROUP BY line_id, device_id
ORDER BY high_temp_events DESC, max_temp_c DESC;

SELECT
  line_id,
  device_id,
  COUNT(*) AS records,
  AVG(power_kw) AS avg_power_kw,
  MAX(power_kw) AS max_power_kw,
  AVG(temp_c) AS avg_temp_c
FROM sagara_zerobus.iot.factory_telemetry_zerobus_dedup
GROUP BY line_id, device_id
ORDER BY avg_power_kw DESC;

SELECT
  device_id,
  status,
  COUNT(*) AS records,
  MAX(temp_c) AS max_temp_c,
  MAX(power_kw) AS max_power_kw
FROM sagara_zerobus.iot.factory_telemetry_zerobus_dedup
WHERE sent_at_ms >= (unix_millis(current_timestamp()) - 5 * 60 * 1000)
GROUP BY device_id, status
ORDER BY device_id, status;

SELECT
  workspace_id,
  stream_id,
  table_name,
  opened_time,
  closed_time,
  errors
FROM system.lakeflow.zerobus_stream
WHERE table_name LIKE '%factory_telemetry_zerobus%'
ORDER BY opened_time DESC
LIMIT 20;

SELECT
  workspace_id,
  stream_id,
  table_name,
  commit_time,
  committed_records,
  committed_bytes,
  errors
FROM system.lakeflow.zerobus_ingest
WHERE table_name LIKE '%factory_telemetry_zerobus%'
ORDER BY commit_time DESC
LIMIT 20;