Pythonデータフレームを用いた構造化ストリーミング

[Demo notebooks \| Databricks on AWS](https://docs.databricks.com/spark/latest/structured-streaming/demo-notebooks.html) [2021/9/24時点]のStructured Streaming demo Python notebookを翻訳・説明する記事です。

Apache Sparkには、高レベルのストリーミング処理APIである[構造化ストリーミング](http://spark.apache.org/docs/latest/structured-streaming-programming-guide.html)が含まれています。ここで説明するノートブックでは、Databricksで構造化ストリーミングアプリケーションを構築するためにどのようにデータフレームAPIを活用するのかをクイックに見てみます。ここでは、ストリームにおけるタイムスタンプを伴うアクション(Open、Closeなど)に対するウィンドウ内のカウントや連続的に更新されるカウントのようなリアルタイムのメトリクスを計算します。

# サンプルノートブックの説明

## サンプルデータ

ここでアプリケーションを構築するために用いるサンプルのアクションデータは、`/databricks-datasets/structured-streaming/events/`にファイルとして格納されています。ディレクトリの中身を見てみましょう。

```sh
%fs ls /databricks-datasets/structured-streaming/events/
```

![Screen Shot 2021-12-17 at 9.42.04.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4f9f923b-328e-51bd-63c0-1cdcf82f0f80.png)

ディレクトリには約50個のJSONファイルが格納されています。個々のJSONファイルの中身を見てみましょう。

```sh
%fs head /databricks-datasets/structured-streaming/events/file-0.json
```

![Screen Shot 2021-12-17 at 10.48.30.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/88c6b97c-8f29-2247-cc21-431018994352.png)

ファイル内のそれぞれの行には、2つのフィールド`time`と`action`が含まれています。インタラクティブにこれらのファイルを解析してみましょう。

## バッチ/インタラクティブ処理

データを処理する最初のステップは、通常はデータに対するインタラクティブなクエリーの実行です。ファイルに対する静的なデータフレームを定義し、テーブル名をつけましょう。

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

inputPath = "/databricks-datasets/structured-streaming/events/"

# すでにデータのフォーマットを知っているので、処理を高速化するためにスキーマを定義しましょう(Sparkがスキーマを推定する必要がなくなります)
jsonSchema = StructType([ StructField("time", TimestampType(), True), StructField("action", StringType(), True) ])

# JSONファイルのデータを表現する静的なデータフレーム
staticInputDF = (
  spark
    .read
    .schema(jsonSchema)
    .json(inputPath)
)

display(staticInputDF)
```

![Screen Shot 2021-12-17 at 9.44.36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/afa73a3e-b04d-1d48-a04c-92011e250f76.png)

これで、1時間のタイムウィンドウにおける`open`と`close`アクションの数を計算することができます。このためには、`action`カラムと`time`カラムに対する1時間のウィンドウでグルーピングを行います。

```py:Python
from pyspark.sql.functions import *      # window()関数を使用するために必要

staticCountsDF = (
  staticInputDF
    .groupBy(
       staticInputDF.action, 
       window(staticInputDF.time, "1 hour"))    
    .count()
)
staticCountsDF.cache()

# データフレームをテーブル'static_counts'として登録します 
staticCountsDF.createOrReplaceTempView("static_counts")
```

これで、テーブルに対してSQLを用いて直接クエリーを実行できます。例えば、以下のように全ての時間におけるアクションの総数を計算します。

```sql:SQL
%sql select action, sum(count) as total_count from static_counts group by action
```

![Screen Shot 2021-12-17 at 9.45.23.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d2bbc9f6-e7d5-05a1-77d1-9b51a8538fec.png)

ウィンドウ内のカウントはどうでしょうか？

```sql:SQL
%sql select action, date_format(window.end, "MMM-dd HH:mm") as time, count from static_counts order by time, action
```

![Screen Shot 2021-12-17 at 9.46.04.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/29cb4327-ee12-b961-bb23-680ccef2ae6a.png)

グラフの最後の2つに注意してください。対応するOpenアクションの後にCloseアクションが生じるように生成されており、最初に多くの"Open"が存在し、最後に多くの"Close"が存在しています。

## ストリーム処理

ここまではデータをインタラクティブに分析しましたが、データの到着に伴って連続的に更新するストリーミングクエリーに切り替えましょう。今回は一連の静的なファイルがあるのみなので、一度に一つのファイルを時系列に読み込むことでストリームをシミュレートします。ここで記述すべきクエリーは、上述のインタラクティブなクエリーと非常に似たものとなります。

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

# 上で定義したstaticInputDFの定義と似ていますが、`read`ではなく`readStream`を使用します
streamingInputDF = (
  spark
    .readStream                       
    .schema(jsonSchema)               # JSONデータのスキーマを設定
    .option("maxFilesPerTrigger", 1)  # 一度に一つのファイルを取り込むことで一連のファイルをストリームとして取り扱います
    .json(inputPath)
)

# staticInputDFと同じクエリー
streamingCountsDF = (                 
  streamingInputDF
    .groupBy(
      streamingInputDF.action, 
      window(streamingInputDF.time, "1 hour"))
    .count()
)

# このデータフレームはストリーミングデータフレームでしょうか？
streamingCountsDF.isStreaming
```

![Screen Shot 2021-12-17 at 9.47.11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/efc86d0a-8227-a247-9617-a0b67eb55fa6.png)

上でわかるように、`streamingCountsDF`はストリーミングデータフレームです(`streamingCountsDF.isStreaming`が`true`でした)。sinkを定義し、起動することで、ストリーミング処理をスタートすることができます。

今回のケースでは、カウントをインタラクティブに取得(上と同じクエリー)したいので、インメモリのテーブルに1時間ごとのカウントをセットします。

```py:Python
spark.conf.set("spark.sql.shuffle.partitions", "2")  # シャッフルのサイズを小さくします

query = (
  streamingCountsDF
    .writeStream
    .format("memory")        # memory = インメモリテーブルに格納 
    .queryName("counts")     # counts = インメモリテーブルの名称
    .outputMode("complete")  # complete = 全てのカウントをテーブルに保持
    .start()
)
```

![Screen Shot 2021-12-17 at 9.47.55.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d8fedf08-ec7b-2d28-2d47-68b1815d7781.png)

`query`はバックグラウンドで実行されるストリーミングクエリーのハンドルとなります。このクエリーは継続的にファイルを取得し、ウィンドウ内のカウントを更新します。

上のセルのクエリーのステータスに注意してください。プログレスバーはクエリーがアクティブであることを示しています。さらに、上の`> counts`を展開すると、すでに処理されたファイルの数を確認することができます。

いくつかのファイルが処理されるまで待ち、インメモリの`counts`テーブルに対してインタラクティブにクエリーを実行してみましょう。

```py:Python
from time import sleep
sleep(5)  # 計算がスタートするまで少し待ちます
```

```sql:SQL
%sql select action, date_format(window.end, "MMM-dd HH:mm") as time, count from counts order by time, action
```

![Screen Shot 2021-12-17 at 9.48.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/05575cd5-be49-07cc-e586-b01c2b8f763e.png)

クエリーを実行するごとに結果が更新されていることがわかります。

![Screen Shot 2021-12-17 at 9.48.56.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/72be0007-2961-5c27-74ab-c901663bc42a.png)

![Screen Shot 2021-12-17 at 9.49.10.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c66b9a5a-0965-4f27-2a86-21b636e27364.png)

上のクエリーを繰り返し実行し続けることで、"Close"が常に対応する"Open"の後に出現するデータストリームで期待されたように、常に"Close"の数より"Open"の数が多いことを確認することができます。これは、構造化ストリーミングが**prefix integrity**を保証していることを示しています。prefix integrityの詳細を知りたい方は以下のブログ記事を参照ください。

[Spark Structured Streaming \- The Databricks Blog](https://databricks.com/blog/2016/07/28/structured-streaming-in-apache-spark.html)

ここでは少数のファイルしか取り扱っていないので、全てのファイルを処理してしまうとカウントは更新されなくなります。再度ストリーミングクエリーを操作したい場合には、クエリーを際実行してください。

最後に、クエリーのセルの**Cancel**リンクをクリックするか、`query.stop()`を実行することで、バックグラウンドで実行しているクエリーを停止します。いずれの方法でも、クエリーが停止されると対応するセルのステータスは自動的に`TERMINATED`となります。

# サンプルノートブック

https://github.com/taka-yayoi/public_repo/tree/main/python_structured_streaming

### Databricks 無料トライアル

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