PySparkことはじめ

PythonからApache Sparkを操作する際に使用するAPIであるPySparkの基本的な使い方を説明します。

こちらの記事で説明しているノートブックはこちらとなります。

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

# Apache Sparkとは

Apache Sparkとは、ビッグデータと機械学習のための非常に高速な分散処理フレームワークです。SparkはDatabricksの創業者たちによって開発されました。Databricksにおける分散処理はSparkによって行われます。

**参考資料**

- [About Spark – Databricks](https://databricks.com/jp/spark/about)
- [Apache Spark as a Service – Databricks](https://databricks.com/jp/glossary/apache-spark-as-a-service)

# PySparkとは

PySparkとは、Sparkを実行するためのPython APIです。Apache SparkとPythonのコラボレーションをサポートするためにリリースされました。開発者はPySparkを用いることで、Pythonからデータフレームを操作する形でSparkを活用することができます。

**参考資料**

- [PySpark – Databricks](https://databricks.com/jp/glossary/pyspark)
- [Databricks Apache Sparkクイックスタート \- Qiita](https://qiita.com/taka_yayoi/items/bf5fb09a0108aa14770b)
- [Databricks Apache Sparkデータフレームチュートリアル \- Qiita](https://qiita.com/taka_yayoi/items/2a7e9bb792eba316de4b)
- [PySpark Documentation — PySpark 3\.2\.1 documentation](https://spark.apache.org/docs/latest/api/python/)
- [Beginner’s Guide on Databricks: Spark Using Python & PySpark \| by Christopher Lewis \| Analytics Vidhya \| Medium](https://medium.com/analytics-vidhya/beginners-guide-on-databricks-spark-using-python-pyspark-de74d92e4885)
- [【PySpark入門】第１弾 PySparkとは？ \- サーバーワークスエンジニアブログ](https://blog.serverworks.co.jp/introducing-pyspark-1)

# PySparkの基本的な使い方

## データのロード

PySparkでデータをロードする際には`spark.read`を使用します。`format`の引数に読み込むデータのフォーマットを指定します。`json`、`parquet`、`delta`などが指定できます。読み込んだデータはSparkデータフレームとなります。

その前に、読み込むデータを以下のコマンドで確認します。こちらのデータは、アメリカの都市ごとの人口と売上のデータとなります。

```py:Python
%fs
ls dbfs:/databricks-datasets/samples/population-vs-price/
```

![Screen Shot 2022-06-10 at 14.20.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f97e050e-8f65-a92c-dcd1-aff804ecebc3.png)

```py:Python
# データフレームにサンプルデータをロードします
df = spark.read.format("csv").option("header", True).load("/databricks-datasets/samples/population-vs-price/data_geo.csv")
```

データフレームを表示します。

```py:Python
df.show(20)
```

![Screen Shot 2022-06-10 at 14.21.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f96790bf-51a1-e0c5-7f4e-3dd6eb1c1a6b.png)

`display`関数を用いることで、よりグラフィカルに結果を表示、操作することができます。簡単に[グラフを表示](https://qiita.com/taka_yayoi/items/36a307e79e9433121c38)することもできます。

```py:Python
# Databricksでデータフレームを表示するにはdisplay関数を使うと便利です
display(df)
```

![Screen Shot 2022-06-10 at 14.22.28.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cc483521-ed95-53bc-d1e4-cfc736888b71.png)

### カラムの確認

```py:Python
df.columns
```

![Screen Shot 2022-06-10 at 14.24.42.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/06c423ef-5db6-5bc1-082d-4ff2f2d273fc.png)

### スキーマの確認

```py:Python
# データフレームのスキーマを表示
df.printSchema()
```

![Screen Shot 2022-06-10 at 14.25.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/403490f5-8f6b-f730-db30-8f44bcd09481.png)

## カラム名の変更

`withColumnRenamed`を使ってカラム名を変更します。

```py:Python
df2 = df.withColumnRenamed('2014 rank', '2014_rank')\
.withColumnRenamed('State Code', 'state_code')\
.withColumnRenamed('2014 Population estimate', '2014_pop_estimate')\
.withColumnRenamed('2015 median sales price', '2015_median_sales_price')

display(df2)
```

![Screen Shot 2022-06-10 at 14.26.19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/343f7feb-4d2a-b386-4b6b-cd3c566ccde4.png)

## データ型の変換

- 既に存在しているデータフレームのカラムを指定するには、`col`関数の引数にカラム名を指定します。
- `cast`にデータ型を指定してキャストします。
- `withColumn`を用いて、キャストした後の値を持つカラムで更新します。

**参考資料**

- [Data Types \- Spark 3\.2\.1 Documentation](https://spark.apache.org/docs/latest/sql-ref-datatypes.html)

```py:Python
df3 = df2.withColumn("2014_rank", col("2014_rank").cast(IntegerType()))\
 .withColumn("2014_pop_estimate", col("2014_pop_estimate").cast(IntegerType()))\
 .withColumn("2015_median_sales_price", col("2015_median_sales_price").cast(FloatType()))

display(df3)
```

![Screen Shot 2022-06-10 at 14.27.20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/9cfda426-52f1-d96c-8ea1-4eb2f1e4e023.png)


## データの操作

### フィルタリング、ソート

以下の例では、`df3`で`2015_median_sales_price`が100より大きいレコードを`2015_median_sales_price`の降順でソートし、カラム`2014_rank`, `City`, `2015_median_sales_price`を取得しています。

```py:Python
display(df3.select("2014_rank", "City", "2015_median_sales_price")\
        .where("2015_median_sales_price > 100")\
        .orderBy(col("2015_median_sales_price").desc()))
```

![Screen Shot 2022-06-10 at 14.28.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/35b50747-2998-8f2f-6861-990c6792d48d.png)

### 集計

以下の処理を行なっています。

1. `state_code`でレコードをグルーピング
1. グループごとの`2015_median_sales_price`の平均値を計算
1. 平均値降順でレコードを取得

```py:Python
display(df3.groupBy("state_code")\
        .agg(avg("2015_median_sales_price").alias("2015_median_sales_price_avg"))
        .orderBy(col("2015_median_sales_price_avg").desc()))
```

![Screen Shot 2022-06-10 at 14.29.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c3b2e34b-10fb-3451-5b5e-474b99850688.png)

## データの書き込み

ファイルシステムにデータフレームを永続化するには、`spark.write`を使用します。

```py:Python
# Databricksユーザー名の取得
username_raw = dbutils.notebook.entry_point.getDbutils().notebook().getContext().tags().apply('user')
print(username)

# データの書き込み先、あとでクリーンアップします
tmp_file_path = f"/tmp/{username}/datacsv"
print(tmp_file_path)

# formatを指定しない場合、デフォルトのDelta形式で保存されます
df3.write.mode("overwrite").save(tmp_file_path)
```

```py:Python
# データフレームに保存したデータをロードします
df_new = spark.read.format("delta").load(tmp_file_path)
display(df_new)
```

![Screen Shot 2022-06-10 at 14.30.26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/a5127be1-782d-1770-2211-84eb7db1dc70.png)

### クリーンアップ

```py:Python
# 上のセルで保存したファイルを削除しておきます
dbutils.fs.rm(tmp_file_path, True)
```

## pandasとのやりとり

matplotlibで可視化したいなどpandas前提の処理を行う場合には、Sparkデータフレームをpandasデータフレームに変換します。

```py:Python
df4 = df3.groupBy("state_code")\
        .agg(avg("2015_median_sales_price").alias("2015_median_sales_price_avg"))\
        .orderBy(col("2015_median_sales_price_avg").desc()).limit(10)
```

```py:Python
import matplotlib.pyplot as plt

# pandasデータフレームに変換します
pdf = df4.toPandas()

# 棒グラフを描画します
plt.bar(pdf['state_code'], pdf['2015_median_sales_price_avg'], align="center")           
plt.show()
```

![Screen Shot 2022-06-10 at 14.31.42.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/44382a90-aad2-b1a5-8af7-6fac3cf9099c.png)

pandasデータフレームをSparkデータフレームに変換することもできます。

```py:Python
# Sparkデータフレームへの変換
sdf = spark.createDataFrame(pdf)
display(sdf)
```

# その他のAPI

PySpark以外のAPIと組み合わせてロジックを組み立てることも可能です。

## Spark SQL

データフレームをテーブルあるいは一時ビューに登録することで、SQLを使用してデータを操作することができるようになります。

テーブルやビューは永続化されますが、一時ビューは永続化されず、クラスターが稼働している間のみ一時ビューを作成したセッションでのみ利用することができます。

```py:Python
# データフレームを一時ビューに登録します
df3.createOrReplaceTempView("pop_price")
```

```py:Python
# '2014_rank' カラムに基づいて上位10位の市を参照します
top_10_results = spark.sql("""SELECT * FROM pop_price 
                              WHERE 2014_rank <= 10
                              SORT BY 2014_rank ASC""")
display(top_10_results)
```

![Screen Shot 2022-06-10 at 14.33.17.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d9bebbe2-15c6-b38c-9744-c98690b739bf.png)

```sql:SQL
%sql
SELECT
  *
FROM
  pop_price
WHERE
  2014_rank <= 10 SORT BY 2014_rank ASC
```

![Screen Shot 2022-06-10 at 14.33.49.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/12890a80-6ac5-e372-f259-b9b400374839.png)

## pandas API on Spark

pandas APIに慣れ親しんでいる方は、pandas API on Spark(旧Koalas)を活用することもできます。

**参考資料**

- [Apache Spark™ 3\.2におけるPandas APIのサポート \- Qiita](https://qiita.com/taka_yayoi/items/63a21a0e5113e33ad6a3)

```py:Python
import pyspark.pandas as ps

psdf = sdf.to_pandas_on_spark()  # pandas-on-Sparkデータフレーム

# pandasのお作法でカラムにアクセスします
psdf['state_code']
```

![Screen Shot 2022-06-10 at 14.35.02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/7a466621-022b-8ee9-c68d-e31f39bfe21f.png)

### Databricks 無料トライアル

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