PySpark大数据分析(3)：使用Python操作RDD_for element in rdd.collect(),在循环内遍历rdd

使用Python编写Spark代码

在Python中使用Spark，首先需要导入PySpark，然后创建一个SparkConf对象配置你的应用，再基于这个SparkConf创建一个SparkContext。以创建一个名为’MyApp’的本地应用为例：

from pyspark import SparkConf, SparkContext


conf = SparkConf().setMaster('local').setAppName('MyApp')
sc = SparkContext(conf=conf)
1
2
3
4
5

也可以通过SparkSession实现之前命令行中的PySpark编程，比如找出’Sepal Width’大于3的数据：

from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession


conf = SparkConf().setMaster('local').setAppName('MyApp')
sc = SparkContext(conf=conf)
spark = SparkSession(sc)

iris = spark.read.csv('/Users/data/iris.csv', header=True)
iris.filter(iris['Sepal Width'] > 3.0).show(5)
1
2
3
4
5
6
7
8
9
10

展示结果为：

+------------+-----------+------------+-----------+-----------+
|Sepal Length|Sepal Width|Petal Length|Petal Width|      Label|
+------------+-----------+------------+-----------+-----------+
|         5.1|        3.5|         1.4|        0.2|Iris-setosa|
|         4.7|        3.2|         1.3|        0.2|Iris-setosa|
|         4.6|        3.1|         1.5|        0.2|Iris-setosa|
|         5.0|        3.6|         1.4|        0.2|Iris-setosa|
|         5.4|        3.9|         1.7|        0.4|Iris-setosa|
+------------+-----------+------------+-----------+-----------+
only showing top 5 rows
1
2
3
4
5
6
7
8
9
10

通过SparkSession读取的数据是以DataFrame的形式存储的，如果想使用RDD需要进行转换：

print(type(iris))
# <class 'pyspark.sql.dataframe.DataFrame'>
rdd = iris.rdd
print(rdd.first())
# Row(Sepal Length='5.1', Sepal Width='3.5', Petal Length='1.4', Petal Width='0.2', Label='Iris-setosa')
1
2
3
4
5

向Spark传递函数

像Python中的filter()函数一样，我们可以将一个自定义的函数传递给RDD的filter()，从而让它按照我们定义的方式进行过滤。当函数比较简单时，可以使用匿名函数lambda传递；

from pyspark import</