センサーデータを用いたSparkストリーミングのアプリケーション

こちらのノートブックギャラリーの[Streaming applications for sensor data](https://www.databricks.com/notebooks/gallery/ContinuousStreamingApplicationWithSensorData.html)をウォークスルーした内容です。

https://databricks.com/discover/notebook-gallery

ノートブックの翻訳版はこちらとなります。

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

# フェイクセンサーデータの生成

100レコードから構成される100ファイルのデータを生成します。

```scala:Scala
%scala

// --- データを生成するパスを設定・適宜変更してください ---
val path = "/tmp/takaakiyayoidatabrickscom/Streaming/sdevices/"
dbutils.fs.mkdirs(path)

val numFiles = 100
val numDataPerFile = 100

import scala.util.Random

val deviceTypes = Seq("SensorTypeA", "SensorTypeB", "SensorTypeC", "SensorTypeD")
val startTime = System.currentTimeMillis
dbutils.fs.rm(path, true)

(1 to numFiles).par.foreach { fileId =>
  val file = s"$path/file-$fileId.json"
  val data = (1 to numDataPerFile).map { x => 
    val timestamp = new java.sql.Timestamp(startTime + (fileId * 60000) + (Random.nextInt() % 10000))
    val deviceId = Random.nextInt(100)
    val deviceType = deviceTypes(Random.nextInt(deviceTypes.size))
    val signalStrength = math.abs(Random.nextDouble % 100)
    s"""{"timestamp":"$timestamp","deviceId":$deviceId,"deviceType":"$deviceType","signalStrength":$signalStrength}"""
  }.mkString("\n")
  dbutils.fs.put(file, data)
}
dbutils.fs.head(dbutils.fs.ls(s"$path/file-1.json").head.path)
```

# ストリーミングセンサーデータの処理

エンドツーエンドの連続アプリケーションを構築する目的においては、構造化ストリーミングはパワフルな機能となります。ハイレベルにおいては、以下の機能を提供します:

1. データのすべてのレコードはプレフィックス(パーティション)に保持され、順序を守って処理・カウントされるので、**出力テーブルは常に一貫性があります**
1. **耐障害性**は、出力シンクとのやり取りを含み、構造化ストリーミングによって全体的に取り扱われます。
1. **遅延データ、順序を守らないデータ**を取り扱う能力

<img src="https://databricks.com/wp-content/uploads/2019/04/streaming_continuous-apps-1024x366.png" alt="" />

## 1-ソース

入力データソースを「入力テーブル」と考えましょう。ストリームに到着するすべてのデータアイテムは、入力テーブルに追加される新たな行なようなものとなります。

## 2-連続処理 & クエリー

次に、開発者は出力シンクに書き込まれる最終の結果テーブルを計算するために、*静的なテーブル*であるかのように、このソース、あるいは入力テーブルに対するクエリーを定義します。Sparkはこのバッチのようなクエリーをストリーミング実行プランに自動で変換します。これはインクリメンタル化とよばれるものです: Sparkは、レコードが到着するたびに結果をアップデートするために、どのような状態を維持する必要があるのかを特定します。最後に、開発者はいつ結果をアップデートするのかをコントロールするためのトリガーを指定します。トリガーが実行されるたびに、Sparkは新規データ(入力テーブルの新規行)をチェックし、結果をインクリメンタルにアップデートを行います。

<img src="https://databricks.com/wp-content/uploads/2019/04/streaming_cloudtrail-unbounded-tables.png" alt="" />

## 3-シンク

このモデルの最後の部分は、出力モードとなります。結果テーブルがアップデートされる都度、開発者はS3、HDFS、データベースのような外部システムに対する変更を書き込みを行いたいと考えます。通常、出力をインクリメンタルに書き込みたいと考えます。このためには、構造化ストリーミングでは3つの出力モードを提供します:
</p>

* __Append__: 最後のトリガー以降に結果テーブルに追加された新規行のみが、外部ストレージに書き込まれます。
* __Complete__: 集計のようにアップデートされた結果テーブル全体が外部ストレージに書き込まれます。
* __Update__: 最後のトリガー以降に結果テーブル更新された行のみが、外部ストレージ上で変更されます。

<img src="https://databricks.com/wp-content/uploads/2019/04/treaming_stream-example1-phone-updated.png"/>

## PySpark構造化ストリーミングAPIを用いた連続処理アプリケーションのサンプル

出力、チェックポイント、不正レコードのためのファイルパスをセットアップします。

```py:Python
# Cmd3の4行目のパスから /sdevices/ を除外してください
base_path = "/tmp/takaakiyayoidatabrickscom/Streaming"

output_path = f"{base_path}/out/iot-stream/"
checkpoint_path = f"{base_path}/out/iot-stream-checkpoint"
#
# チェックポイントパスの作成
#
dbutils.fs.rm(checkpoint_path,True) # チェックポイントの上書き
dbutils.fs.mkdirs(checkpoint_path)
#
#
bad_records_path = f"{base_path}/badRecordsPath/streaming-sensor/"
dbutils.fs.rm(bad_records_path, True) # ディレクトリを空に
dbutils.fs.mkdirs(bad_records_path)
```

### センサーからのデータはのどうなものでしょうか？

```py:Python
sensor_path = f"{base_path}/sdevices/"
sensor_file_name= sensor_path + "file-1.json"
dbutils.fs.head(sensor_file_name, 233)
```
![Screen Shot 2022-08-08 at 13.25.56.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4cc17d5d-20f2-cccc-a531-7b4274eff161.png)

### 入力ストリームと出力ストリームのスキーマを定義

良いベストプラクティスはパフォーマンス上の理由からSparkにスキーマを推定させるのではなく、スキーマを定義するというものです。スキーマが無い場合、Sparkはいくつかのジョブを起動します: ヘッダーを読み込むためのジョブ、データが合致するようにスキーマを検証するためにパーティションの一部を読み込むジョブです。

問題があれば即座にエラーを発生させ、欠損値やデータ型のミスマッチがあった際に、許容するか、NaNやnullで置き換えるように、スキーマを設定するオプションが存在しています。

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

# オリジナルの入力スキーマ
jsonSchema = (
  StructType()
  .add("timestamp", TimestampType()) # ソースのイベント時間
  .add("deviceId", LongType())
  .add("deviceType", StringType())
  .add("signalStrength", DoubleType())
)
# いくつかのETL(変換およびカラムの追加)を行うのでカラムを追加してスキーマを変更します。
# この変換データは、処理やレポート生成に使用できるようにSQLテーブルを作成元としてのParquetファイルに格納されます。
parquetSchema = (
  StructType()
  .add("timestamp", TimestampType()) # ソースのイベント時間
  .add("deviceId", LongType())
  .add("deviceType", StringType())
  .add("signalStrength", DoubleType())
  .add("INPUT_FILE_NAME", StringType()) # このデータアイテムを読み込んだファイル名
  .add("PROCESSED_TIME", TimestampType())) # 処理中のエグゼキューターの時間
```

### オブジェクトストアソースからのストリームの読み込み

このケースでは、ファイルから一度に読み込むことでKafkaライブストリームをシミュレートします。しかし、これをApache Kafkaのトピックにすることもできます。

:::note info
**注意**
チュートリアルのために意図的に処理を遅くしています。
:::

```py:Python
inputDF = ( spark 
          .readStream 
          .schema(jsonSchema) 
          .option("maxFilesPerTrigger", 1)  # チュートリアルのために処理を遅くしています
          .option("badRecordsPath", bad_records_path) # いかなる不正レコードはこちらに格納されます
          .json(sensor_path) # ソース
          .withColumn("INPUT_FILE_NAME", input_file_name()) # ファイルパスを保持
          .withColumn("PROCESSED_TIME", current_timestamp()) # 処理時刻のタイムスタンプを追加
          .withWatermark("PROCESSED_TIME", "1 minute") # オプション: 順序が遅れたデータに対するウィンドウ
         )
```

### Parquetファイルシンクへのストリームの書き込み

```py:Python
query = (inputDF
         .writeStream
         .format("parquet") # 後段処理あるいは必要に応じてバッチクエリーのために保存を行うシンク
         .option("path", output_path)
         .option("checkpointLocation", checkpoint_path) # 障害復旧のためのチェックポイントの追加
         .outputMode("append")
         .queryName("devices") # オプションとして、クエリーを実行する際に指定するクエリー名を指定
         .trigger(processingTime='5 seconds')
         .start() 
        )
```
ストリーミングクエリーが起動し、専用のダッシュボードで処理の進捗を確認することができます。
![Screen Shot 2022-08-08 at 13.28.55.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4c4b68c6-ce86-70ae-7ac2-8c79cae11a5b.png)

#### クイックにSQLクエリーを実行できるように入力ストリームから一時テーブルを作成

```py:Python
inputDF.createOrReplaceTempView("parquet_sensors")
```

### 入力ストリームから作成された一時テーブルに対してクエリーを実行

```sql:SQL
%sql select * from parquet_sensors where deviceType = 'SensorTypeD' or deviceType = 'SensorTypeA'
```

![Screen Shot 2022-08-08 at 13.30.05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/bf47906c-0aff-0586-5351-406e3240466e.png)
これらのクエリーはストリーミングクエリーとなりますので、逐次内容は更新されていきます。

### 入力ストリームから保存されたParquetファイルに対して追加の処理を行い、クエリーを実行

```py:Python
spark.conf.set("spark.sql.shuffle.partitions", "1") # 優れたクエリー性能のためにシャッフルサイズを小さく維持
devices = (spark.readStream
           .schema(parquetSchema)
           .format("parquet")
           .option("maxFilesPerTrigger", 1) # デモのために遅くしています
           .load(output_path)
           .withWatermark("PROCESSED_TIME", "1 minute") # 順序を守らないデータに対するウィンドウ
          )

 # より複雑な集計クエリーを行うために一時テーブルを作成
devices.createOrReplaceTempView("sensors")
```

以降でもストリーミングクエリーも実行していきます。

#### 何ファイルが処理されましたか？

```py:Python
display(
  devices.
  select("INPUT_FILE_NAME", "PROCESSED_TIME")
  .groupBy("INPUT_FILE_NAME", "PROCESSED_TIME")
  .count()
  .orderBy("PROCESSED_TIME", ascending=False)
)
```
![Screen Shot 2022-08-08 at 13.31.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/26682da5-76ca-88d3-523d-7134c0e86da7.png)

#### どれだけのデータが通過しましたか？

```py:Python
%sql select count(*) from sensors
```
![Screen Shot 2022-08-08 at 13.32.14.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/31b3d200-3da7-f257-77e1-6972058fa3c2.png)

#### それぞれのセンサータイプにおける強度の最小値、最大値、平均値はどのようになっていますか？

Spark SQLの関数であるmin()、max()、avg()を使います。

:::note info
**注意**
Pythonノートブックで`%sql`マジックコマンドを使うことでSQLを使うことができます。
:::

```sql:SQL
%sql 

select count(*), deviceType, min(signalStrength), max(signalStrength), avg(signalStrength) 
  from sensors 
    group by deviceType 
    order by deviceType asc
```
![Screen Shot 2022-08-08 at 13.33.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/19dea393-0d54-241a-9ff0-6deae55cd503.png)

#### デバイスと5秒ウィンドウごとのシグナルのカウントを集計するストリームを作成してみましょう

:::note info
**注意**
これはタンブリングウィンドウであり、スライディングウィンドウではありません。サイズは5秒です。
:::

例えば、サイズ5の日、時間、分のタンブリングウィンドウは以下のようになります。

*[(00:00 - 00:05), (00:05: 00:10), (00:10: 00:15)]*

イベントは、これらのタンブリングウィンドウのいずれかに属することになります。

```py:Python
(devices
 .groupBy(
   window("timestamp", "5 seconds"),
   "deviceId"
 )
 .count()
 .createOrReplaceTempView("sensor_counts")) # データフレームを用いて一時ビューを作成
```

#### これらの5秒ウィンドウにおいて、どのデバイスがシグナルのロスを経験しているのか？

```py:Python
%sql select * from sensor_counts where count < 5 order by window.start desc
```
![Screen Shot 2022-08-08 at 13.34.39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/a535594c-5410-f058-fe10-e58eb32f8b2d.png)

### シグナルを送信していないダウンしている可能性があるセンサーのアラートの送信

一時テーブル`sensor_counts`からデータフレームを作成しましょう。

```py:Python
lost_sensor_signals = (spark.table("sensor_counts")
         .filter(col("count") < 5)
         .select("window.start", "window.end", "deviceId", "count")
         )

# データフレームの表示
display(lost_sensor_signals)
```
![Screen Shot 2022-08-08 at 13.35.14.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/421ebefc-1303-670d-a9b2-01dab108b89b.png)

##### ワーカーのログに書き込むために、`foreach`メカニズムを活用

これは、モニタリングの目的で使用することができます。別のジョブがアラートのためにログをスキャンし、Kafkaのトピックに公開するジョブや、Gangliaにポストすることができます。利用できるKafkaサーバーや、REST API経由で利用できるGangliaサービスがあるのであればトライしてみるといい練習になります。

```py:Python
def processRow(row):
  # 今時点ではログファイルに書き込みを行いますが、このロジックは容易にKafkaのトピックや、GangliaやPagerDutyのようなモニタリング、ページングサービスにアラートを発行するように拡張することができます
  print("ALERT from Sensors: Between {} and {}, device {} reported only {} times".format(row.start, row.end, row.deviceId, row[3]))
  
(lost_sensor_signals
 .writeStream
 .outputMode("complete") # モニタリングのためにKafkaの"alerts"トピックにすることもできます
 .foreach(processRow)
 .start()
)
```

_こちらが書き出された結果のサンプルとなります。_

<img src="https://databricks.com/wp-content/uploads/2018/12/Screen-Shot-2018-12-18-at-5.53.39-PM.png" alt=""/>

## データのクリーンアップ

### 稼働中のすべてのクエリーを取得

```py:Python
sqm = spark.streams
[q.name for q in sqm.active]
```
![Screen Shot 2022-08-08 at 13.36.28.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d2d03c4b-c0c5-491d-227b-8b6349ff3926.png)

### ストリームの停止

```py:Python
[q.stop() for q in sqm.active]
```


### Databricks 無料トライアル

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