Spark简明教程

(Spark简明教程 1.1.2.md)

1.What is Apache Spark?

面向大规模数据分析的统一引擎（Unified engine for large-scale data analytics）[1]。大规模，为了处理几百GB的TSP数据，我们需要一个面向大规模数据处理的工具。

Apache Spark是一个用于在单节点机器或集群机器上执行数据工程、数据科学和机器学习的多语言引擎（Apache Spark is a multi-language engine for executing data engineering, data science, and machine learning on single-node machines or clusters）[1]。单节点机器或集群机器，这样的灵活性使得使用者在开发、调试和部署等各个环节之间的使用是透明地。数据工程、数据科学和机器学习，面向TSP数据这样的无监督数据分析用户行为从而实现用户行为预测，是一种典型的机器学习应用。

2.What are competitors of Apache Spark?

• Apache Hadoop
• Google BigQuery
• Amazon EMR
• IBM Analytics Engine
• Apache Flink
• Lumify
• Presto
• Apache Pig

3.Why we use Apache Spark?

• 大规模分布式计算
• 免费
• 支持高级的数据分析需求
• 支持机器学习、流处理、SQL和图数据处理的模块
• 支持广泛的编程语言比如Python、R、Java
• 社区活跃

4.Installation of Apache Spark

4.1 Analysis various installation methods

首先，给出几乎所有可能的安装方式：

• Windows 10
• Windows Subsystem Linux (WSL)
• ubuntu / CentOS
• docker

其次，最佳的安装方式是docker。

最后，解释各种安装方式的利弊：

• Windows 10：不适合开发程序，因为不支持命令行工具、隐藏坑较多、解决方案的资料较少
• Windows Subsystem Linux (WSL)：需要安装较多软件和配置较多环境变量，非常麻烦
• ubuntu / CentOS：未尝试，但与WSL比较相似
• docker：简单、高效、可迁移

4.2 Tutorial of installing

安装过程，第一步：安装docker（略）。

安装过程，第二步，拉取镜像：

docker pull jupyter/pyspark-notebook

安装过程，第三步，创建容器：

docker run \
    -d \
    -p 8022:22 \
    -p 4040:4040 \
    -v /home/fyb:/data \
    -e GRANT_SUDO=yes \
    --name myspark \
    jupyter/pyspark-notebook

安装过程，第四步，配置docker容器的SSH登录：

docker exec \
    -u 0 \
    -it \
    myspark \
    /bin/bash

安装openssh-server软件

apt update && apt install openssh-server

设置允许root通过ssh登录

echo "PermitRootLogin yes" >> /etc/ssh/sshd_config

重启ssh服务

service ssh --full-restart

设置root用户的密码

passwd root

通过如下命令测试docker容器内的ssh是否设置成功

ssh root@127.0.0.1 -p 8022

安装过程，第五步，容器内的配置python环境：

以root用户登录SSH会话后，安装python依赖工具

apt install pip

安装PySpark依赖包

pip3 install pyspark numpy pandas tqdm

测试是否正确安装并执行了全部修改

python3 /usr/local/spark/examples/src/main/python/pi.py

4.3 Testing of installation

进入spark的根目录

cd /usr/local/spark

测试集群计算\(\pi\)的效果

python3 examples/src/main/python/pi.py

当控制台输出如下所示的信息，表示集群可用

Pi is roughly 3.130000

测试集群计算交替最小二乘法（Alternating Least Square, ALS）的效果。ALS是一种可以并行的矩阵分解算法，适用于大规模的协同过滤问题。

python3 examples/src/main/python/als.py

当控制台输出如下所示的信息，表示集群可用

Running ALS with M=100, U=500, F=10, iters=5, partitions=2

Iteration 0:                                                                            
RMSE: 0.2229

Iteration 1:
RMSE: 0.0731

Iteration 2:
RMSE: 0.0317

Iteration 3:
RMSE: 0.0315

Iteration 4:
RMSE: 0.0315

测试集群状态监控的效果

python3 examples/src/main/python/status_api_demo.py

当控制台输出如下所示的信息，表示集群可用

Job 0 status:  RUNNING
Stage 0: 10 tasks total (1 active, 0 complete)
Stage 1: 10 tasks total (0 active, 0 complete)
Job 0 status:  RUNNING
Stage 0: 10 tasks total (1 active, 0 complete)
Stage 1: 10 tasks total (0 active, 0 complete)
Job 0 status:  RUNNING
Stage 0: 10 tasks total (1 active, 0 complete)
Stage 1: 10 tasks total (0 active, 0 complete)
Job 0 status:  RUNNING
Stage 0: 10 tasks total (9 active, 0 complete)
Stage 1: 10 tasks total (0 active, 0 complete)
Job 0 status:  RUNNING
Stage 0: 10 tasks total (9 active, 0 complete)
Stage 1: 10 tasks total (0 active, 0 complete)
Job 0 status:  RUNNING
Stage 0: 10 tasks total (0 active, 10 complete)
Stage 1: 10 tasks total (0 active, 0 complete)
Job 0 status:  RUNNING
Stage 0: 10 tasks total (0 active, 10 complete)
Stage 1: 10 tasks total (0 active, 0 complete)
Job 0 status:  RUNNING
Stage 0: 10 tasks total (0 active, 10 complete)
Stage 1: 10 tasks total (9 active, 0 complete)
Job 0 status:  RUNNING
Stage 0: 10 tasks total (0 active, 10 complete)
Stage 1: 10 tasks total (9 active, 0 complete)
Job results are: [(0, 1), (1, 1), (2, 1), (3, 1), (4, 1), (5, 1), (6, 1), (7, 1), (8, 1), (9, 1)]

测试集群计算图的传递闭包性

python3 examples/src/main/python/transitive_closure.py

当控制台输出如下所示的信息，表示集群可用

TC has 6816 edges

5.Spark in action

Spark 程序框架，主要由几大模块构成，各个模块及其含义如下表所示。

5.1 DataFrames实战

任务1：PySpark数据处理

from pyspark.sql import SparkSession


# 步骤1：使用Python链接Spark环境
spark = SparkSession \
    .builder \
    .appName('pyspark') \
    .getOrCreate()

# 步骤2：创建dateframe数据
df = spark.createDataFrame(
    [
        ('001','1',100,87,67,83,98),
        ('002','2',87,81,90,83,83),
        ('003','3',86,91,83,89,63),
        ('004','2',65,87,94,73,88),
        ('005','1',76,62,89,81,98),
        ('006','3',84,82,85,73,99),
        ('007','3',56,76,63,72,87),
        ('008','1',55,62,46,78,71),
        ('009','2',63,72,87,98,64)],
    ['number','class','language','math','english','physic','chemical'])
df.show()

# 步骤3：用spark执行以下逻辑：找到数据行数、列数
print("the number of rows in this DataFrame: %d" % df.count())
print("the number of columns in this DataFrame: %d" % len(df.columns))

# 步骤4：用spark筛选class为1的样本
df.filter("class = 1").show(n=3)

# 步骤5：用spark筛选language >90 或 math> 90的样本
df.filter("language > 90 or math> 90").show(n=3)

# 步骤6：关闭spark会话
spark.stop()

任务2：PySpark数据统计

from pyspark import SparkFiles
from pyspark.sql import SparkSession


# 步骤0：使用Python链接Spark环境
spark = SparkSession \
    .builder \
    .appName('pyspark') \
    .getOrCreate()

# 步骤1：读取文件https://cdn.coggle.club/Pokemon.csv
spark.sparkContext.addFile("https://cdn.coggle.club/Pokemon.csv")
path = "file://"+SparkFiles.get("Pokemon.csv")

# 步骤2：将读取的进行保存，表头也需要保存
df = spark.read.csv(path=path, header=True, inferSchema= True)
df.show(n=3)
df = df.withColumnRenamed('Sp. Atk', 'Sp Atk')
df = df.withColumnRenamed('Sp. Def', 'Sp Def')

# 步骤3：分析每列的类型，取值个数
for column_name, column_type in df.dtypes:
    print((column_name, column_type))
    df.groupby(column_name).count().show(n=3)

# 步骤4：分析每列是否包含缺失值
for col in df.columns:
    number =  df.filter(df[col].isNull()).count()
    print("Name of column: %s \t ,Number of null values: %d" % (col, number))

# 步骤5：关闭spark会话
spark.stop()

Ref

[1] Apache Spark. https://spark.apache.org/