[2024年版] Apache Sparkデータフレームによるデータの読み込みと変換のチュートリアル

2年前にこちらのマニュアルを翻訳しました。

https://qiita.com/taka_yayoi/items/6cf24cd4ad921aa61356

今ではマニュアルも日本語化されているので、今回はサンプルノートブックをウォークスルーします。

# ステップ1: 変数を定義し、 CSVファイルを読み込む

Unity Catalogの[ボリューム](https://docs.databricks.com/ja/connect/unity-catalog/volumes.html)にCSVファイルを格納します。

```py
catalog = "takaakiyayoi_catalog"
schema = "default"
volume = "data"
download_url = "https://health.data.ny.gov/api/views/jxy9-yhdk/rows.csv"
file_name = "rows.csv"
table_name = "health_data"
path_volume = "/Volumes/" + catalog + "/" + schema + "/" + volume
path_tables = catalog + "." + schema
print(path_tables) # 完全パスを表示
print(path_volume) # 完全パスを表示
```
```
takaakiyayoi_catalog.default
/Volumes/takaakiyayoi_catalog/default/data
```
```py
dbutils.fs.cp(f"{download_url}", f"{path_volume}" + "/" + f"{file_name}")
```
CSVファイルが格納されました。
![Screenshot 2024-04-28 at 18.16.28.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/b0edde14-8447-d37b-7359-af21a6502d25.png)

# ステップ2: DataFrameを作成する

```py
data = [[2021, "test", "Albany", "M", 42]]
columns = ["Year", "First_Name", "County", "Sex", "Count"]

df1 = spark.createDataFrame(data, schema="Year int, First_Name STRING, County STRING, Sex STRING, Count int")
display(df1) # display()メソッドはDatabricksノートブック固有のものであり、よりリッチなビジュアライゼーションを提供します。
# df1.show() show()メソッドはApache Spark DataFrame APIに含まれており、基本的なビジュアライゼーションを提供します。
```
![Screenshot 2024-04-28 at 18.17.50.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/676487c5-dec1-9fdd-d004-7d70f98761d2.png)

# ステップ3: CSVファイルからDataFrameにデータをロードする

```py
# ボリュームに格納されているCSVファイルからデータフレームを作成
df_csv = spark.read.csv(f"{path_volume}/{file_name}",
                        header=True,
                        inferSchema=True,
                        sep=",")
display(df_csv)
```
![Screenshot 2024-04-28 at 18.18.36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/83127474-eff1-df73-091f-badf14a58733.png)

# ステップ4: DataFrameを表示して操作する

## DataFrameのスキーマを表示する

```py
# スキーマの表示
df_csv.printSchema()
df1.printSchema()
```
```
root
 |-- Year: integer (nullable = true)
 |-- First Name: string (nullable = true)
 |-- County: string (nullable = true)
 |-- Sex: string (nullable = true)
 |-- Count: integer (nullable = true)

root
 |-- Year: integer (nullable = true)
 |-- First_Name: string (nullable = true)
 |-- County: string (nullable = true)
 |-- Sex: string (nullable = true)
 |-- Count: integer (nullable = true)
```

## DataFrameの列名を変更する

```py
df_csv = df_csv.withColumnRenamed("First Name", "First_Name")
df_csv.printSchema
```
```
<bound method DataFrame.printSchema of DataFrame[Year: int, First_Name: string, County: string, Sex: string, Count: int]>
```

## DataFramesを結合する

これは縦方向の`union`になります。

```py
df = df1.union(df_csv)
display(df)
```

先頭に`test`の行が追加されていることがわかります。
![Screenshot 2024-04-28 at 18.20.03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/83497a8b-3547-c6d8-3f3d-559987b2b7c8.png)

## DataFrame内の行をフィルタリングする

`filter`を用いて**Count**列の値が50よりも大きいものにフィルタリングします。

```py
display(df.filter(df["Count"] > 50))
```
![Screenshot 2024-04-28 at 18.21.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/a0144e74-a5ab-9e08-5c6e-263f49be98b8.png)

こちらも同じ挙動をします。

```py
display(df.where(df["Count"] > 50))
```

## DataFrame から列を選択し、頻度順に並べ替える

`desc`で**Count**の降順、`orderBy`でソートを支持します。`select`はSQLと同じように列を選択しています。考え方自体、SQLと似ています。

```py
from pyspark.sql.functions import desc
display(df.select("First_Name", "Count").orderBy(desc("Count")))
```
![Screenshot 2024-04-28 at 18.23.47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/98b08a57-d7a0-61bc-12ef-d798a43a5416.png)

## DataFrameのサブセットを作成する

`filter`で以下の条件を**AND**で指定しています。

- **Year**が`2009`
- **Count**が`100`より大きい
- **Sex**が`F`
- 列は**First_Name**、**County**、**Count**を選択
- **Count**の降順でソート

```py
subsetDF = (
    df.filter((df["Year"] == 2009) & (df["Count"] > 100) & (df["Sex"] == "F"))
    .select("First_Name", "County", "Count")
    .orderBy(desc("Count"))
)
display(subsetDF)
```
![Screenshot 2024-04-28 at 18.25.47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/5b5886b6-00ae-d785-de20-66f72954983d.png)

# ステップ5: DataFrameを保存する

```py
df.write.saveAsTable(f"{path_tables}" + "." + f"{table_name}")

# 既存のテーブルを上書きするには以下のコードを使います:
# df.write.mode("overwrite").saveAsTable(f"{path_tables}" + "." + f"{table_name}")
```

上を実行するとデータフレームがテーブルとして保存されます。
![Screenshot 2024-04-28 at 18.26.47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/b2192b5f-222b-dc57-f304-37fc5e21e01b.png)

## DataFrameをJSONファイルに保存する

パスを指定して他のフォーマットで保存することもできます。

```py
df.write.format("json").save("/tmp/json_data")

# 既存のファイルを上書きするには以下のコードを使います:
# df.write.format("json").mode("overwrite").save("/tmp/json_data")
```

## JSONファイルからDataFrameを読み込む

```py
display(spark.read.format("json").json("/tmp/json_data"))
```
![Screenshot 2024-04-28 at 18.28.03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e4cc0f98-925f-419b-cd20-4362a0cde3d3.png)

# 追加タスク: PySparkでSQLクエリを実行する

:::note info
**注意**
元のノートブックでは、Scala、RでのSQL実行もカバーされていますが、こちらでは割愛します。
:::

## SQLクエリとして列を指定する

selectExprやexprを用いることで、PySparkの中でSQLエクスプレッションを活用することができます。

```py
display(df.selectExpr("Count", "upper(County) as big_name"))
```
![Screenshot 2024-04-28 at 18.29.18.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/8a65458e-fe97-9a5f-656a-ceac938ef2e5.png)

```py
from pyspark.sql.functions import expr
display(df.select("Count", expr("lower(County) as little_name")))
```
![Screenshot 2024-04-28 at 18.29.49.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/2ec68de8-038b-4526-9e3a-b801450dbf07.png)

詳細はこちらをご覧ください。

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

## spark.sql() を使用して任意の SQL クエリを 実行する

`spark.sql()`の引数にはSQL文を文字列として指定できます。Pythonの中でクイックにSQLを使いたい際には便利です。

```py
display(spark.sql(f"SELECT * FROM {path_tables}" + "." + f"{table_name}"))
```
![Screenshot 2024-04-28 at 18.33.00.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e5b101f6-54bd-cc01-2488-8d895ea0a9aa.png)

上の例ではf-stringを使ってパラメータを埋め込んでいますが、Spark 3.4からはネイティブのパラメータ化がサポートされています。詳細はこちらをご覧ください。

https://qiita.com/taka_yayoi/items/72c2a93a5c863710daa1

### はじめてのDatabricks

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

### Databricks無料トライアル

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