Windows上でPysparkを用いたデータ処理③　～PySpark実践編：csvから読み込み、簡単な操作まで～


# はじめに
2年目に突入したエンジニアでございます。

今回は突然Sparkを用いてデータ処理を行うことになり、
周りがMacPCやVirtualBoxでのLinux環境を使用する中、
なんとかWindowsメインで何とかならないかと試行錯誤した過程の第三弾です。

前回まではWindows上でJupyterNotebookを起動するところまで確認しました。

https://qiita.com/ys_kmr/items/160d8453e8a43dc98adf

https://qiita.com/ys_kmr/items/10147791a0e187dc3cc4

# 今回の目的：csvからサンプルデータを読み込んで簡単な処理をしたい

[1. サンプルデータの用意](#1-サンプルデータの用意)
[2. sparkの初期化](#2-sparkの初期化)
[3. csvからデータフレームに読み込み](#3-csvからデータフレームに読み込み)
[4. 読み込んだデータの中身、構成を表示](#4-読み込んだデータの中身、構成を表示)
[5. スキーマの設定](#5-スキーマの設定)
[6. フィルターと並び替え](#6-フィルターと並び替え)

## 1. サンプルデータの用意
まずは今回使用するためのデータをcsvファイルで用意します。
簡単な処理しか行わないため、
名前と年齢のみのデータを下記のようにデータを入れるフォルダを作成してその中に用意しました。
```csv:csv/sample_01.csv
name,age
tanaka,20
satou,30
suzuki,25
```
## 2. sparkの初期化
さて、ここからようやくSparkについてコードを記載していきます。
Sparkアプリケーションを始める際には下記の処理が必要です。

```python
from pyspark.sql import SparkSession

# Sparkセッションの作成
spark = SparkSession.builder.getOrCreate()
```
この処理の中で設定オプションを指定することでSparkをカスタマイズすることが可能ですが、
動かすだけであれば上記のシンプルなコードで問題ありません。

要は、Sparkで処理したい場合には必要なステップで今後上記コードで作成した「spark」からSparkの機能を利用できるということです。

## 3. csvからデータフレームに読み込み

Sparkを利用する準備がようやく整ったので、実際にcsvからデータを読み込んでみましょう。
オプションで「hedder=True」とすることで、csvの一行目をカラム名として読み込んでくれます。

```python
# csvファイルをDataFrameとして読み込む
df = spark.read.csv("./csv/sample_01.csv",hedder=True)
```

これでcsvからデータを読み込むことができました。
DataFrameはSQLのテーブルのような形でデータを持つことができ、
一つ一つのデータに対してではなく、DataFrame全体に一括して処理をするのに適しています。

## 4. 読み込んだ中身、構成の表示

前項で読み込んだデータをJupyterNotebook上で表示して確認します。
「show()」がデータの一覧を表示、「printSchema()」がデータフレームのスキーマ（データ型）を表示する関数です。
show関数は表示する件数や省略される文字列の長さなどを調整可能なので、
データの中身によっては調整が必要です。

```python
# DataFrameの内容を表示する
df.show()
df.printSchema()
```
```text:出力
+------+---+
|  name|age|
+------+---+
|tanaka| 20|
| satou| 30|
|suzuki| 25|
+------+---+

root
 |-- name: string (nullable = true)
 |-- age: string (nullable = true)
```
読み込めてますね！！
ただ、「age」の方がinteger（整数型）ではなくstring（文字列型）になってますね…

## 5. スキーマの設定

「age」はintegerであってほしいので処理を加えましょう。
そのためには、読み込む際にスキーマ（データ型）を指定する必要があります。
具体的には下記の通り記載することで実現可能です。
```python:
from pyspark.sql.types import StructType,StructField,StringType,IntegerType
schema = StructType([
    StructField("name", StringType(), nullable=False),
    StructField("age", IntegerType(), nullable=False)
])

# csvファイルをDataFrameとして読み込む
df = spark.read.csv("./csv/sample_01.csv",header=True,schema=schema)

# DataFrameの内容を表示する
df.show()
df.printSchema()
```
```text:出力
+------+---+
|  name|age|
+------+---+
|tanaka| 20|
| satou| 30|
|suzuki| 25|
+------+---+

root
 |-- name: string (nullable = true)
 |-- age: integer (nullable = true)
```

少し複雑になってきましたが、しっかり想定通りの型でcsvからデータフレームを作成することができました。

（一応読み込んでからカラムをキャストして置き換えることで型を後から変更することは可能ですが、初めからわかっているのであればスキーマを用いるほうを推奨したいです。）

```python
# csvファイルをDataFrameとして読み込む
df = (
    spark.read.csv("./csv/sample_01.csv",header=True)
    .withColumn("age",col("age").cast("integer"))
)

# DataFrameの内容を表示する
df.show()
df.printSchema()
```
```text:出力
+------+---+
|  name|age|
+------+---+
|tanaka| 20|
| satou| 30|
|suzuki| 25|
+------+---+

root
 |-- name: string (nullable = true)
 |-- age: integer (nullable = true)
```

## 6. フィルターと並び替え

最後に最も基本的な処理だけ紹介します。
SQLでもよく用いるものといえばフィルターと並び替えですよね。
Pysparkでは下記のようにわかりやすい関数で使用可能です。

```python
from pyspark.sql.functions import col

# 年齢20歳以上のデータを名前順で表示
(
    df
    .filter(col("age")>20)
    .orderBy(col("name"))
    .show()
)

# 名前に「a」が含まれるデータを年齢降順で表示
(
    df
    .filter(col("name").like("%a%"))
    .orderBy(col("age").desc())
    .show()
)
```
```text:出力
+------+---+
|  name|age|
+------+---+
| satou| 30|
|suzuki| 25|
+------+---+

+------+---+
|  name|age|
+------+---+
| satou| 30|
|tanaka| 20|
+------+---+
```

できました！

### 余談

個人的にお勧めの設定を一つ紹介したいと思います。
それは「**列番号表示**」です。

エラーなどで修正したいときに列番号が分かると嬉しいと思います。
初期設定でオンにしておいてほしいですね…

設定の仕方は、左上の「View」タブから「Show Line Numbers」をクリック！それだけです。
![show_linenumber_02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2660549/e5080add-5bb2-f22e-4141-26a68c864b6a.png)


- 表示前
![show_linenumber_01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2660549/c419372f-2a4a-f83e-653c-4c3861586c43.png)
- 表示後
![show_linenumber_03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2660549/f3c0446b-e374-1317-1713-bb2ca7831e71.png)


# おわりに

最後まで読んでいただきありがとうございます。

今回はようやくWindows上に起動したJupyterNotebookで、
sparkに初期化してデータを読み込む方法をまとめました。
少しでも参考になると幸いです。

ありがとうございました。

# 参考

https://stackoverflow.com/questions/tagged/apache-spark-dataframe

https://spark-packages.org/

https://docs.databricks.com/

https://spark.apache.org/documentation.html