構造化ストリーミングとDatabricks SQLを組み合わせてリアルタイムダッシュボードを作成してみる

こちらの記事で紹介されていたデータが興味深かったので、説明されている内容に沿ってストリーミング処理を組み、Databricks SQLでダッシュボードに表示するところまでやってみました。

https://blog.coeo.com/databricks-structured-streaming-part2

https://blog.coeo.com/databricks-structured-streaming-part3

# ソースデータ

上の記事で言及しているデータはこちら。シアトルの911への通報をリアルタイムでJSONファイルとして公開しています。まさにオープンデータ。5分周期で更新されます。

https://data.seattle.gov/Public-Safety/Seattle-Real-Time-Fire-911-Calls/kzjm-xkqj

![Screen Shot 2022-06-05 at 11.02.11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/98bf4eb2-9347-ebb1-ea78-0395c8d3efb0.png)

1. 右上の**API**をクリックし**JSON**をクリックするとURLが表示されるのでこれをコピーしておきます。
![Screen Shot 2022-06-05 at 11.02.50.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e4019c93-64cc-a8b0-3145-95c71fcb5c5a.png)
1. ブラウザで表示すると内容を確認することができます。
![Screen Shot 2022-06-05 at 11.03.39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/960fc063-6e04-c1ab-878f-86672900df82.png)
1. これを定期的にJSONファイルとして保存して、ストリーミングのデータソースとします。

# JSONファイルに保存する

Databricks[ノートブック](https://qiita.com/taka_yayoi/items/24a897cf40bba6d9e305)を作成して、以下のロジックを記述していきます。

```py:Python
# APIのパス
url = "<上のステップで取得したURL>"
# ファイル保存場所
path = "/tmp/takaaki.yayoi@databricks.com/sfd/json"
file = "sfd-"
extension = ".json"
```

```py:Python
def save_json_to_file(api, path, file, extension):
  import urllib.request as request
  import json
  import time
  
  # ファイル保存場所の作成
  dbutils.fs.mkdirs(path)
  
  # APIからデータを取得
  with request.urlopen(api) as response:
    if response.getcode() == 200:
      source = response.read()
      data = json.loads(source)
    else:
      print("APIからデータを取得しようとした際にエラーが発生しました。")
      
  time_str = time.strftime("%Y%m%d-%H%M%S")
  filename = f"/dbfs{path}/{file}{time_str}{extension}"
  
  if filename:
    # JSONデータの書き込み
    with open(filename, "w") as f:
      json.dump(data, f)

  return filename
```

以下を実行することで、JSONファイルが保存されることを確認しておきます。このノートブックは後で[ジョブ](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a)として実行することになります。

```py:Python
jsonfile = save_json_to_file(url, path, file, extension)
print(jsonfile)
```

# ノートブックをジョブとして実行

上で作成したノートブック用いて[ジョブ](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a)を作成し、ジョブを定期的に実行することで、ストリーミングの入力とします。

:::note info
**注意**
- ここでは5分周期で実行していますが、アクセス先の負荷を考慮した設定にするようにしてください。
- それぞれのJSONファイルではレコードの重複がありますが、ここではそのままとします。
:::

1. ノートブックの右上にある**スケジュール**をクリックします。
1. **間隔5分、from: 00**とし、5分間隔でノートブックが実行されるようにします。
1. アラートの設定などは必要に応じて追加してください。
1. **作成**をクリックします。
![Screen Shot 2022-06-05 at 11.10.48.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d125aebb-4112-9b98-2d82-84353b53be69.png)

これでジョブが設定されました。

# ストリーミング処理の実装

これで定期的にオブジェクトストレージにJSONファイルが到着するようになりました。

![Screen Shot 2022-06-05 at 11.14.37.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/41bbcc01-c454-1de8-432a-c8e08cd63597.png)

これが自動でストリーミング処理に入力されるように[Auto Loader](https://qiita.com/taka_yayoi/items/df143647dcf5942b51c6)を使用します。これによって、未処理のファイルのみがストリーミングで処理されるようになります。

そして、ここではDeltaアーキテクチャ(メダリオンアーキテクチャ)に従って、ストリーミング処理のパイプラインを構築していきます。Deltaアーキテクチャの詳細は、[デルタ vs ラムダ: なぜデータパイプラインのシンプルさは複雑さに勝るのか](https://qiita.com/taka_yayoi/items/3695983039517bc9e909)をご覧ください。

## ブロンズテーブル

生データの保管場所になるのがブロンズテーブルです。

次のステップでJSONのスキーマを指定する必要があるので、上のステップで取得したJSONファイルを退避しておき、そのファイルからスキーマを取得するようにします。

```py:Python
# JSONスキーマのパーシング
jsonFile = spark.read.json("/tmp/takaaki.yayoi@databricks.com/sfd/sfd-20220604-073941.json")
jsonFile.schema
```

以下の例では、上のジョブによってJSONファイルは[DBFS](https://qiita.com/taka_yayoi/items/897264c486e179d72247)の`/tmp/takaaki.yayoi@databricks.com/sfd/json/`に到着するものとします。

```py:Python
homeDir = "/tmp/takaaki.yayoi@databricks.com/sfd/json/"

bronzeCkpt = homeDir + "bronze.checkpoint"
silverCkpt = homeDir + "silver.checkpoint"
goldCkpt = homeDir + "gold.checkpoint"

# 保存先をリセット
print(homeDir)
dbutils.fs.rm(bronzeCkpt , True)
dbutils.fs.rm(silverCkpt , True)
dbutils.fs.rm(goldCkpt , True)
```

以下を実行することで保存されたJSONが順次処理されるストリームが起動します。formatを`cloudFiles`と指定することでAuto Loaderを利用することができます。結果をDeltaテーブル`20210712_demo_takaakiyayoi.sfd_bronze_table`に保持しています。

```py:Python
from pyspark.sql.types import *

# upload_pathに投入されるファイルを読み込むストリームをセットアップ
(spark.readStream.format('cloudFiles')
  .option('cloudFiles.format', 'json')
  .schema(jsonFile.schema)
  .load(homeDir)
  .writeStream.format('delta')
  .outputMode("append")
  .option('checkpointLocation', bronzeCkpt)
  .queryName('Bronze Stream')
  .toTable("20210712_demo_takaakiyayoi.sfd_bronze_table")
)
```

ストリーム処理はバッチ処理と異なり、明示的に停止するまで処理を継続します。ストリーム処理のダッシュボードを表示することもできます。
![Screen Shot 2022-06-05 at 11.22.02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f8919616-0ec5-8edb-7d59-dce08c467d59.png)

テーブルをクエリーすることで中身を確認することができます。
![Screen Shot 2022-06-05 at 11.22.50.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/6bfdec34-3819-ad17-0dba-9d5132aa82d3.png)

## シルバーテーブル

ブロンズテーブルはデータ型が文字列型だったり、不要な列が含まれているので、以降の処理がしやすいようにクレンジング、型変換を行います。この結果がシルバーテーブルに流れ込む形になります。

```py:Python
from pyspark.sql.functions import unix_timestamp, col

(spark
  .readStream
  .format("delta")
  .table("20210712_demo_takaakiyayoi.sfd_bronze_table")
  .withColumn("datetime", unix_timestamp(col("datetime"), "yyyy-MM-dd'T'HH:mm:ss.SSS").cast("timestamp"))
  .withColumn("latitude", col("latitude").cast("float"))
  .withColumn("longitude", col("longitude").cast("float"))
  .select(col("incident_number"), col("address"), col("datetime"), col("longitude"), col("latitude"), col("type"))
  .writeStream.format('delta')
  .outputMode("append")
  .option('checkpointLocation', silverCkpt)
  .queryName('Silver Stream')
  .toTable("20210712_demo_takaakiyayoi.sfd_silver_table")
)
```

ブロンズテーブルよりシンプルな形になりました。
![Screen Shot 2022-06-05 at 11.24.49.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/eec42d37-6f4d-54a4-fbd3-3ce60e3fb858.png)

## ゴールドテーブル

BIや機械学習の特徴量としてデータを用いるには、何かしらの集計処理が必要となるケースがほとんどです。そういった集計結果を保持するのがゴールドテーブルです。

以下の例では5分の時間間隔ごとに通報タイプ別の件数をカウントしています。

```py:Python
from pyspark.sql.functions import count, window

( # 上記セルと同様の処理ロジックにWatermarkingを付け加える
  spark
   .readStream
   .format("delta")
   .table("20210712_demo_takaakiyayoi.sfd_silver_table")
   .withWatermark('datetime','5 minutes')  # withWatermark()でWindowの有効期間を設ける
                                            # 即ち一定時間以上遅れて到着したイベントは無視される
   .groupBy(col("type"), window('datetime', '5 minutes')) # 時間ウィンドウとタイプで集計
   .count()
   .selectExpr('type', 
              'window.start as Start', 
              'window.end as End',
              'count as Total')
   .writeStream                             # Windowの集計結果をGoldテーブルに保存
   .format('delta')
   .option('checkpointLocation', goldCkpt)
   .outputMode('append')
   .queryName('Gold Stream')
   .toTable("20210712_demo_takaakiyayoi.sfd_gold_table")
)
```

上で使用している`withWatermark`のウォーターマークに関しては、以下の記事が勉強になりました。

https://developers.microad.co.jp/entry/2019/07/12/063000

さらにここからストリームを伸ばして、一時テーブルとすることでリアルタイムで更新される一覧やグラフをノートブック上で表示することができます。

```py:Python
# GoldテーブルからREADストリームを作成
gold_stream = spark.readStream.format("delta").table("20210712_demo_takaakiyayoi.sfd_gold_table")

# READストリームからTEMPテーブルを作成
gold_stream.createOrReplaceTempView("gold_stream_table")
```

![Screen Shot 2022-06-05 at 11.27.14.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/627f47dd-b31e-91a5-4bbe-5b969513ff83.png)
![Screen Shot 2022-06-05 at 11.28.52.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/bbd64535-486d-4f70-6bbb-ecfc13eaf8e3.png)

# Databricks SQLで可視化を行う

これまでのステップで(ニア)リアルタイムに処理を行うデータパイプラインを構築できたわけですが、準備したデータには緯度経度情報も含まれているので、これを活用して可視化を行います。

1. Databricksワークスペースの左にあるサイドメニュー上部のペルソナスイッチャーからSQLを選択します。
![Screen Shot 2022-06-05 at 11.30.29.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/47d54285-279c-4203-0014-8142cb109bc0.png)
1. Databricks SQLに移動します。
1. サイドメニューから**SQLエディタ**を選択します。
![Screen Shot 2022-06-05 at 11.31.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/90c1ff7a-cc7a-2e6c-ab6d-b8b0aaac7ed8.png)
1. こちらの画面で上のステップで作成したDeltaテーブルに対するクエリーを定義します。
![Screen Shot 2022-06-05 at 11.32.04.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/fcc90850-58c2-cbcc-b152-c7f51edabff6.png)
1. 以下のようなクエリーを記述してシルバーテーブルから最新のレコードを取得します。

    ```sql:SQL
    SELECT DISTINCT * FROM `20210712_demo_takaakiyayoi`.sfd_silver_table ORDER BY datetime DESC LIMIT 500
    ```

1. 右上の**実行**をクリックしてクエリーの結果を確認します。
1. **+ ビジュアライゼーションを追加**をクリックします。
1. Visualization typeから**Map (Markers)** を選択します。
1. Latitude columnには`latitude`、Longitude columnには`longitude`を指定します。
1. 地図上にマーカーがプレビュー表示されるので、見やすいように表示範囲を調整します。
![Screen Shot 2022-06-05 at 11.40.07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e1323935-4b38-0257-cb6e-de3c767c6ecf.png)
1. **Format**タブをクリックし、**Show popup**にチェックを入れ、Popup templateに`{{ type }} @ {{ address }}`と入力します。これで、地図上のマーカーをクリックすると情報が表示されるようになります。**保存**をクリックします。
![Screen Shot 2022-06-05 at 11.40.40.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/10451a26-77c6-1340-b1fd-06864963e424.png)
1. 他にも表示したい情報があれば、クエリーとビジュアライゼーションを定義していきます。
1. これらビジュアライゼーションの部品が出来上がったらサイドメニューから**ダッシュボード**を選択し、**ダッシュボードを作成**をクリックします。
1. ダッシュボードに名前をつけます。
1. ダッシュボードのキャンバスが表示されるので、ここに上で作成したビジュアライゼーションを追加していきます。
1. 右上の**Add > Visualization**を選択します。
![Screen Shot 2022-06-05 at 11.44.32.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/3cd98c62-f485-845f-3984-9b5f44c12d67.png)
1. 上で作成したビジュアライゼーションを選択し、**ダッシュボードに追加**をクリックします。
![Screen Shot 2022-06-05 at 11.45.12.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/063a00fc-0ea4-f3ed-31bb-a0e7f5b7489c.png)
1. 追加したいビジュアライゼーションに対して上の操作を繰り返します。
1. **編集完了**をクリックします。
1. これで以下のようなダッシュボードが出来上がります。
![Screen Shot 2022-06-05 at 11.33.51.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4d4e5305-28f3-5685-f5ce-b7fae94d6737.png)
1. さらにこのダッシュボードを自動で更新されるようにするために、スケジュール処理を設定します。右上の**スケジュール**をクリックします。
![Screen Shot 2022-06-05 at 11.47.10.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f07744b8-417d-82e1-3ad6-62e045c85ae9.png)
1. `5分間隔`などを指定して**保存**をクリックします。
1. これで自動で更新されるダッシュボードの完成です！

:::note warn
**重要！**
処理の実行が不要になったら明示的にストリーミング処理のノートブックやジョブを停止してください。ノートブックで使用したクラスターも停止してください。
:::

このように、Databrikcsのジョブ、Sparkの構造化ストリーミング、Databricks SQLを組み合わせることで、最新の情報を提供することができるデータパイプライン、ダッシュボードを容易に構築することができます。是非試してみてください。

**参考資料**

https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a

https://qiita.com/taka_yayoi/items/ae258e2e160c41239435

https://qiita.com/taka_yayoi/items/df143647dcf5942b51c6

https://qiita.com/taka_yayoi/items/eaf79254afd4eb620a6e

### Databricks 無料トライアル

[Databricks 無料トライアル](https://databricks.com/jp/try-databricks)