前言
PySpark是Spark 實(shí)現(xiàn) Unify BigData && Machine Learning目標(biāo)的基石之一芍耘。通過(guò)PySpark,我們可以用Python在一個(gè)腳本里完成數(shù)據(jù)加載立莉，處理茉帅，訓(xùn)練，預(yù)測(cè)等完整Pipeline,加上DB良好的notebook的支持捍歪，數(shù)據(jù)科學(xué)家們會(huì)覺(jué)得非常開心负溪。當(dāng)然缺點(diǎn)也是有的，就是帶來(lái)了比較大的性能損耗妆棒。
性能損耗點(diǎn)分析
如果使用PySpark,大概處理流程是這樣的(注意，這些都是對(duì)用戶透明的)
- python通過(guò)socket調(diào)用Spark API(py4j完成)沸伏，一些計(jì)算邏輯糕珊，python會(huì)在調(diào)用時(shí)將其序列化，一并發(fā)送給Spark毅糟。
- Spark 觸發(fā)計(jì)算红选，比如加載數(shù)據(jù)，然后把數(shù)據(jù)轉(zhuǎn)成內(nèi)部存儲(chǔ)格式InternalRow,接著啟動(dòng)Python Deamon, Python Deamon再啟動(dòng)多個(gè)Worker,
- 數(shù)據(jù)通過(guò)socket協(xié)議發(fā)送給Python Worker（不跨網(wǎng)絡(luò)）姆另，期間需要將InternalRow轉(zhuǎn)化為Java對(duì)象喇肋，然后再用Java Pickle進(jìn)行序列化(一次)，這個(gè)時(shí)候才能通過(guò)網(wǎng)絡(luò)發(fā)送給Worker
- Worker接收后迹辐，一條一條反序列化(python pickle蝶防，兩次),然后轉(zhuǎn)化為Python對(duì)象進(jìn)行處理。拿到前面序列化好的函數(shù)反序列化明吩，接著用這個(gè)函數(shù)對(duì)這些數(shù)據(jù)處理间学，處理完成后，再用pickle進(jìn)行序列化（三次）印荔，發(fā)送給Java Executor.
- Java Executor獲取數(shù)據(jù)后低葫，需要反序列化（四次），然后轉(zhuǎn)化為InternalRow繼續(xù)進(jìn)行處理躏鱼。
所以可以看到氮采，前后需要四次編碼/解碼動(dòng)作殷绍。序列化反序列化耗時(shí)應(yīng)該占用額外耗時(shí)的70%左右染苛。我們說(shuō)，有的時(shí)候把序列化框架設(shè)置為Kyro之后，速度明顯快了很多茶行，可見序列化的額外耗時(shí)是非常明顯的躯概。
前面是一個(gè)點(diǎn)，第二個(gè)點(diǎn)是畔师，數(shù)據(jù)是按行進(jìn)行處理的娶靡，一條一條，顯然性能不好看锉。
第三個(gè)點(diǎn)是姿锭，Socket協(xié)議通訊其實(shí)還是很快的，而且不跨網(wǎng)絡(luò)伯铣，只要能克服前面兩個(gè)問(wèn)題呻此，那么性能就會(huì)得到很大的提升。 另外可以跟大家說(shuō)的是腔寡，Python如果使用一些C庫(kù)的擴(kuò)展焚鲜，比如Numpy,本身也是非常快的放前。
如何開啟Arrow進(jìn)行加速忿磅，以及背后原理
開啟方式很簡(jiǎn)單，啟動(dòng)時(shí)加上一個(gè)配置即可：
if __name__ == '__main__':
conf = SparkConf()
conf.set("spark.sql.execution.arrow.enabled", "true")
你也可以在submit命令行里添加凭语。
那么Arrow是如何加快速度的呢葱她？主要是有兩點(diǎn)：
- 序列化友好
- 向量化
序列化友好指的是，Arrow提供了一個(gè)內(nèi)存格式似扔，該格式本身是跨應(yīng)用的览效，無(wú)論你放到哪，都是這個(gè)格式虫几，中間如果需要網(wǎng)絡(luò)傳輸這個(gè)格式锤灿，那么也是序列化友好的，只要做下格式調(diào)整（不是序列化）就可以將數(shù)據(jù)發(fā)送到另外一個(gè)應(yīng)用里辆脸。這樣就大大的降低了序列化開銷但校。
向量化指的是，首先Arrow是將數(shù)據(jù)按block進(jìn)行傳輸?shù)姆惹猓浯问强梢詫?duì)立面的數(shù)據(jù)按列進(jìn)行處理的状囱。這樣就極大的加快了處理速度。
實(shí)測(cè)效果
為了方便測(cè)試倘是，我定義了一個(gè)基類：
from pyspark import SQLContext
from pyspark import SparkConf
from pyspark import SparkContext
from pyspark.sql import SparkSession
import os
os.environ["PYSPARK_PYTHON"] = "/Users/allwefantasy/deepavlovpy3/bin/python3"
class _SparkBase(object):
@classmethod
def start(cls, conf=SparkConf()):
cls.sc = SparkContext(master='local[*]', appName=cls.__name__, conf=conf)
cls.sql = SQLContext(cls.sc)
cls.session = SparkSession.builder.getOrCreate()
cls.dataDir = "/Users/allwefantasy/CSDNWorkSpace/spark-deep-learning_latest"
@classmethod
def shutdown(cls):
cls.session.stop()
cls.session = None
cls.sc.stop()
cls.sc = None
接著提供了一個(gè)性能測(cè)試輔助類：
import time
from functools import wraps
import logging
logger = logging.getLogger(__name__)
PROF_DATA = {}
def profile(fn):
@wraps(fn)
def with_profiling(*args, **kwargs):
start_time = time.time()
ret = fn(*args, **kwargs)
elapsed_time = time.time() - start_time
if fn.__name__ not in PROF_DATA:
PROF_DATA[fn.__name__] = [0, []]
PROF_DATA[fn.__name__][0] += 1
PROF_DATA[fn.__name__][1].append(elapsed_time)
return ret
return with_profiling
def print_prof_data(clear):
for fname, data in PROF_DATA.items():
max_time = max(data[1])
avg_time = sum(data[1]) / len(data[1])
logger.warn("Function %s called %d times. " % (fname, data[0]))
logger.warn('Execution time max: %.3f, average: %.3f' % (max_time, avg_time))
if clear:
clear_prof_data()
def clear_prof_data():
global PROF_DATA
PROF_DATA = {}
很簡(jiǎn)單亭枷，就是wrap一下實(shí)際的函數(shù)，然后進(jìn)行時(shí)間計(jì)算〔笳福現(xiàn)在叨粘，我們寫一個(gè)PySpark的類：
import logging
from random import Random
import pyspark.sql.functions as F
from pyspark import SparkConf
from pyspark.sql.types import *
from example.allwefantasy.base.spark_base import _SparkBase
import example.allwefantasy.time_profile as TimeProfile
import pandas as pd
logger = logging.getLogger(__name__)
class PySparkOptimize(_SparkBase):
def trick1(self):
pass
if __name__ == '__main__':
conf = SparkConf()
conf.set("spark.sql.execution.arrow.enabled", "true")
PySparkOptimize.start(conf=conf)
PySparkOptimize().trick1()
PySparkOptimize.shutdown()
這樣骨架就搭建好了旨别。
我們寫第一個(gè)方法撮奏，trick1,做一個(gè)簡(jiǎn)單的計(jì)數(shù)：
def trick1(self):
df = self.session.range(0, 1000000).select("id", F.rand(seed=10).alias("uniform"),
F.randn(seed=27).alias("normal"))
# 更少的內(nèi)存和更快的速度
TimeProfile.profile(lambda: df.toPandas())()
TimeProfile.print_prof_data(clear=True)
并且將前面的arrow設(shè)置為false.結(jié)果如下：
Function <lambda> called 1 times.
Execution time max: 6.716, average: 6.716
然后同樣的代碼续扔，我們把a(bǔ)rrow設(shè)置為true,是不是會(huì)好一些呢?
Function <lambda> called 1 times.
Execution time max: 2.067, average: 2.067
當(dāng)然我這個(gè)測(cè)試并不嚴(yán)謹(jǐn)疮丛，但是對(duì)于這種非常簡(jiǎn)單的示例，提升還是有效三倍的驴党，不是么瘪撇？而這，只是改個(gè)配置就可以達(dá)成了港庄。
分組聚合使用Pandas處理
另外值得一提的是倔既，PySpark是不支持自定義聚合函數(shù)的，現(xiàn)在如果是數(shù)據(jù)處理鹏氧，可以把group by的小集合發(fā)給pandas處理叉存，pandas再返回，比如
def trick7(self):
df = self.session.createDataFrame(
[(1, 1.0), (1, 2.0), (2, 3.0), (2, 5.0), (2, 10.0)], ("id", "v"))
@F.pandas_udf("id long", F.PandasUDFType.GROUPED_MAP)
def normalize(pdf):
v = pdf.v
return pdf.assign(v=(v - v.mean()) / v.std())[["id"]]
df.groupby("id").apply(normalize).show()
這里是id進(jìn)行g(shù)ourp by 度帮，這樣就得到一張id列都是1的小表歼捏，接著呢把這個(gè)小表轉(zhuǎn)化為pandas dataframe處理，處理完成后笨篷，還是返回一張小表瞳秽，表結(jié)構(gòu)則在注解里定義，比如只返回id字段率翅，id字段是long類型练俐。
