这两天写pyspark的一些总结

2,708 阅读4分钟
原文链接: www.jianshu.com

构建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 udf
from 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的实现,所以效率会更高。