Effective PySpark(PySpark 常见问题)

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

另外,在使用UDF函数的时候,发现列是NoneType 或者null,那么有两种可能:

在PySpark里,有时候会发现udf函数返回的值总为null,可能的原因有:

  1. 忘了写return
def abc(c):
    "yes"
  1. 返回的类型不匹配。

比如你明明是一个FloatType,但是你定义的时候说是一个ArrayType,这个时候似乎不会报错,而是udf函数执行会是null. 这个问题之前在处理二进制字段时遇到了。我们理所当然的认为二进制应该是类型 ArrayType(Byte(),True) ,但实际上是BinaryType.

dataframe.show 问题

详细问题可参看: https://stackoverflow.com/questions/39662384/pyspark-unicodeencodeerror-ascii-codec-cant-encode-character

主要是python方面的问题。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏hotqin888的专栏

engineercms利用pdf.js制作连续看图功能

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/hotqin888/article/det...

19010
来自专栏Java帮帮-微信公众号-技术文章全总结

Java多线程详解5【面试+工作】

Java多线程详解【面试+工作】 Java线程:新特征-信号量 Java的信号量实际上是一个功能完毕的计数器,对控制一定资源的消费与回收有着很重要的意义,信号量...

426100
来自专栏青青天空树

趣味题:恺撒Caesar密码(c++实现)

描述:Julius Caesar 生活在充满危险和阴谋的年代。为了生存,他首次发明了密码,用于军队的消息传递。假设你是Caesar 军团中的一名军官,需要把Ca...

8520
来自专栏移动端开发

Android学习--跨程序共享数据之内容提供其探究

      跨程序共享数据之内容提供器,这是个什么功能?看到这个名称的时候最能给我们提供信息的应该是“跨程序”这个词了,是的重点就是这个词,这个内容提供器的作用...

13430
来自专栏calvin

【nodejs】让nodejs像后端mvc框架(asp.net mvc)一orm篇【如EF般丝滑】typeorm介绍(8/8)

在使用nodejs开发过程中,刚好碰到需要做一个小工具,需要用到数据库存储功能。而我又比较懒,一个小功能不想搞一个nodejs项目,又搞一个后端项目。不如直接在...

38220
来自专栏博客园

讲一下Asp.net core MVC2.1 里面的 ApiControllerAttribute

转自:https://www.cnblogs.com/sheldon-lou/p/9495377.html

12220
来自专栏MasiMaro 的技术博文

hook键盘驱动中的分发函数实现键盘输入数据的拦截

我自己在看《寒江独钓》这本书的时候,书中除了给出了利用过滤的方式来拦截键盘数据之外,也提到了另外一种方法,就是hook键盘分发函数,将它替换成我们自己的,然后再...

11020
来自专栏潇涧技术专栏

Pury Project Analysis

Pury的源码:https://github.com/NikitaKozlov/Pury

9520
来自专栏haifeiWu与他朋友们的专栏

golang重构博客统计服务

作为一个后端开发,在docker,etcd,k8s等新技术不断涌现的今天,其背后的功臣golang在语言排行榜上持续走高,因此楼主也就开了这次使用golang自...

14120
来自专栏Spark学习技巧

Spark与mongodb整合完整版本

一,准备阶段 MongoDB Connector for spark是的spark操作mongodb数据很简单,这样方便使用spark去分析mongodb数据,...

2.8K100

扫码关注云+社区

领取腾讯云代金券