Sparkを用いたサンフランシスコ消防署の通報データの分析

[ノートブックギャラリー](https://databricks.com/discover/notebook-gallery)で公開されている[Analysis of the San Fransisco fire calls with Spark](https://databricks.com/notebooks/gallery/SanFranciscoFireCallsAnalysis.html)をウォークスルーする内容となっています。

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

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

このノートブックは書籍**Learning Spark 2nd Edition**の第3章のエンドツーエンドのサンプルを示すものであり、[San Francisco Fire Department Calls ](https://data.sfgov.org/Public-Safety/Fire-Department-Calls-for-Service/nuek-vuh3)データセットに対する一般的な分析パターンとオペレーションのために、どのようにデータフレームとSpark SQLを用いるのかを説明します。また、分析のためにどのようにETLやデータの検証、データのクエリーを行うのかをデモします。さらに、インメモリのSparkデータフレームをどのようにParquetファイルとして保存するのか、SparkがサポートするParquetデータソースとしてどのように読み込むのかを説明します。

# データの確認

```
%fs ls /databricks-datasets/learning-spark-v2/sf-fire/sf-fire-calls.csv
```
![Screen Shot 2022-07-19 at 10.16.59.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/21f46360-6d7f-1a87-1043-d79126a75b44.png)


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

sf_fire_file = "/databricks-datasets/learning-spark-v2/sf-fire/sf-fire-calls.csv"
```

```
%fs head /databricks-datasets/learning-spark-v2/sf-fire/sf-fire-calls.csv
```
![Screen Shot 2022-07-19 at 10.17.39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/9f5f05f3-c4cb-10b1-3125-2985de91d97c.png)

# スキーマの定義

ファイルには400万レコードが含まれているのでスキーマを定義します。大規模ファイルにおけるスキーマ推定には大きなコストが必要となります。

```py:Python
fire_schema = StructType([StructField('CallNumber', IntegerType(), True),
                     StructField('UnitID', StringType(), True),
                     StructField('IncidentNumber', IntegerType(), True),
                     StructField('CallType', StringType(), True),                  
                     StructField('CallDate', StringType(), True),      
                     StructField('WatchDate', StringType(), True),
                     StructField('CallFinalDisposition', StringType(), True),
                     StructField('AvailableDtTm', StringType(), True),
                     StructField('Address', StringType(), True),       
                     StructField('City', StringType(), True),       
                     StructField('Zipcode', IntegerType(), True),       
                     StructField('Battalion', StringType(), True),                 
                     StructField('StationArea', StringType(), True),       
                     StructField('Box', StringType(), True),       
                     StructField('OriginalPriority', StringType(), True),       
                     StructField('Priority', StringType(), True),       
                     StructField('FinalPriority', IntegerType(), True),       
                     StructField('ALSUnit', BooleanType(), True),       
                     StructField('CallTypeGroup', StringType(), True),
                     StructField('NumAlarms', IntegerType(), True),
                     StructField('UnitType', StringType(), True),
                     StructField('UnitSequenceInCallDispatch', IntegerType(), True),
                     StructField('FirePreventionDistrict', StringType(), True),
                     StructField('SupervisorDistrict', StringType(), True),
                     StructField('Neighborhood', StringType(), True),
                     StructField('Location', StringType(), True),
                     StructField('RowID', StringType(), True),
                     StructField('Delay', FloatType(), True)])
```

# データの読み込み

```py:Python
fire_df = spark.read.csv(sf_fire_file, header=True, schema=fire_schema)
```

データに対して複数のオペレーションを実行するので、データフレームをキャッシュします。

```py:Python
fire_df.cache()
```

```py:Python
fire_df.count()
```
![Screen Shot 2022-07-19 at 10.21.11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4fc63f91-d2cc-49a2-7b81-820a28642b79.png)

データフレームの最初の50レコードを表示します。以下ではグラフが設定されているので、呼び出しごとの遅延時間が棒グラフで表示されます。

```py:Python
display(fire_df.limit(50))
```
![Screen Shot 2022-07-19 at 10.22.04.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/279bfa00-c6ff-a9d1-64e1-5049fd5aa253.png)

# フィルタリング

Call Typeの`"Medical Incident"`を除外します。

データフレームに対する`filter()`と`where()`メソッドは同じものであることに注意してください。引数のタイプに関しては[ドキュメント](https://spark.apache.org/docs/3.1.1/api/python/reference/api/pyspark.sql.DataFrame.filter.html)をチェックしてください。

```py:Python
few_fire_df = (fire_df.select("IncidentNumber", "AvailableDtTm", "CallType") 
              .where(col("CallType") != "Medical Incident"))

few_fire_df.show(5, truncate=False)
```
![Screen Shot 2022-07-19 at 10.23.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/365d4366-4419-8a8e-acc8-6482c77123ad.png)

## Q-1) 消防署に対して何種類の呼び出しの電話がありましたか？

念のために、カラムに含まれる"null"文字列はカウントしないようにしましょう。

```py:Python
fire_df.select("CallType").where(col("CallType").isNotNull()).distinct().count()
```
![Screen Shot 2022-07-19 at 10.24.16.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e6645d4f-3abe-8ebd-a3c0-976bc46c3d26.png)

## Q-2) 消防署に対してどのようなタイプの呼び出しがありましたか？

これらがSF Fire Departmentに対する全ての呼び出しタイプとなります。

```py:Python
fire_df.select("CallType").where(col("CallType").isNotNull()).distinct().show(10, False)
```
![Screen Shot 2022-07-19 at 10.25.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/526d93fd-f755-ed45-7246-911ec0e72330.png)

`display()`メソッドを使用することもできます。

```py:Python
fire_df.select("CallType").where(col("CallType").isNotNull()).distinct().display()
```
![Screen Shot 2022-07-19 at 10.26.37.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/1b6587fd-af74-b538-5060-7849eb79fe0e.png)

## Q-3) 遅延時間が5分以上の全ての応答を見つけ出してください

1. カラム名`Delay`を`ReponseDelayedinMins`に変更します。
1. 新たなデータフレームを返却します。
1. 出火地点への反応時間が5分以上遅延した全ての電話を探し出します。

```py:Python
new_fire_df = fire_df.withColumnRenamed("Delay", "ResponseDelayedinMins")
new_fire_df.select("ResponseDelayedinMins").where(col("ResponseDelayedinMins") > 5).show(5, False)
```
![Screen Shot 2022-07-19 at 10.27.48.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/6303850d-c767-2b29-12d0-281f9bae8b39.png)

# ETL(Extract/Transform/Load)

いくつかのETL処理をやってみましょう:

1. 後で時間に基づくクエリーを行えるように、文字列の日付をSparkのTimestampデータ型に変換します。
1. 変換された結果が返却されます。
1. 新たなデータフレームをキャッシュします。

```py:Python
fire_ts_df = (new_fire_df
              .withColumn("IncidentDate", to_timestamp(col("CallDate"), "MM/dd/yyyy")).drop("CallDate") 
              .withColumn("OnWatchDate",   to_timestamp(col("WatchDate"), "MM/dd/yyyy")).drop("WatchDate")
              .withColumn("AvailableDtTS", to_timestamp(col("AvailableDtTm"), "MM/dd/yyyy hh:mm:ss a")).drop("AvailableDtTm"))        
```

```py:Python
fire_ts_df.cache()
fire_ts_df.columns
```
![Screen Shot 2022-07-19 at 10.29.43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c9937241-bfa1-2bf3-ec6e-4820f05db2af.png)

変換されたカラムがSpark Timestamp型であることを確認します。

```py:Python
fire_ts_df.select("IncidentDate", "OnWatchDate", "AvailableDtTS").show(5, False)
```
![Screen Shot 2022-07-19 at 10.30.20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/a491d280-0fa9-9f31-c948-eec1ee4aa555.png)

## Q-4) 最も多い呼び出しタイプは何ですか？

降順で並び替えをしましょう。

```py:Python
(fire_ts_df
 .select("CallType").where(col("CallType").isNotNull())
 .groupBy("CallType")
 .count()
 .orderBy("count", ascending=False)
 .show(n=10, truncate=False))
```
![Screen Shot 2022-07-19 at 10.31.30.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/40e52cc4-b8e3-67bb-eae9-af8d76efeed4.png)

## Q-4a) 呼び出しの大部分を占めているzipコードはなんですか？

サンフランシスコ消防署への通報においてどのzipコードが多いのか、どの場所でどのタイプが多いのかを調査してみましょう。

1. `CallType`でフィルタリングします。
2. `CallType`と`Zip code`でグルーピングします。
3. カウントを行い、降順で表示します。

最も共通する電話は`Medical Incident`に関連するものであり、多いzipコードは`94102`と`94103`です。

```py:Python
display(fire_ts_df
 .select("CallType", "ZipCode")
 .where(col("CallType").isNotNull())
 .groupBy("CallType", "Zipcode")
 .count()
 .orderBy("count", ascending=False))
```
![Screen Shot 2022-07-19 at 10.32.17.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d379fc34-52be-9099-69b2-849d48d6400d.png)

## Q-4b) Zipコード94102と94103はサンフランシスコのどの地域ですか？

これらの2つのZipコードに関連づけられる地域を見つけ出しましょう。おそらく、通報率が高い地域は隣接しているケースがあるのでしょう。

```py:Python
display(fire_ts_df.select("Neighborhood", "Zipcode").where((col("Zipcode") == 94102) | (col("Zipcode") == 94103)).distinct())
```
![Screen Shot 2022-07-19 at 10.32.56.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c4309706-8fa4-56e6-24dc-fb0b51672cb3.png)

## Q-5) 全ての呼び出しの合計、呼び出しに対する反応時間の平均値、最小値、最大値は何ですか？

いくつかのカラムに対して合計、平均値、最小値、最大値を計算するためにビルトインのSpark SQL関数を使いましょう。

* 通報の合計数
* 通報地点に消防隊員が到着するまでの反応時間の平均値、最小値、最大値

```py:Python
fire_ts_df.select(sum("NumAlarms"), avg("ResponseDelayedinMins"), min("ResponseDelayedinMins"), max("ResponseDelayedinMins")).show()
```
![Screen Shot 2022-07-19 at 10.33.34.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/31385e3f-2494-6255-6f56-5a98a0bcfb64.png)

## Q-6a) CSVファイルには何年分のデータが含まれていますか？

Timestamp型の`IncidentDate`から年を取り出すためにSpark SQL関数`year()`を使うことができます。

全体的には2000-2018のデータが含まれていることがわかります。

```py:Python
fire_ts_df.select(year('IncidentDate')).distinct().orderBy(year('IncidentDate')).show()
```
![Screen Shot 2022-07-19 at 10.34.09.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/816550e4-72b2-b7b0-9a9c-1297c9a60392.png)

## Q-6b) 2018年のどの週が最も通報が多かったですか？

**注意**: Week 1は新年の週で、Week 25は7/4の週となります。花火の季節を考えると、この週の通報が多いのは納得できます。

```py:Python
fire_ts_df.filter(year('IncidentDate') == 2018).groupBy(weekofyear('IncidentDate')).count().orderBy('count', ascending=False).show()
```
![Screen Shot 2022-07-19 at 10.34.46.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/46f58bdd-eeb5-5b40-5d95-be0155fb1322.png)

## Q-7) 2018年で最も反応が悪かったサンフランシスコの地域はどこですか？

Presidio Heightsに住んでいると消防隊員は3分以内に到着し、Mission Bayに住んでいるのであれば6分以上かかるようです。

```py:Python
fire_ts_df.select("Neighborhood", "ResponseDelayedinMins").filter(year("IncidentDate") == 2018).show(10, False)
```
![Screen Shot 2022-07-19 at 10.35.20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e050a507-7e9b-0fd6-6bf5-3febf7020a7e.png)

## Q-8a) どのようにしてデータフレームをParquetファイルに保存し、読み戻すことができますか？

```py:Python
fire_ts_df.write.format("parquet").mode("overwrite").save(work_path)
```

## Q-8b) データを保存し、読み戻せるようにするためにどのようにSQLテーブルを使用できますか？

```py:Python
fire_ts_df.write.format("parquet").mode("overwrite").saveAsTable("FireServiceCalls")
```

```sql:SQL
%sql
CACHE TABLE FireServiceCalls
```

```sql:SQL
%sql
SELECT * FROM FireServiceCalls LIMIT 10
```
![Screen Shot 2022-07-19 at 10.37.18.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/7a2d4ed9-142a-bd38-c615-5d60272f9886.png)

## Q-8c) どのようにParquetファイルを読み込むことができますか？

Parquetメタデータの一部にスキーマが格納されているので、スキーマを指定する必要がないことに注意してください。

```py:Python
file_parquet_df = spark.read.format("parquet").load(work_path)
display(file_parquet_df.limit(10))
```
![Screen Shot 2022-07-19 at 10.38.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/3bdcd962-675a-e1b4-1121-5eb1d45ae7c7.png)


### Databricks 無料トライアル

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