Spark DataFrameの基本的な操作の確認をしてみた

# はじめに
Sparkの学習をするにあたり、DatabricksでApache Spark Python (PySpark) DataFrame APIを使用して、基本的なDataFrameの操作をしてみました。
操作するにあたり、以下のDatabricksのチュートリアルを参考に実施しました。
https://docs.databricks.com/aws/ja/getting-started/dataframes
本記事では、試してみた操作を実際のコードとともに説明します。

# DataFrameとは
Sparkにおいて、DataFrameは名前付きの列で構成された分散データの集合です。概念的には、リレーショナルデータベースのテーブルやR/Pythonのデータフレームに相当しますが、内部的にはより高度な最適化が施されています。
DataFrameは、以下のようなさまざまなデータソースから構築できます：
・構造化データファイル（CSV、JSONなど）
・Hiveのテーブル
・外部データベース
・既存のRDD（Resilient Distributed Dataset）

※参考：https://www.databricks.com/blog/2015/02/17/introducing-dataframes-in-spark-for-large-scale-data-science.html

# 基本的なDataFrameの操作
ここからは、[PySpark](https://www.databricks.com/jp/glossary/pyspark)というSparkを実行するためのPython APIを使用した、DataFrameの操作をいくつか紹介します。

## DataFrameの作成
### 手動で作成する方法
データとカラム名を定義し、`createDataFrame`で作成します。
```ruby:Python
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")
```
### PySparkSQLを使用する方法
[PySparkSQL](https://www.databricks.com/jp/glossary/pyspark)は、膨大な量の構造化・半構造化データにSQLライクな分析を行うPySparkライブラリです。PySparkSQLの`spark.sql()`を使用してSQLクエリを実行することができます。
ここでは、事前にDatabricksのカタログに作成しておいたテーブルから全件を取得し、そのデータをDataFrameとして格納しています。
```ruby:Python
df2 = spark.sql("SELECT * FROM {workspace_name}.{schema_name}.{table_name}")
```

## DataFrameの表示
`show`または`display`でDataFrameを表示します。
`display`の方が、よりグラフィカルに結果を表示することができます。
```ruby:Python
df1.show()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2726759/2260c06b-f51e-4a51-938b-322f2a20da99.png)

```ruby:Python
display(df1)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2726759/d1413060-8123-41ae-8a10-ffb98137b3b6.png)

## スキーマの表示
`printSchema`でDataFrameのスキーマを表示します。
```ruby:Python
df2.printSchema()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2726759/c81b6f11-f8a0-4c4c-867c-b2bd32700cf3.png)

## カラム名の変更
`withColumnRenamed`でDataFrameのカラム名を変更します。
ここでは、カラム名を"First Name"から"First_Name"に変更しています。
```ruby:Python
df2 = df2.withColumnRenamed("First Name", "First_Name")
df2.printSchema()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2726759/6bc240d1-8dac-4353-974b-12c29bafdbbb.png)

## DataFrameの結合
`union`でDataFrameを結合します。
ここでは、「DataFrameの作成」で作成した2つのDataFrameを結合しています。
1行目にdf1のデータが追加されてることを確認できました。
```ruby:Python
df3 = df1.union(df2)
display(df3)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2726759/693838b1-cbb9-433d-b742-5a15c88f727d.png)

## フィルタリング
`filter`または`where`でDataFrameの行をフィルタリングします。
（`filter`と`where`でパフォーマンスや構文に違いはありません。）
ここでは、"Count"カラムが50より大きいレコードでフィルタリングしています。
```ruby:Python
display(df3.filter(df3["Count"] > 50))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2726759/d1431568-f4f5-44de-b634-341ff5a0927a.png)
```ruby:Python
display(df3.where(df3["Count"] > 50))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2726759/23ba8ab2-c625-4fe7-87ef-613ebc3f66ac.png)

## カラムの選択とソート
`select`でDataFrameから取得するカラムを指定します。
また、`orderby`および`desc`でソートします。
ここでは、DataFrameから"First_Name"と"Count"カラムを選択し、"Count"カラムの降順でソートしています。
```ruby:Python
display(df2.select("First_Name", "Count").orderBy(desc("Count")))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2726759/d0898c6e-8747-46a1-9034-4b94eeed0b0e.png)

## DataFrameのサブセットの作成
既存のDataFrameから新しいDataFrame（サブセット）を作成します。
ここでは、`filter`や`select`を使用して既存のDataFrameからデータを取得し、新しいDataFrameとして格納しています。
```ruby:Python
subsetDF = df3.filter((df3["Year"] == 2009) & (df3["Count"] > 100) & (df3["Sex"] == "F")).select("First_Name", "County", "Count").orderBy(desc("Count"))
display(subsetDF)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2726759/fd7b5616-a169-4142-ac95-4fa1cbcd6e68.png)

## DataFrameをテーブルとして保存
`saveAsTable`でDataFrameをテーブルとして保存します。
ここでは、「DataFrameのサブセットの作成」で作成したDataFrameを、保存するパスやテーブル名を指定して、保存しています。
```ruby:Python
subsetDF.write.saveAsTable("{workspace_name}.{schema_name}.{table_name}")
```

# おわりに
シンプルな操作が多く、初心者でもDataFrameの操作の感触を掴むことができました。
今回試したのはほんの一部なので、今後はさらに実践を重ねることで、より複雑なデータ処理にも挑戦していきたいと思います。

# 関連記事

https://qiita.com/Akita123/private/6396fc6f489ce6446eaf

https://qiita.com/Tatikawa/items/47a57300fbcb5d47b162