炼数成金 门户 大数据 Spark 查看内容

用Python语言写Spark

2018-9-4 14:31| 发布者: 炼数成金_小数| 查看: 10758| 评论: 0|原作者: 祝威廉|来自: Spark技术日报

摘要: Spark 是目前大数据处理的事实标准。PySpark能让你使用Python语言来写Spark程序。我们先做一个最简单的字符数统计程序。这样我们就知道一个PySpark程序是什么样子,以及如何运转起来。在很多情况下,单机模式足够我 ...

Python SQL Hadoop Spark 分布式 函数

001、PySpark 基础
Spark 是目前大数据处理的事实标准。PySpark能让你使用Python语言来写Spark程序。

我们先做一个最简单的字符数统计程序。这样我们就知道一个PySpark程序是什么样子,以及如何运转起来。

我们准备一个文件a.csv。里面的内容如下:
a b c,1.0a b,2.0c,3.0d,4.0
然后我们打开编辑器,比如我这里是Intellij IDEA。 新建一个myfirstpyspark.py文件。

PySpark需要你有一个SparkSession对象,这是一切的开始,在这里你可以做一些配置。

from pyspark.SQL import SparkSession
session = SparkSession.builder.master("local[*]").appName("test").getOrCreate()
其中"local[*]"表示我们执行的是local模式,也就是单机模式,*表示使用所有的CPU核心,如果你写成local[2]那么表示单机模式,并且使用两个核。

在很多情况下,单机模式足够我们使用,因为他已经是多线程并行运行了,这比我们自己完成一个多线程的程序来的简单,而且,Spark可以让你的代码看起来就像单机,他自动完成分布式工作。

如果愿意,还可以运行在Standalone,Yarn,Mesos等模式下,它们都是真正的分布式模式。Spark是一个典型的Master-Slave结构。Master负责解释你写的代码,Slave则负责执行你的代码。

构建好了session之后,现在可以去读文件了。

from pyspark.sql.types import *import pyspark.sql.functions as f

df = session.read.csv(        "a.csv", 
        encoding="utf-8",
        header=False, 
        schema=StructType(
            [StructField("text", StringType()),
             StructField("index", StringType())]))
这里我使用了一个较为复杂的方式构建Dataframe, Dataframe你可以简单理解为SQL的编程表达形式。不使用常规教程提及的RDD API的原因是因为Spark花了大力气将大部分东西都迁移到Dataframe上,我们也就不要倒行逆施了。实际场景中,RDD非常灵活,但是往往导致代码难以管理和维护。Dataframe API 比较受限,但是更"SQL"化,更"结构化",大家写出来的会是基本一致的,并且性能更好。

session.read 是一个datasource API,这个时候你获得了一个“读”对象。聪明的读者可能会思索是不是还有session.write, 实际上是有的,session.write 可以让你获得了一个写的对象。

除了读取CSV文件,你还可以读取parquet, json, elasticsearch,mysql 等其他存储器。

让我们回转过来,读取一个csv文件的方式如同上面的示例代码,具体参数我会简单说明下。

header 指的是是否有头部。CSV文件通常第一行是列名。我们这里设置为False,因为构建的文件并没有这个需求。

schema 可以指定CSV每列的名字,类型。大家只要记住是这么写就OK了。这种方式非常有用。对于非结构化数据,你可以简单认为他是只有一列的结构化数据。

现在,我们已经有了一张名字叫'df'的表,这张表有两个字段,text 和 index。 现在想统计text里每个词汇出现的次数。这在pyspark里只需要一行代码就可以搞定。

df.select(f.explode(f.split("text", " ")).alias("word")).\
    groupBy("word").\
    count().\
    show()
最后运行的结果如下:


通过导入import pyspark.sql.functions as f 我们可以使用Spark内置的UDF函数并且使用f来进行引用,额外的收获是你可以获得更友好的代码提示。

select 你可以理解为SQL中select语法的一种编程形态。 先对text字段按空格切分,然后通过expode展开为多行,并且宠幸把行命名为word。对应的SQL为:

select explode(split("text"," ")) as word from ....
接着按word字段进行groupBy,然后count。count其实是一个简写形式,我们还可以这么写:

df.select(f.explode(f.split("text", " ")).alias("word")).\
    groupBy("word").\
    agg(f.count("word").alias("wordCount")).\
    show()
agg 里面会填写一个聚合函数,配合groupBy使用,这和我们使用SQL的方式是一致的。现在增加个额外需求,我想排个序,应该怎么弄呢?也不是难事:

df.select(f.explode(f.split("text", " ")).alias("word")).\
    groupBy("word").\
    agg(f.count("word").alias("wordCount")).\
    orderBy(f.desc("wordCount")).\
    show()
如果觉得官方提供的UDF函数满足不了需求,我们也是可以自定义函数的。 定义一个普通的python函数:
def my_split(col):
    return col.split(" ")

把my_split封装为一个UDF函数:
udf_my_split = f.udf(my_split, ArrayType(StringType()))
ArrayType(StringType()) 指的是返回类型。接着就是使用它:

df.select(f.explode(udf_my_split("text")).alias("word")). \
    groupBy("word"). \
    agg(f.count("word").alias("wordCount")). \
    orderBy(f.desc("wordCount")). \
    show()
现在我们看看如何运行这个程序。截止到现在,你的代码看起来应该是这个样子的。

# -*- coding: UTF-8 -*-import osfrom pyspark.sql import SparkSessionfrom pyspark.sql.types import *import pyspark.sql.functions as f

spark = SparkSession.builder.master("local[*]").appName("test").getOrCreate()

df = spark.read.csv(    "YOUR-PATH/a.csv",
    encoding="utf-8",
    header=False,
    schema=StructType(
        [StructField("text", StringType()),
         StructField("index", StringType())]))def my_split(col):
    return col.split(" ")

udf_my_split = f.udf(my_split, ArrayType(StringType()))

df.select(f.explode(udf_my_split("text")).alias("word")). \
    groupBy("word"). \
    agg(f.count("word").alias("wordCount")). \
    orderBy(f.desc("wordCount")). \
    show()

我们现在需要运行这段程序,具体方式如下:
export PYTHONIOENCODING=utf8;./bin/spark-submit \--driver-memory 2g \--master "local[*]" [YOUR-PATH]/my_first_pyspark.py
截止目前为止,我们已经能够用PySpark开发应用了,并且了解了里面的一些高级用法,比如自定义UDF函数的使用。

欢迎加入本站公开兴趣群
软件开发技术群
兴趣范围包括:Java,C/C++,Python,PHP,Ruby,shell等各种语言开发经验交流,各种框架使用,外包项目机会,学习、培训、跳槽等交流
QQ群:26931708

Hadoop源代码研究群
兴趣范围包括:Hadoop源代码解读,改进,优化,分布式系统场景定制,与Hadoop有关的各种开源项目,总之就是玩转Hadoop
QQ群:288410967 

鲜花

握手

雷人

路过

鸡蛋

相关阅读

最新评论

热门频道

  • 大数据
  • 商业智能
  • 量化投资
  • 科学探索
  • 创业

即将开课

 

GMT+8, 2018-9-21 00:05 , Processed in 0.805883 second(s), 25 queries .