SparkのUDF、Spark SQL

2024/4/12に翔泳社より**Apache Spark徹底入門**を出版します！

<a href="https://www.amazon.co.jp/dp/4798182281/"><img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/87910d7c-8a63-c198-8e82-216ea7a8ee07.jpeg" width=300></a>

書籍の[サンプルノートブック](https://github.com/databricks/LearningSparkV2/blob/master/notebooks/LearningSparkv2.dbc)をウォークスルーしていきます。`Python/Chapter04/4-1 Example 4.1`となります。

翻訳ノートブックのリポジトリはこちら。

https://github.com/taka-yayoi/SparkSample/tree/main

ノートブックはこちら

https://github.com/taka-yayoi/SparkSample/blob/main/Chapter04/4-1%20Example%204.1.py

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

日付フォーマットを読めるフォーマットに変換するUDFを定義します。

:::note
**注意**
日付は文字列で年が含まれていないため、SQLの`year()`関数を使うことが困難な場合があります。
:::

```py
def to_date_format_udf(d_str):
  l = [char for char in d_str]
  return "".join(l[0:2]) + "/" +  "".join(l[2:4]) + " " + " " +"".join(l[4:6]) + ":" + "".join(l[6:])

to_date_format_udf("02190925")
```
```
'02/19  09:25'
```

ユーザー定義関数(UDF: User Defined Function)を登録します。

```py
spark.udf.register("to_date_format_udf", to_date_format_udf, StringType())
```
```
<function __main__.to_date_format_udf(d_str)>
```

USを出発するフライトデータを読み込みます。

```py
df = (spark.read.format("csv")
      .schema("date STRING, delay INT, distance INT, origin STRING, destination STRING")
      .option("header", "true")
      .option("path", "/databricks-datasets/learning-spark-v2/flights/departuredelays.csv")
      .load())
```
![Screenshot 2024-03-27 at 13.32.47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/804b0878-a97f-8aba-2b26-23c32bc59b22.png)

UDFをテストします。

```py
df.selectExpr("to_date_format_udf(date) as data_format").show(10, truncate=False)
```
```
+------------+
|data_format |
+------------+
|01/01  12:45|
|01/02  06:00|
|01/02  12:45|
|01/02  06:05|
|01/03  12:45|
|01/03  06:05|
|01/04  12:43|
|01/04  06:05|
|01/05  12:45|
|01/05  06:05|
+------------+
only showing top 10 rows
```

SQLクエリーを実行する一次ビューを作成します。

```py
df.createOrReplaceTempView("us_delay_flights_tbl")
```

クエリーに都合が良いようにテーブルをキャッシュします。

```sql
%sql
CACHE TABLE us_delay_flights_tbl
```

読みやすくなるようにすべての`date`を`date_fm`に変換します。

:::note
**注意**
オンザフライで変換するためにUDFを使用しています。
:::

```py
spark.sql("SELECT *, date, to_date_format_udf(date) AS date_fm FROM us_delay_flights_tbl").show(10, truncate=False)
```
```
+--------+-----+--------+------+-----------+--------+------------+
|date    |delay|distance|origin|destination|date    |date_fm     |
+--------+-----+--------+------+-----------+--------+------------+
|01011245|6    |602     |ABE   |ATL        |01011245|01/01  12:45|
|01020600|-8   |369     |ABE   |DTW        |01020600|01/02  06:00|
|01021245|-2   |602     |ABE   |ATL        |01021245|01/02  12:45|
|01020605|-4   |602     |ABE   |ATL        |01020605|01/02  06:05|
|01031245|-4   |602     |ABE   |ATL        |01031245|01/03  12:45|
|01030605|0    |602     |ABE   |ATL        |01030605|01/03  06:05|
|01041243|10   |602     |ABE   |ATL        |01041243|01/04  12:43|
|01040605|28   |602     |ABE   |ATL        |01040605|01/04  06:05|
|01051245|88   |602     |ABE   |ATL        |01051245|01/05  12:45|
|01050605|9    |602     |ABE   |ATL        |01050605|01/05  06:05|
+--------+-----+--------+------+-----------+--------+------------+
only showing top 10 rows
```

```py
spark.sql("SELECT COUNT(*) FROM us_delay_flights_tbl").show()
```
```
+--------+
|count(1)|
+--------+
| 1391578|
+--------+
```

# Query 1: 出発地と到着地との距離が1000より大きい全てのフライトを検索

```py
spark.sql("SELECT distance, origin, destination FROM us_delay_flights_tbl WHERE distance > 1000 ORDER BY distance DESC").show(10, truncate=False)
```
```
+--------+------+-----------+
|distance|origin|destination|
+--------+------+-----------+
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
+--------+------+-----------+
only showing top 10 rows
```

Sparkデータフレームで行う同じクエリー

```py
df.select("distance", "origin", "destination").where(col("distance") > 1000).orderBy(desc("distance")).show(10, truncate=False)
```
```
+--------+------+-----------+
|distance|origin|destination|
+--------+------+-----------+
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
|4330    |HNL   |JFK        |
+--------+------+-----------+
only showing top 10 rows
```

```py
df.select("distance", "origin", "destination").where("distance > 1000").orderBy("distance", ascending=False).show(10)
```
```
+--------+------+-----------+
|distance|origin|destination|
+--------+------+-----------+
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
+--------+------+-----------+
only showing top 10 rows
```

```py
df.select("distance", "origin", "destination").where("distance > 1000").orderBy(desc("distance")).show(10)
```
```
+--------+------+-----------+
|distance|origin|destination|
+--------+------+-----------+
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
|    4330|   HNL|        JFK|
+--------+------+-----------+
only showing top 10 rows
```

# Query 2: サンフランシスコとシカゴの間で2時間より長い遅延のあったフライトを検索

```py
spark.sql("""
SELECT date, delay, origin, destination 
FROM us_delay_flights_tbl 
WHERE delay > 120 AND ORIGIN = 'SFO' AND DESTINATION = 'ORD' 
ORDER by delay DESC
""").show(10, truncate=False)
```
```
+--------+-----+------+-----------+
|date    |delay|origin|destination|
+--------+-----+------+-----------+
|02190925|1638 |SFO   |ORD        |
|01031755|396  |SFO   |ORD        |
|01022330|326  |SFO   |ORD        |
|01051205|320  |SFO   |ORD        |
|01190925|297  |SFO   |ORD        |
|02171115|296  |SFO   |ORD        |
|01071040|279  |SFO   |ORD        |
|01051550|274  |SFO   |ORD        |
|03120730|266  |SFO   |ORD        |
|01261104|258  |SFO   |ORD        |
+--------+-----+------+-----------+
only showing top 10 rows
```

# Query 3: より複雑なクエリー

SQLにおけるより複雑なクエリーとして、目的地に関係なく空港から出発する全てのUSフライトにラベル、_high_, _medium_, _low_, *no delays*をつけましょう。

```py
spark.sql("""SELECT delay, origin, destination,
              CASE
                  WHEN delay > 360 THEN 'Very Long Delays'
                  WHEN delay > 120 AND delay < 360 THEN  'Long Delays '
                  WHEN delay > 60 AND delay < 120 THEN  'Short Delays'
                  WHEN delay > 0 and delay < 60  THEN   'Tolerable Delays'
                  WHEN delay = 0 THEN 'No Delays'
                  ELSE 'No Delays'
               END AS Flight_Delays
               FROM us_delay_flights_tbl
               ORDER BY origin, delay DESC""").show(10, truncate=False)
```
```
+-----+------+-----------+-------------+
|delay|origin|destination|Flight_Delays|
+-----+------+-----------+-------------+
|333  |ABE   |ATL        |Long Delays  |
|305  |ABE   |ATL        |Long Delays  |
|275  |ABE   |ATL        |Long Delays  |
|257  |ABE   |ATL        |Long Delays  |
|247  |ABE   |ATL        |Long Delays  |
|247  |ABE   |DTW        |Long Delays  |
|219  |ABE   |ORD        |Long Delays  |
|211  |ABE   |ATL        |Long Delays  |
|197  |ABE   |DTW        |Long Delays  |
|192  |ABE   |ORD        |Long Delays  |
+-----+------+-----------+-------------+
only showing top 10 rows
```

```py
df = spark.sql("""SELECT delay, origin, destination,
              CASE
                  WHEN delay > 360 THEN 'Very Long Delays'
                  WHEN delay > 120 AND delay < 360 THEN  'Long Delays '
                  WHEN delay > 60 AND delay < 120 THEN  'Short Delays'
                  WHEN delay > 0 and delay < 60  THEN   'Tolerable Delays'
                  WHEN delay = 0 THEN 'No Delays'
                  ELSE 'No Delays'
               END AS Flight_Delays
               FROM us_delay_flights_tbl
               ORDER BY origin, delay DESC""")
display(df)
```
![Screenshot 2024-03-27 at 13.38.19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/fab870f1-965f-41f4-cba4-d123afd13c22.png)

# その他のクエリー

```py
df1 =  spark.sql("SELECT date, delay, origin, destination FROM us_delay_flights_tbl WHERE origin = 'SFO'")
```
```py
df1.createOrReplaceGlobalTempView("us_origin_airport_SFO_tmp_view")
```
```sql
%sql
SELECT * FROM global_temp.us_origin_airport_SFO_tmp_view
```
![Screenshot 2024-03-27 at 13.39.18.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/481b4bd1-21df-03f0-afce-6cd0cc427d26.png)

```sql
%sql
DROP VIEW IF EXISTS global_temp.us_origin_airport_JFK_tmp_view
```
```py
df2 = spark.sql("SELECT date, delay, origin, destination from us_delay_flights_tbl WHERE origin = 'JFK'")
```
```py
df2.createOrReplaceTempView("us_origin_airport_JFK_tmp_view")
```
```sql
%sql
SELECT * FROM us_origin_airport_JFK_tmp_view
```
![Screenshot 2024-03-27 at 13.40.34.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/10d44309-4756-3156-ead7-14112970561b.png)

```sql
%sql
DROP VIEW IF EXISTS us_origin_airport_JFK_tmp_view
```
```py
spark.catalog.listTables(dbName="global_temp")
```
```
[Table(name='us_origin_airport_sfo_tmp_view', catalog=None, namespace=['global_temp'], description=None, tableType='TEMPORARY', isTemporary=True),
 Table(name='us_delay_flights_tbl', catalog=None, namespace=[], description=None, tableType='TEMPORARY', isTemporary=True)]
```

### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

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