データサイエンス100本ノックをPySparkでやってみた

# はじめに
本記事では、「データサイエンス100本」をPySparkを使用して実施する方法について解説します。データサイエンス100本は、実践的なデータ分析のスキルを身に付けるための有名な課題集です。PySparkは、Apache SparkのPython向けAPIであり、大規模なデータセットを分散処理するための強力なツールです。

本記事の目的は、PySparkを使用したデータサイエンス100本ノックの実施方法を詳しく解説し、読者がPySparkを活用してデータ分析のスキルを向上させる手助けをすることです。具体的には、PySparkの基礎知識から始め、データセットの読み込み、クリーニング、変換、集計のタスクをPySparkを使って実際に解決していきます。


## PySparkの特徴
PySparkの特徴は以下の通りです。
- 分散処理: PySparkは、Apache Sparkの分散処理エンジンを活用しており、大規模なデータセットの処理に優れています。複数のマシンやクラスタ上で処理を分散させることで、高速なデータ処理を実現します。
- データ操作の柔軟性: PySparkは、高レベルのデータ操作インターフェースであるDataFrameを提供しています。DataFrameは、異なるデータ形式やソースからのデータの読み込みや変換を容易に行うことができます。
- 拡張性: PySparkはPythonをベースとしており、Pythonの豊富なライブラリやエコシステムとの連携が可能です。また、ユーザー定義関数（UDF）の作成やカスタム処理の追加などもサポートしています。
- パフォーマンスの最適化: PySparkは、データのパーティショニングやクエリの最適化など、パフォーマンスの最適化に関する豊富な機能を提供しています。これにより、大規模データセットでも効率的に処理を行うことができます。

本記事では、PySparkの基礎知識から始めながら、データサイエンティスト100本ノックの各課題をPySparkを使って解決していきます。PySparkの特徴を活かしながら、データ分析の手法を紹介します。パフォーマンスの最適化については本記事では対象外としています。
次章では、データサイエンティスト100本ノックについて詳しく説明します。



# データサイエンス100本ノックについて
データサイエンティスト協会が提供している構造化データ（テーブルデータ）についての読み込みやクリーニング、変換、集計の問題が100問収録されている課題集です。

https://github.com/The-Japan-DataScientist-Society/100knocks-preprocess

PySparkは、データサイエンス100本ノックにおいて有用なツールの1つです。以下の理由から、PySparkを使用することで効率的なデータ処理と分析が可能となります。


- スケーラビリティ: PySparkは、分散処理エンジンであるApache Sparkを基にしており、複数のマシンやクラスタで処理を分散することができます。これにより、大容量のデータを高速に処理することが可能です。

- 柔軟なデータ操作: PySparkは、高レベルのデータ操作インターフェースであるDataFrameを提供しています。DataFrameを使用することで、異なるデータ形式やソースからのデータの読み込みや変換を容易に行うことができます。また、SQLライクなクエリ言語もサポートしており、データの操作や集計を簡潔に記述することができます。

- 拡張性と豊富な機能: PySparkはPythonをベースとしており、Pythonの豊富なライブラリやエコシステムとの連携が可能です。また、ユーザー定義関数（UDF）の作成やカスタム処理の追加などもサポートされています。これにより、データの前処理や特徴量エンジニアリングにおいて、柔軟かつ高度な処理を行うことができます。

データサイエンス100本ノックで扱うデータは小規模ですが、解説記事でよく扱われているPandasと同様の処理をPySparkで実現することができることを目標としています。

次章では、PySparkの基本的な使い方とデータの読み込みについて詳しく説明します。


# PySparkのセットアップと基礎知識
PySparkのセットアップと基礎知識について説明します。
なお、本記事では環境設定を省略するため、Google Colabを利用しています。

まずは、PySparkをインストールします。
```
!pip install pyspark
```

その後、Sparkのセッションを立ち上げます。appNameは適当で大丈夫です。
```
from pyspark.sql import SparkSession 
spark = SparkSession.builder \
                    .master("local[1]") \
                    .appName("IAB") \
                    .getOrCreate()
```

以下のようなSession情報が出れば、成功です。
```
SparkSession - in-memory
SparkContext

Version   v3.3.0
Master    local[1]
AppName   IAB
```


PySparkの基本的なデータ操作と変換については以下のような内容です。
- データの読み込みと表示: PySparkでは、様々なデータソースからデータを読み込むことができます。例えば、CSVファイル、JSONファイル、データベース、Parquetファイルなどです。データの読み込みには、spark.readメソッドを使用します。読み込んだデータはDataFrameとして表現され、.show()メソッドを使用して表示することができます。

- データのフィルタリングと選択: DataFrameでは、SQLライクなクエリを使用してデータのフィルタリングや選択を行うことができます。例えば、filter()メソッドやwhere()メソッドを使用して条件に一致する行を抽出することができます。また、select()メソッドを使用して特定の列を選択することも可能です。

- データの集計とグループ化: DataFrameでは、データの集計やグループ化を行うためのメソッドが提供されています。例えば、groupBy()メソッドを使用して特定の列でグループ化し、agg()メソッドを使用して集計関数（例: sum、avg、maxなど）を適用することができます。

- データの変換と追加: DataFrameでは、既存の列に対して変換を行ったり、新たな列を追加することができます。例えば、withColumn()メソッドを使用して既存の列に対して演算や関数を適用し、新たな列を追加することができます。

- DataFrameとRDDの違い: PySparkでは、DataFrameとRDD（Resilient Distributed Dataset）という2つのデータ構造が使用されます。DataFrameは、データを構造化して表現するための高レベルのAPIであり、スキーマ情報を持った行と列のテーブルとして扱います。一方、RDDは低レベルのAPIであり、分散処理の基本単位となる抽象化されたデータセットです。DataFrameはRDD上に構築されており、RDDよりも高レベルで使いやすく効率的な処理が可能です。

以上がPySparkの基本的なセットアップと基礎知識の説明です。次章では、実際にPySparkを使用してデータセットの読み込みや基本的なデータ操作を行う方法について詳しく解説します。


# データサイエンス100本ノックで実践
https://github.com/t-hashiguchi1995/100knock_pyspark/blob/main/preprocess_knock_Python_Spark.ipynb

以下で説明する内容を使って、対象のデータを変更していることが多いため、各課題についての詳細な解説については省略します。一部特殊なものについては解説いたします。

## PySparkでよく利用する機能
```
from pyspark.sql.types import *
import pyspark.sql.functions as F
from pyspark.sql.window import Window
```
**pyspark.sql.types** ：型変換する際にとても便利
**pyspark.sql.functions**：sum()やmean()などの計算用以外にもwhereのような条件指定処理を行うのにも使うので、とても便利
**pyspark.sql.window** ：pandasでいうlagなどの処理をするのに利用する。windowの分け方を指定できるので、活用できると便利になる


## データの読み込みと表示
まず、データサイエンス100本ノックで扱うデータを読み込みます。
**read** という機能があるので、こちらを利用します。

https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrameReader.html#pyspark.sql.DataFrameReader

```
df_customer = spark.read.format("csv")\
                   .options(header="true", inferSchema="true")\
                   .load("customer.csv")
```
や
```
df_customer = spark.read.csv("customer.csv")
```

ほとんどの課題がデータの内容を「**10件表示させよ。**」 というものなので、データの内容を表示する必要があります。
**show** という機能があるので、こちらを利用します。

https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.show.html
```
df_receipt.show(10, truncate=False)
```


## データのフィルタリングと選択
ある条件を満たすデータを抽出する課題があるため、条件指定を行う必要があります。
**filter** という機能があるので、こちらを利用します。

https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.filter.html

```
df_receipt.select(["sales_ymd", "customer_id", "product_cd", "amount"])\
          .filter(df_receipt["customer_id"]=="CS018205000001").show()
```
や
```
df_receipt.select(["sales_ymd", "customer_id", "product_cd", "amount"])\
          .filter((F.col("customer_id")=="CS018205000001") & ((F.col("amount")>=1000) & (F.col("amount")<=2000))).show()
```

データの対象を指定して表示するものがあるため、指定のカラム（列）のみを取得する必要があります。
**select** という機能があるので、こちらを利用します。

https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.select.html
```
df_receipt.select(["sales_ymd", "customer_id", "product_cd", "amount"]).show(10)
```

## データの集計とグループ化
データをグループ化して集約する方法として**groupby** という機能があるので、こちらを利用します。
groupby後にデータの集計を行うための多くの組み込み関数を利用することで、データの合計や平均、最大値・最小値の算出が可能になります。

https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.groupBy.html

https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.GroupedData.agg.html

```
df_receipt.groupBy("customer_id").max("sales_ymd").show(10)
```
や
```
df_receipt.groupBy("customer_id").agg({"sales_ymd":"min"}).show(10)
```

## データの変換と追加
既存のデータフレームに新しい列を追加したり、不要な列を削除したりすることができます。
**withColumn** や**drop**を使用して列の追加・変換と削除が可能になります。

https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.withColumn.html

https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.drop.html

「販売価格 - 費用」から利益を計算し、列として追加する
```
df_product.withColumn("unit_profit", F.col("unit_price") - F.col("unit_cost")).show(10)
```

カラムの型を変更する
```
df_customer.withColumn("birth_day", F.to_date("birth_day")
```

カラムを削除する
```
df_customer.drop(F.col("gender"))
```



## 一部問題の解説

### P-020:
> レシート明細データフレーム（df_receipt）に対し、1件あたりの売上金額（amount）が高い順にランクを付与し、先頭10件を抽出せよ。項目は顧客ID（customer_id）、売上金額（amount）、付与したランクを表示させること。なお、売上金額（amount）が等しい場合でも別順位を付与すること。

P-020以外にもWindowを利用する場面がありますが、例として紹介します。
```
df_receipt.withColumn("row_number", F.row_number().over(Window.orderBy(F.col("amount").desc())))\
          .select(["customer_id", "amount", "row_number"]).show(10)
```

Windowは **.partitionBy** と **.orderBy** を利用して活用します。
- partitionBy：データの単位をどこで切るのか。例：日単位でデータを使う場合は```partitionBy("日付カラム")```になります。
- orderBy：昇順、降順の対象となるカラムはどれか。例：日単位で1時間前のデータを使う場合は```partitionBy("日付カラム").orderBy("時間カラム")```になります。



### P-046:
> 顧客データフレーム（df_customer）の申し込み日（application_date）はYYYYMMD形式の文字列型でデータを保有している。これを日付型（dateやdatetime）に変換し、顧客ID（customer_id）とともに抽出せよ。データは10件を抽出すれば良い。

回答としては以下になります。
直接**to_date** メソッドを使って変換することができない点が注意点になります。
```
df_customer.withColumn("application_date", F.col('application_date').cast(StringType()))\
           .withColumn("application_date", F.to_date(F.unix_timestamp(F.col("application_date"), "yyyyMMdd").cast("timestamp")))\
           .select(["customer_id", "application_date"]).show(10)
```

### P-054:
> 顧客データデータフレーム（df_customer）の住所（address）は、埼玉県、千葉県、東京都、神奈川県のいずれかとなっている。都道府県毎にコード値を作成し、顧客ID、住所とともに抽出せよ。値は埼玉県を11、千葉県を12、東京都を13、神奈川県を14とすること。結果は10件表示させれば良い。

```
df_customer.withColumn("prefecture_cd", F.when(F.substring("address", 1, 3)=="埼玉県", "11")\
                                         .when(F.substring("address", 1, 3)=="千葉県", "12")\
                                         .when(F.substring("address", 1, 3)=="東京都", "13")\
                                         .when(F.substring("address", 1, 3)=="神奈川", "14")).show(10)
```

列追加の際に**when** を利用することである条件の場合の値を入れることが出来ます。
whenの条件に当てはまらない場合、**otherwise**を利用することで値を入れることが可能です。（otherwiseがない場合はNull値になります。）



# 応用編
PySparkでは、カスタム機能の実装や拡張が可能です。（P-028などで利用しています。）
以下に、PySparkのUDF（ユーザー定義関数）の作成方法と使用例、そしてPySparkの拡張機能やカスタムライブラリの利用方法を紹介します。
- PySparkのUDFの作成方法と使用例:PySparkでは、UDFを作成してDataFrameの列に適用することができます。
- UDFを作成するには、Pythonの関数を定義してudf()メソッドを使用します。

具体的な手順は以下の通りです。

## 1. Pythonの関数を定義する
```
@F.udf(DoubleType())
def udf_median(values_list):
    med = np.median(values_list)
    return float(med)
```

## 2. DataFrameにUDFを適用する
```
df_receipt.groupBy("store_cd").agg(udf_median(F.collect_list("amount")).alias("median_amount")).show(5)
```