Google ColabでPySparkとSpark SQLを動かしてみよう

## はじめに

このハンズオンでは、Google Colabを使ってPySparkとSpark SQLを利用し、データ変換、操作、結合を行います。両方の方法で同じ処理を実現し、それぞれの違いや使いどころを学びます。

PySparkはPythonのフレームワークでSparkを利用するためのライブラリであり、Spark SQLはSQL文を用いてデータ操作を行う方法です。どちらもApache Sparkの機能を利用していますが、プログラミングスタイルや記述方法が異なります。本記事では、それぞれの特徴を比較しながら進めていきます。

## **1. 環境セットアップ**

Google ColabでのApache Sparkセットアップを以下のコードで行います。

```python
# Sparkのインストール
!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget -q https://archive.apache.org/dist/spark/spark-3.4.0/spark-3.4.0-bin-hadoop3.tgz
!tar xf spark-3.4.0-bin-hadoop3.tgz
!pip install -q findspark

# 環境変数の設定
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-3.4.0-bin-hadoop3"

# findsparkを初期化
import findspark
findspark.init()
```

Sparkセッションを作成します。

```python
# Sparkセッションの作成
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("PySparkAndSparkSQL") \
    .getOrCreate()

print("Sparkセッションが開始されたよん")
```


## **2. サンプルデータの準備**

2つのデータセットを用意して、欠損値を含むデータ処理や結合操作を含む処理を行います。

```python
# メインデータ
data_people = [
    ("たかし", 29, "データサイエンティスト"),
    ("はなこ", 35, "医者"),
    ("たろう", None, "先生"),
    ("みき", 29, None),
    ("ひろき", 40, "マネージャ")
]
columns_people = ["Name", "Age", "Profession"]

# サブデータ（職業の説明）
data_profession = [
    ("エンジニア", "システム開発"),
    ("医者", "医療行為"),
    ("先生", "教育"),
    ("データサイエンティスト", "データ分析")
]
columns_profession = ["Profession", "Description"]

# データフレーム作成
df_people = spark.createDataFrame(data_people, columns_people)
df_profession = spark.createDataFrame(data_profession, columns_profession)

# データの表示
print("Peopleデータ:")
df_people.show()

print("Professionデータ:")
df_profession.show()
```
#### おまけに
CSVデータなどを読み込む際は以下のように記述して、hedder=Trueを指定することで、最初の１行目をカラム名として認識してくれます。
```python
# csvファイルをDataFrameとして読み込む
df = spark.read.csv("ファイルパス",hedder=True)
```


## **3. データ操作: フィルタリングと変換**

### 3.1 フィルタリング
**目的**: 年齢が30歳以上の人を抽出。

#### PySparkでの処理
```python
# PySparkでフィルタリング
df_filtered = df_people.filter(df_people["Age"] >= 30)
df_filtered.show()
```

#### Spark SQLでの処理
```python
# テーブル登録
df_people.createOrReplaceTempView("people")

# Spark SQLでフィルタリング
sql_filtered = spark.sql("SELECT * FROM people WHERE Age >= 30")
sql_filtered.show()
```



### 3.2 列の選択と変換
**目的**: 名前と年齢の列を選択し、年齢を1歳加算。

#### PySparkでの処理
```python
# PySparkで列操作
from pyspark.sql.functions import col

df_transformed = df_people.select(
    col("Name"),
    (col("Age") + 1).alias("AgePlusOne")
)
df_transformed.show()
```

#### Spark SQLでの処理
```python
# Spark SQLで列操作
sql_transformed = spark.sql("""
    SELECT 
        Name, 
        Age + 1 AS AgePlusOne
    FROM people
""")
sql_transformed.show()
```


## **4. データ結合（LEFT JOIN）**

### 4.1 PySparkでの結合
```python
# PySparkでLEFT JOIN
df_joined = df_people.join(df_profession, on="Profession", how="left")
df_joined.show()
```

### 4.2 Spark SQLでの結合
```python
# テーブル登録
df_profession.createOrReplaceTempView("profession")

# Spark SQLでLEFT JOIN
sql_joined = spark.sql("""
    SELECT 
        p.Name, 
        p.Age, 
        p.Profession, 
        pr.Description
    FROM people p
    LEFT JOIN profession pr
    ON p.Profession = pr.Profession
""")
sql_joined.show()
```


## **5. 欠損値の処理**

### 5.1 欠損値の確認

#### PySparkでの確認
```python
# 欠損値の確認
df_people.select([col for col in df_people.columns if df_people[col].isNull().any()]).show()
```

#### Spark SQLでの確認
```python
# Spark SQLで欠損値確認
sql_nulls = spark.sql("SELECT * FROM people WHERE Age IS NULL OR Profession IS NULL")
sql_nulls.show()
```



### 5.2 欠損値の補完

#### PySparkでの補完
```python
# 欠損値の補完
df_filled = df_people.fillna({"Age": 0, "Profession": "未設定"})
df_filled.show()
```

#### Spark SQLでの補完
```python
# Spark SQLで欠損値補完
sql_filled = spark.sql("""
    SELECT 
        Name, 
        COALESCE(Age, 0) AS Age, 
        COALESCE(Profession, '未設定') AS Profession
    FROM people
""")
sql_filled.show()
```

お疲れ様でした。Sparkのセッションを終了するときは **spark.stop()** で終了できます。
## **6. PySparkとSpark SQLの比較まとめ**

| **操作内容**       | **PySpark**                     | **Spark SQL**                  |
|--------------------|----------------------------------|---------------------------------|
| データフィルタリング | `df.filter(df["Age"] >= 30)`    | `SELECT * FROM people WHERE Age >= 30` |
| 列の選択と変換      | `df.select(col("Name"), ...)`   | `SELECT Name, Age + 1 AS ...`  |
| データの結合         | `df.join(df2, on="key", how="left")` | `SELECT ... FROM table1 LEFT JOIN table2` |
| 欠損値の補完        | `df.fillna({"col": value})`     | `COALESCE(col, value)`         |



## **8. 終わりに**

PySparkとSpark SQLを比較しながら操作を学ぶことで、両方の特徴と利点を理解できます。どちらの方法もデータ分析やETL処理に役立ちます。用途や目的に応じて適切な方法を選択してください。何か質問や追加の課題があれば、ぜひコメントしてください！