并行计算入门-PySpark的使用

Spark应用程序作为独立的进程集运行，由驱动程序中的Spark context进行协调。

它可以自动创建(例如，如果您从shell中调用pyspark(然后将Spark上下文称为sc)。

但是我们还没有建立它，所以需要进行定义:

<code>from pyspark import SparkContext
sc = SparkContext('local[2]', 'Spark 101') 
/<code>

如果想要使用全部可用资源，也可以使用：

<code>sc = SparkContext('local[*]', 'Spark 101') 
/<code>

创建完之后，首先处理一个简单的摄氏华氏温度转换的并行计算

<code>temp_c = [10, 3, -5, 25, 1, 9, 29, -10, 5]

rdd_temp_c = sc.parallelize(temp_c)
rdd_temp_K = rdd_temp_c.map(lambda x: x + 273.15).collect()

print(rdd_temp_K)  
/<code>

结果

<code>[283.15, 276.15, 268.15, 298.15, 274.15, 282.15, 302.15, 263.15, 278.15]
/<code>

通过使用sc.parallelize来并行处理待计算的摄氏温度，然后使用map来进行一一映射。

除了map这种映射外，常用还有reduce函数，例如

<code>numbers = [1, 4, 6,2, 9, 10]

rdd_numbers=sc.parallelize(numbers)

# Use reduce to combine numbers
rdd_reduce = rdd_numbers.reduce(lambda x,y: "(" + str(x) + ", " + str(y) + "]")
#rdd_reduce = rdd_numbers.reduce(lambda x,y: "(" + str(x) + str(y))


print(rdd_reduce)
/<code>

结果

<code>(((1, (4, 6]], 2], (9, 10]]
/<code>

通常我们使用PySpark是为了部署机器学习项目，在PySpark中也有多样的数据处理手段。

首先我们创建一个会话(Session)

<code>from pyspark.sql import SparkSession

session = SparkSession.builder.appName('data_processing').getOrCreate()
/<code>

读入数据集

<code>df = session.read.csv('iris.csv',inferSchema=True,header=True)
df.show()
/<code>

<code>df.printSchema()
/<code>

结果

<code>root
 |-- sepal_length: double (nullable = true)
 |-- sepal_width: double (nullable = true)
 |-- petal_length: double (nullable = true) 

 |-- petal_width: double (nullable = true)
 |-- species: string (nullable = true)
/<code>

也可以选择指定列输出

<code>df.select('sepal_width','petal_width').show(3)
/<code>

结果

<code>+-----------+-----------+
|sepal_width|petal_width|
+-----------+-----------+
|        3.5|        0.2|
|        3.0|        0.2|
|        3.2|        0.2|
+-----------+-----------+
only showing top 3 rows
/<code>

创建新列

<code>df.withColumn("new_col_name",(df["sepal_width"]*10)).show(6)
/<code>

筛选特征

<code>df.filter(df["species"]=="setosa").show(9)
df.filter((df["species"]=="setosa") | (df["sepal_width"]<3.0)).show(9)
/<code>

查看列的不同值

<code>df.select("species").distinct().show()
/<code>

按列统计

<code>df.groupBy('petal_length').count().show()
/<code>

按列统计排序

<code>df.groupBy('petal_length').count().orderBy("count",ascending=False).show()
/<code>

按列聚合

<code>df.groupBy('species').agg({'sepal_length':'sum'}).show()
/<code>

以上只是简单的数据处理部分，PySpark还有很多其他吸引人的特性，将在后续文章中继续介绍。

閱讀更多 leo的學習之旅 的文章

關鍵字: 并行计算 摄氏温度 使用