构建PySpark环境
首先确保安装了python 2.7 ,强烈建议你使用Virtualenv方便python环境的管理。之后通过pip 安装pyspark
pip install pyspark
文件比较大，大约180多M,有点耐心。
下载 spark 2.2.0,然后解压到特定目录，设置SPARK_HOME即可。
其实如果通过spark-submit 提交程序，并不会需要额外安装pyspark, 这里通过pip安装的主要目的是为了让你的IDE能有代码提示。
PySpark worker启动机制
PySpark的工作原理是通过Spark里的PythonRDD启动一个（或者多个，以pythonExec, 和envVars为key）Python deamon进程，然后一旦有task过来了，就通过python deamon进程fork一个新的python worker。 python worker是可以复用的，并不会用完就立马销毁。一个task过来的流程为， 看看worker里有清闲的么，如果有，就直接返回。没有就fork一个新的worker.
PySpark 如何实现某个worker 里的变量单例
从前面PySpark worker启动机制里，我们可以看到，一个Python worker是可以反复执行任务的。在NLP任务中，我们经常要加载非常多的字典，我们希望字典只会加载一次。这个时候就需要做些额外处理了。做法如下：
class DictLoader(object):
clf = None
def __init__(self, baseDir, archive_auto_extract, zipResources):
if not DictLoader.is_loaded():
DictLoader.load_dic(baseDir) @staticmethod
def load_dic(baseDir):
globPath = baseDir + "/dic/*.dic"
dicts = glob.glob(globPath) for dictFile in dicts:
temp = dictFile if os.path.exists(dictFile) else SparkFiles.get(dictFile)
jieba.load_userdict(temp)
jieba.cut("nice to meet you")
DictLoader.clf = "SUCCESS" @staticmethod
def is_loaded():
return DictLoader.clf is not None定义一个cls对象，并且使用staicmethod annotation,这样就可以模拟类似Java的静态方法了。之后你可以随心所欲的loader = DictLoader ()
如何加载资源文件
在NLP处理了，字典是少不了，前面我们避免了一个worker多次加载字典，现在还有一个问题，就是程序如何加载字典。通常我们希望能够把字典打成一个zip包，代码也打成一个zip包，然后通过下面的命令进行提交：
./bin/spark-submit \ --py-files dist/jobs.zip \ --files dist/dics.zip \ --master "local[*]" python/src/batch.py
自己开发的模块可以打包成jobs.zip,对应的spark任务单独成一个batch.py文件，然后字典打包成dics.zip.
那么程序中如何读取dics.zip里的文件呢？ 在Spark standalone 和 local模式下，dics.zip在各个worker的工作目录里并不会被解压，所以需要额外处理下：
def __init__(self, baseDir, archive_auto_extract, zipResources):
if not DictLoader.is_loaded():
for zr in zipResources: if not archive_auto_extract: with zipfile.ZipFile(SparkFiles.getRootDirectory() + '/' + zr, 'r') as f:
f.extractall(".")
DictLoader(baseDir)archive_auto_extract 判定是不是会自动解压（yarn模式下回自动解压），判断的方法为：
archive_auto_extract = spark.conf.get("spark.master").lower().startswith("yarn")zipResources 则是所有需要解压的zip包的名字，对应获取的方法为：
zipfiles = [f.split("/")[-1] for f in spark.conf.get("spark.files").split(",") if f.endswith(".zip")]对应的zipfiles所在的目录你可以这样拼接：
SparkFiles.getRootDirectory() + '/' + zfilename
所以如果你不是运行在yarn模式的情况下，你需要先解压，然后进行加载。获取路径的方式建议如下：
temp = dictFile if os.path.exists(dictFile) else SparkFiles.get(dictFile)
这样可以兼容IDE里运行，local/standalone/yarn 模式运行。
前面的jobs.zip文件里面全部是python文件，并不需要压缩就可以直接读到。
主动定义schema,避免spark auto inference schema
我之前写过这么一段代码：
oldr = df.rdd.map( lambda row: Row(ids=row['ids'], mainId=row["mainId"].item(), tags=row["tags"]))
然后我需要把oldr 变回为rdd,这个时候我这么用：
resultDf = spark.createDataFrame(oldr)
resultDf.mode("overwrite").format(...).save(...这会导致oldr被执行两次，一次是为了做schema推测，一次是为了做实际的计算。
我们可以这么写：
from pyspark.sql.types import StructType, IntegerType, ArrayType, StructField, StringType, MapType
fields = [StructField("ids", ArrayType(IntegerType())), StructField("mainId", IntegerType()),
StructField("tags", MapType(StringType(), IntegerType()))]
resultDf = spark.createDataFrame(resultRdd, StructType(fields=fields)这样显示的为rdd定义schema,就可以避免额外的推测了。
lambda 和 函数的选择
lambda可以定义匿名函数，但是表现力有限：
.map( lambda row: Row(ids=row['ids'], mainId=row["mainId"].item(), tags=row["tags"]))
我们也可以定义函数：
def create_new_row(row): return Row(ids=row['ids'], mainId=row["mainId"].item(), tags=row["tags"])
然后直接使用：
.map(create_new_row).....
如何定义udf函数/如何避免使用Python UDF函数
先定义一个常规的python函数：
# 自定义split函数def split_sentence(s):
return s.split(" ")转化为udf函数并且使用。
from pyspark.sql.functions import udffrom pyspark.sql.types import *
ss = udf(split_sentence, ArrayType(StringType()))
documentDF.select(ss("text").alias("text_array")).show()唯一麻烦的是，定义好udf函数时，你需要指定返回值的类型。
使用Python 的udf函数，显然效率是会受到损伤的，我们建议使用标准库的函数，具体这么用：
from pyspark.sql import functions as f
documentDF.select(f.split("text", "\\s+").alias("text_array")).show()pyspark.sql. functions 引用的都是spark的实现，所以效率会更高。
另外，在使用UDF函数的时候，发现列是NoneType 或者null,那么有两种可能：
在PySpark里，有时候会发现udf函数返回的值总为null,可能的原因有：
忘了写return
def abc(c): "yes"
返回的类型不匹配。
比如你明明是一个FloatType,但是你定义的时候说是一个ArrayType,这个时候似乎不会报错，而是udf函数执行会是null.
这个问题之前在处理二进制字段时遇到了。我们理所当然的认为二进制应该是类型 ArrayType(Byte(),True) ，但实际上是BinaryType.
作者：祝威廉
链接：https://www.jianshu.com/p/0045d144850e