Apache Sparkの初心者が環境構築とPySparkでのデータ集計までやってみる

# はじめに

Apache Sparkの初心者がPySparkで、DataFrame API、SparkSQL、Pandasを動かしてみた際のメモです。
Hadoop、Sparkのインストールから始めていますが、インストール方法等は何番煎じか分からないほどなので自分用のメモの位置づけです。

# 環境

お試し用に以下のバージョンのHadoop(HDFS)とSparkをインストールします。

・Apache Hadoop 3.1.1
・Apache Spark 2.3.2

OSはCentOS 7.4を使用します。
お試しですので、Hadoopは疑似分散を使用します。
HadoopはSpark付属のものでも良かったのですが、今回は別にインストールしています。
Sparkはスタンドアロンモードを使用します。

## JDKインストール

最初にOpenJDK 8をインストールします。

```terminal
# yum -y install java-1.8.0-openjdk-devel

# java -version
openjdk version "1.8.0_191"
OpenJDK Runtime Environment (build 1.8.0_191-b12)
OpenJDK 64-Bit Server VM (build 25.191-b12, mixed mode)
```

JAVA_HOMEを設定し、PATHにJavaのパスを追加しています。

```terminal
# echo "export JAVA_HOME=$(readlink -e $(which java)|sed 's:/bin/java::')" > /etc/profile.d/java.sh
# echo "PATH=\$PATH:\$JAVA_HOME/bin" >> /etc/profile.d/java.sh
# source /etc/profile
```

# Hadoopインストール

Hadoopを疑似分散モードでインストールします。
以下のサイト・投稿を参考にしています。

- [Hadoop: Setting up a Single Node Cluster.](http://hadoop.apache.org/docs/r3.1.1/hadoop-project-dist/hadoop-common/SingleCluster.html)
- [@IT Hadoopの疑似分散モードと完全分散モードを試す](http://www.atmarkit.co.jp/ait/articles/1701/01/news015.html)
- [Hadoop 2.7.1を擬似分散モードで動かす](https://qiita.com/bwtakacy/items/6491b3d11754492f9702)
- [hadoopを疑似分散で動かす](https://qiita.com/cironaga/items/06ad0cacfeaa7c5e0c52)
- [Apache Hadoopのインストール手順](https://qiita.com/Esfahan/items/39fd1e2f8b755eacec65)
 
まず、Hadoopのモジュールを以下からダウンロードします。

　http://hadoop.apache.org/releases.html 

以下の3.1.1.をダウンロードし、「/opt」以下に展開しました。

![image.png](https://qiita-image-store.s3.amazonaws.com/0/179607/5ec5dc67-11b4-012c-b250-1aea21f295ef.png)

```
# tar xvzf hadoop-3.1.1.tar.gz -C /opt
# ln -s /opt/hadoop-3.1.1 /opt/hadoop
HDFSのデータ用ディレクトリを作成します。
# mkdir -p /var/hadoop/data
```

必要な環境変数を定義します。

```/etc/profile.d/hadoop.sh
export HADOOP_HOME=/opt/hadoop
export PATH=$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$PATH
export HADOOP_COMMON_LIB_NATIVE_DIR=$HADOOP_HOME/lib/native
export HADOOP_OPTS="-Djava.library.path=$HADOOP_HOME/lib/native"
export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
```

Hadoop/Spark用のユーザ（Hadoop）を作成します。

```
# useradd -s /bin/bash -m hadoop -p XXXXXXXX ※XXXXXXXXはパスワード（任意）を指定してください
# chown -R hadoop:hadoop /opt/hadoop-3.1.1
# chown -R hadoop:hadoop /opt/hadoop
# chown -R hadoop:hadoop /var/hadoop/data
# su - hadoop
```

擬似分散モードで動かすための設定の以下のファイルを修正します。

・HDFS用のディレクトリを/var/hadoop/dataに指定

```/opt/hadoop/etc/hadoop/core-site.xml
<configuration>
    <property>
        <name>fs.default.name</name>
        <value>hdfs://localhost:9000</value>
    </property>
    <property>
        <name>hadoop.tmp.dir</name>
        <value>/var/hadoop/data</value>
    </property>
</configuration>
```

・今回は1ノードのためレプリケーションを"1"に設定

```/opt/hadoop/etc/hadoop/hdfs-site.xml
<configuration>
    <property>
        <name>dfs.replication</name>
        <value>1</value>
    </property>
</configuration>
```

Hadoop は SSH 経由でホストを操作するので、"ssh localhost" でパスフレーズなしでログインできる必要があります。

```
$ ssh-keygen -t dsa -P '' -f ~/.ssh/id_dsa
$ cat ~/.ssh/id_dsa.pub >> ~/.ssh/authorized_keys
$ chmod 600 ~/.ssh/authorized_keys
パスフレーズなしで設定できることを確認します。
$ ssh localhost
```

## ファイルシステムをフォーマット

```bash
$ hdfs namenode -format
WARNING: /opt/hadoop/logs does not exist. Creating.
2018-10-19 23:00:14,379 INFO namenode.NameNode: STARTUP_MSG: 
/************************************************************
STARTUP_MSG: Starting NameNode
STARTUP_MSG:   host = sparkserver1/127.0.0.1
STARTUP_MSG:   args = [-format]
STARTUP_MSG:   version = 3.1.1

～省略～
2018-10-19 23:00:32,388 INFO common.Storage: Storage directory /var/hadoop/data/dfs/name has been successfully formatted.
～省略～

/************************************************************
SHUTDOWN_MSG: Shutting down NameNode at sparkserver1/127.0.0.1
************************************************************/
```

## HDFSの起動

```
$ start-dfs.sh
Starting namenodes on [localhost]
Starting datanodes
Starting secondary namenodes [sparkserver1]

jpsでDataNode、NameNode、SecondaryNameNodeの3つのプロセスが起動していることを確認します。

$ jps 
21505 DataNode
21829 Jps
21388 NameNode
21709 SecondaryNameNode
```

HDFSを参照できることを確認するために以下のコマンドを実行します。
フォーマット直後のため、結果は何も表示されません。

```
$ hadoop fs -ls /
```


# Apache Sparkのインストール

以下を参考にしています。

[Spark Standalone Mode](https://spark.apache.org/docs/latest/spark-standalone.html)
[Apache Spark で分散処理入門](https://qiita.com/Hiroki11x/items/4f5129094da4c91955bc)

以下から、version 2.3.2をダウンロードし、インストールします。

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

package typeは、既にHadoopをインストールしているため、Hadoopを含まない「Pre-build with user-provided Apache Hadopp」を選択しています。

![image.png](https://qiita-image-store.s3.amazonaws.com/0/179607/1f45e8ac-2f4f-d6f5-c4fa-af111856a634.png)

「/opt」以下に展開しました。

```
# tar xvzf spark-2.3.2-bin-without-hadoop.tgz -C /opt
# ln -s /opt/spark-2.3.2-bin-without-hadoop /opt/spark
```

```/etc/profile.d/spark.sh
export PATH=$SPARK_HOME/bin:$SPARK_HOME/sbin:$PATH
export SPARK_HOME=/opt/spark
```

```
# chown -R hadoop:hadoop /opt/spark-2.3.2-bin-without-hadoop
# chown -R hadoop:hadoop /opt/spark
# su - hadoop
```

「[Using Spark's "Hadoop Free" Build](https://spark.apache.org/docs/latest/hadoop-provided.html)」に記載されていましたが、HadoopなしのSparkをインストールする場合は、以下のようにSPARK_DIST_CLASSPATHにHadoopのクラスパスを設定します。

```
$ cp -p /opt/spark/conf/spark-env.sh.template /opt/spark/conf/spark-env.sh
$ vi /opt/spark/conf/spark-env.sh
export SPARK_DIST_CLASSPATH=`hadoop classpath`
```

Sparkを起動します。

```
$ start-all.sh 
starting org.apache.spark.deploy.master.Master, logging to /opt/spark/logs/spark-hadoop-org.apache.spark.deploy.master.Master-1-sparkserver1.out
localhost: starting org.apache.spark.deploy.worker.Worker, logging to /opt/spark/logs/spark-hadoop-org.apache.spark.deploy.worker.Worker-1-sparkserver1.out
```

Sparkを起動すると、Spark-UIも利用できるようになります。

http://[Sparkが起動しているサーバのIPアドレス]:8080/

![image.png](https://qiita-image-store.s3.amazonaws.com/0/179607/13a15634-7976-c163-758e-8c6fc899b419.png)



# いよいよPySparkを使う

HDFS上のCSVを読み込んで、PySparkからデータを操作しています。

## データを準備する

まずは手頃なCSVファイルが欲しかったので、kaggleから以下のデータをダウンロードしました。

https://www.kaggle.com/datasets

![image.png](https://qiita-image-store.s3.amazonaws.com/0/179607/a671210a-809d-e915-c230-4e613492b63c.png)

データは以下のように、パン屋さんのお客さんごとの購入物になっています。
PySparkで商品ごとの売り上げを集計してみます。

PySparkといっても色々やり方があるようで、DataFrame API、SparkSQL、Pandasの3つで同じように試してみます。

```csv
Date,Time,Transaction,Item
2016-10-30,09:58:11,1,Bread
2016-10-30,10:05:34,2,Scandinavian
2016-10-30,10:05:34,2,Scandinavian
2016-10-30,10:07:57,3,Hot chocolate
2016-10-30,10:07:57,3,Jam
2016-10-30,10:07:57,3,Cookies
2016-10-30,10:08:41,4,Muffin
2016-10-30,10:13:03,5,Coffee
2016-10-30,10:13:03,5,Pastry
2016-10-30,10:13:03,5,Bread
2016-10-30,10:16:55,6,Medialuna
2016-10-30,10:16:55,6,Pastry
2016-10-30,10:16:55,6,Muffin
```

## DataFrame API

以下を参考にしています。

[Spark SQLとDataFrame API入門](https://dev.classmethod.jp/hadoop/hadoop-advent-calendar-16-spark-sql-and-dataframe/)
[PySparkのチートシート](https://www.datacamp.com/community/blog/pyspark-sql-cheat-sheet)

まずは、HDFS にディレクトリを作成し、サンプルデータのCSVファイルを格納します。

```bash
$ hadoop fs -mkdir /data/
$ hadoop fs -put BreadBasket_DMS.csv /data/
$ hadoop fs -ls /data
Found 1 items
-rw-r--r--   1 hadoop supergroup     710518 2018-10-21 02:56 /data/BreadBasket_DMS.csv
```

PySparkを起動します。

```
$ pyspark --name sample-app --master spark://localhost:7077
```

まず、HDFS上のCSVファイルを読み込んで、最初の5行を表示します。

```python
from pyspark.sql.types import *
struct = StructType([
        StructField('bb_date', DateType(), False),
        StructField('bb_time', StringType(), False),
        StructField('bb_tran', IntegerType(), False),
        StructField('bb_item', StringType(), False)
     ])

df = spark.read.csv('hdfs://localhost:9000/data/BreadBasket_DMS.csv', schema=struct)

df.show(5)
```

```
+----------+--------+-------+-------------+
|   bb_date| bb_time|bb_tran|      bb_item|
+----------+--------+-------+-------------+
|      null|    null|   null|         null|
|2016-10-30|09:58:11|      1|        Bread|
|2016-10-30|10:05:34|      2| Scandinavian|
|2016-10-30|10:05:34|      2| Scandinavian|
|2016-10-30|10:07:57|      3|Hot chocolate|
+----------+--------+-------+-------------+
only showing top 5 rows
```

最初のnullはCSVのヘッダのようです。集計に影響はないのでこのまま進めます。

次に、商品ごとにグルーピングし、売上をカウントして降順に並び替えます。

```python
>>> df.groupBy('bb_item').count().sort('count', ascending=False).show(5)
+-------+-----+                                                                 
|bb_item|count|
+-------+-----+
| Coffee| 5471|
|  Bread| 3325|
|    Tea| 1435|
|   Cake| 1025|
| Pastry|  856|
+-------+-----+
only showing top 5 rows
```

トップはCoffeeで、5471杯の売り上げでした。

SparkUIではDAGが以下のように表示できます。

![image.png](https://qiita-image-store.s3.amazonaws.com/0/179607/8e2a1eb6-0256-226c-98b7-9bf736afb98e.png)


## SparkSQL

次にSparkSQLで同様のことを試してみます。
CSVファイルを読み込むところは同じです。

参考：
[Spark SQLサンプルアプリの実行](https://qiita.com/t-yotsu/items/4cabd1ae5406cfd7d741)

```python
struct = StructType([
        StructField('bb_date', DateType(), False),
        StructField('bb_time', StringType(), False),
        StructField('bb_tran', IntegerType(), False),
        StructField('bb_item', StringType(), False)
     ])

df = spark.read.csv('hdfs://localhost:9000/data/BreadBasket_DMS.csv', schema=struct)
```

bbという一時的なテーブルを作成し、最初の5行を表示しています。

```python
>>> df.registerTempTable('bb')
>>> spark.sql('select * from bb').show(5)
+----------+--------+-------+-------------+
|   bb_date| bb_time|bb_tran|      bb_item|
+----------+--------+-------+-------------+
|      null|    null|   null|         null|
|2016-10-30|09:58:11|      1|        Bread|
|2016-10-30|10:05:34|      2| Scandinavian|
|2016-10-30|10:05:34|      2| Scandinavian|
|2016-10-30|10:07:57|      3|Hot chocolate|
+----------+--------+-------+-------------+
only showing top 5 rows

```

最後に集計しています。当然ですが、DataFrame APIを用いた時と同じ結果になっています。

```python
>>> spark.sql('select bb_item, count(*) as n from bb group by bb_item order by n desc').show(5)
+-------+----+                                                                  
|bb_item|   n|
+-------+----+
| Coffee|5471|
|  Bread|3325|
|    Tea|1435|
|   Cake|1025|
| Pastry| 856|
+-------+----+
only showing top 5 rows
```

## Pandas

最後はPandasで試してみます。
Pandasは、Pythonでのデータ解析に関する機能を提供するライブラリです。
以下を参考にしました。

[Pandas 公式チートシートを翻訳しました](https://qiita.com/s_katagiri/items/4cd7dee37aae7a1e1fc0)
[https://qiita.com/ysdyt/items/9ccca82fc5b504e7913a](https://qiita.com/ysdyt/items/9ccca82fc5b504e7913a)

まずは、pipが入っていなかったのでインストールして、そのままPandasもインストールします。

```
# wget https://bootstrap.pypa.io/get-pip.py
# python get-pip.py
# pip install pandas
```

PySparkで、最初にCSVファイルを読み込んでいます。DataFrame API、SparkSQLではHDFSから読み込んでいたのですが、Pandasでは若干面倒なようなので今回はローカルファイルから読み込んでいます。

```python
import pandas as pd
names = ['bb_date', 'bb_time', 'bb_tran', 'bb_item']
pandasdf = pd.read_csv('/tmp/BreadBasket_DMS.csv', names=names)
```

最初の5行を表示。

```python
>>> pandasdf.head(5)
      bb_date   bb_time      bb_tran        bb_item
0        Date      Time  Transaction           Item
1  2016-10-30  09:58:11            1          Bread
2  2016-10-30  10:05:34            2   Scandinavian
3  2016-10-30  10:05:34            2   Scandinavian
4  2016-10-30  10:07:57            3  Hot chocolate
```

最後に集計結果を表示します。

```python
>>> pandasdf['bb_item'].value_counts().head(5)
Coffee    5471
Bread     3325
Tea       1435
Cake      1025
Pastry     856
Name: bb_item, dtype: int64
```

# 最後に

最後にpyファイルで実行しようとしたが失敗。
pythonファイルは以下のとおり。

```python
#!/usr/bin/env python
# -*- coding: utf-8 -*-

import pandas as pd

names = ['bb_date', 'bb_time', 'bb_tran', 'bb_item']
pandasdf = pd.read_csv('/tmp/BreadBasket_DMS.csv', names=names)
pandasdf['bb_item'].value_counts().head(5)
```

これをpysparkで実行しようとすると、spark-submitを使えと怒られる。

```
pyspark --name sample-app --master spark://localhost:7077 sample.py
Running python applications through 'pyspark' is not supported as of Spark 2.0.
Use ./bin/spark-submit <python file> 
```

spark-submitで実行すると、エラーにはならなかったが、結果が表示されず。これはまた今度調べてみよう。

```
$ spark-submit --name sample-app --master spark://localhost:7077 sample.py 
2018-10-21 06:49:39 WARN  Utils:66 - Your hostname, sparkserver1 resolves to a loopback address: 127.0.0.1; using 10.0.2.15 instead (on interface enp0s3)
2018-10-21 06:49:39 WARN  Utils:66 - Set SPARK_LOCAL_IP if you need to bind to another address
2018-10-21 06:49:40 WARN  NativeCodeLoader:60 - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2018-10-21 06:49:40 INFO  ShutdownHookManager:54 - Shutdown hook called
2018-10-21 06:49:40 INFO  ShutdownHookManager:54 - Deleting directory /tmp/spark-0edd9099-fe73-4ebb-b7dd-5b9b9cc7829e

```