PySpark 简介 |使用 Apache Spark 进行分布式计算

数据集变得庞大。事实上，数据的增长速度超过了处理速度。因此，涉及大数据和高计算量的算法通常运行在分布式计算系统上。分布式计算系统涉及并行运行进程并进行通信（如果有必要）的节点（联网计算机）。

MapReduce – 用于分布式计算的编程模型称为 MapReduce。 MapReduce 模型涉及两个阶段，Map 和 Reduce。

映射——映射器处理输入数据的每一行（它是文件的形式），并产生键值对。
```
Input data → Mapper → list([key, value])
```
Reduce – reducer 处理键值对列表（在 Mapper函数之后）。它输出一组新的键值对。
```
list([key, value]) → Reducer → list([key, list(values)])
```

Spark – Spark（Apache 的开源大数据处理引擎）是一个集群计算系统。与其他集群计算系统（例如 Hadoop）相比，它更快。它在Python、Scala 和Java中提供高级 API。并行作业很容易在 Spark 中编写。我们将介绍 PySpark（Python + Apache Spark），因为这将使学习曲线更平坦。要在 linux 系统上安装 Spark，请按照此操作。要在多集群系统中运行 Spark，请按照此操作。我们将看到如何创建 RDD（Spark 的基本数据结构）。

RDD（弹性分布式数据集） ——RDD 是不可变的对象集合。由于我们使用的是 PySpark，因此这些对象可以是多种类型。这些将变得更加清晰。

SparkContext——为了在 Spark 中创建一个独立的应用程序，我们首先定义一个 SparkContext——

from pyspark import SparkConf, SparkContext
conf = SparkConf().setMaster("local").setAppName("Test")
# setMaster(local) - we are doing tasks on a single machine
sc = SparkContext(conf = conf)

RDD 转换——现在，创建了一个 SparkContext 对象。现在，我们将创建 RDD 并查看它们的一些转换。

# create an RDD called lines from ‘file_name.txt’
lines = sc.textFile("file_name.txt", 2)
  
# print lines.collect() prints the whole RDD
print lines.collect()

使用 Spark 的一个主要优点是它不会将数据集加载到内存中，lines 是指向“file_name.txt”文件的指针。

一个简单的 PySpark 应用程序，用于计算给定图的每个顶点的度数-

from pyspark import SparkConf, SparkContext
conf = SparkConf().setMaster("local").setAppName("Test")
# setMaster(local) - we are doing tasks on a single machine
sc = SparkContext(conf = conf)
def conv(line):
    line = line.split()
    return (int(line[0]), [int(line[1])])
def numNeighbours(x, y):
    return len(x) + len(y)
lines = sc.textFile('graph.txt')
edges = lines.map(lambda line: conv(line))
Adj_list = edges.reduceByKey(lambda x, y: numNeighbours(x, y))
print Adj_list.collect()

理解上面的代码——

我们的文本文件格式如下——（每条线代表有向图的一条边）
1 2
1 3
2 3
3 4
. .
. .
. .PySpark
大型数据集可能包含数百万个节点和边。
前几行设置了 SparkContext。我们从中创建一个 RDD行。
然后，我们将线RDD 转换为边RDD。函数conv 作用于每条线和形式为 (1, 2)、(1, 3)、(2, 3)、(3, 4)、... 的键值对存储在边RDD 中。
在此之后， reduceByKey聚合所有键 - 对应于特定键的对和numNeighbours函数用于在单独的 RDD Adj_list中生成每个顶点的度数，其形式为 (1, 2), (2, 1), (3, 1），……

运行代码——

上面的代码可以通过以下命令运行——

$ cd /home/arik/Downloads/spark-1.6.0/
$ ./bin/spark-submit degree.py

您可以在第一行使用您的 Spark 安装路径。

我们将看到更多关于如何使用 Spark 在机器集群中运行 MapReduce 任务，以及完成其他 MapReduce 任务。

参考资料–

http://lintool.github.io/SparkTutorial/
https://spark.apache.org/