pyspark spark 快速入门 懒人版本

pyspark spark 快速入门 懒人版本

安装

docker 安装方式

最简单的是直接docker，有一下几个比较快速的安装方式参考：

• https://github.com/actionml/docker-spark
• https://github.com/wongnai/docker-spark-standalone
• https://github.com/epahomov/docker-spark
• https://towardsdatascience.com/a-journey-into-big-data-with-apache-spark-part-1-5dfcc2bccdd2

如果是进一步的简化，那就选择安装spark standalone version， 不依赖hadoop

安装完成后，就可以启动 sbin的脚本

    ./start-all.sh

其web的界面效果（这里有两个worker）：

注意:
由于上述的docker用的是基础镜像 openjdk:8-alpine，比较适合scala的环境，pyspark 需要python3 ，所以要在Dockerfile，另外处理python3 的安装

直接官网下载bin

https://spark.apache.org/downloads.html

运行例子

最简单直接在安装的目标主机上，运行,(否则把localhost替换成spark host)

    # Run a Python application on a Spark standalone cluster
    ./bin/spark-submit 
    --master spark://localhost:7077 
    examples/src/main/python/pi.py 
    1000

观察CPU，利用了多个核

pyspark

    ./bin/pyspark    

运行pyspark的wordcount (helloworld)

    >>> p='/usr/local/spark/README.md'
    >>> text_file = sc.textFile(p)
    >>> counts = text_file.flatMap(lambda line: line.split(" ")).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b)
    >>> counts
    PythonRDD[6] at RDD at PythonRDD.scala:53
    >>> counts.collect()
    [Stage 0:>                                                          (0 + 2) / 2]/usr/local/spark/python/lib/pyspark.zip/pyspark/shuffle.py:60: UserWarning: Please install psutil to have better support with spilling
    /usr/local/spark/python/lib/pyspark.zip/pyspark/shuffle.py:60: UserWarning: Please install psutil to have better support with spilling
    [('#', 1), ('Apache', 1), ('Spark', 15), ('', 73), ('is', 7), ('unified', 1), ('analytics', 1), ('engine', 2), ('It', 2), ('provides', 1), ('high-level', 1), ('APIs', 1), ('in', 6), ('Scala,', 1), ('Java,', 1), ('an', 4), ('optimized', 1), ('supports', 2), ('computation', 1), ('analysis.', 1), ('set', 2), ('of', 5), ('tools', 1), ('SQL', 2 
    ...

注意点：

• counts 的数据类型是RDD
• counts = text_file.flatMap(lambda line: line.split(" ")).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b) 的时候没有真正跑， count.collect()的时候才真正运行任务。
  这是spark特色：需要真正用到变量的时候才运行
• sc 是启动pyspark 后自动初始化的变量

RDD (resilient distributed dataset )

RDD 是提供一直操作collection里面每个元素的方法，该方法可以在集群的各个node 并行运行。

例子教程：

• https://spark.apache.org/docs/latest/rdd-programming-guide.html

注意：看起来这个RDD用处像是dataframe （除了跨node 并行计算），他是spark dataframe ，但实际语法与pandas的dataframe是不一样的。

这就带来一个问题，我需要重新熟悉学习他的用法。
为了解决这个问题，databricks社区有一个方案 Koalas

用了这个之后，就可以转化到类普通pandas 的熟悉的dataframe处理

    kdf = ks.from_pandas(pdf) #pandas的dataframe -> koalas
    kdf = sdf.to_koalas() # spark dataframe ->koalas的dataframe

参考：

• https://www.iteblog.com/archives/2549.html
• https://koalas.readthedocs.io/en/latest/getting_started/10min.html

提交任务

上面最简单的例子是提交一个python 文件到spark ， 但实际我们需要多个files ，并且有依赖，事实上这里比较多坑

• 最简单的依赖例子1 - 无第三方依赖的case：

    [klg@ira-r740 wade-test]$ ls
    application   requirements.txt  t.py  
    [klg@ira-r740 wade-test]$ tree application
    application
    └── myfile.py

    0 directories, 1 file
    

t.py

from pyspark.sql import SparkSession
SPARK_MASTER_HOST = 'ira-r740'
SPARK_MASTER_PORT = 7077
def test(x):
    from application.myfile import simple_function
    return simple_function(x)

if __name__ == "__main__":
    spark = SparkSession
        .builder
        .config('spark.master', 'spark://{spark_master_host}:{spark_master_port}' 
        .format(spark_master_host=SPARK_MASTER_HOST, spark_master_port=SPARK_MASTER_PORT)) 
        .appName("PythonPiwade")
        .getOrCreate()

    # test by mapping function over RDD
    rdd = spark.sparkContext.parallelize(range(1,100))
    result = rdd.map(lambda x: test(x)).collect()
    print("result", result)
    spark.stop()

先压缩，后提交

    $cd wade-test
    $zip -rq application.zip 
    $./bin/spark-submit   --py-files /home/klg/pyspark/wade-test/application.zip  /home/klg/pyspark/wade-test/t.py

• 最简单的依赖例子 - 有第三方依赖的case

t.py

...   
if __name__ == "__main__":
    ...
    rdd = spark.sparkContext.parallelize(range(1,100))
    result = rdd.map(lambda x: test(x)).collect()
    print("result", result)
    import bs4 ### 变动在这里
    print(bs4.__version__) ### 变动在这里
    spark.stop()

[klg@ira-r740 wade-test]$ ls
application application.zip requirements.txt t.py venv venv.zip
先压缩，后提交

    $cd wade-test
    $python3 -m venv venv # 建立虚拟环境
    $source venv/bin/activate
    $pip install -r requirements.txt
    $zip -rq venv.zip venv # 打包依赖
    $export PYSPARK_PYTHON="venv/bin/python"
    $spark-submit 
    --name "Sample Spark Application" 
    --conf "spark.yarn.appMasterEnv.SPARK_HOME=$SPARK_HOME" 
    --conf "spark.yarn.appMasterEnv.PYSPARK_PYTHON=$PYSPARK_PYTHON" 
    --archives "venv.zip#venv" 
    --py-files "application.zip" 
    t.py

注意:这里有个命令 export PYSPARK_PYTHON="venv/bin/python" , 指定了PYSPARK_PYTHON 的变量，以便运行的时候指定了python的路径、依赖的包的路径。不然会有找不到依赖包

• 最简单的依赖例子 - 有第三方依赖的case (推荐的解决方案) , 弊端是多worker 需要重复操作。看自己balance, 为快速懒人入门：

  • 只在少量worker or甚至1台worker 的情况下，直接在宿主机安装spark
  • 安装后需要依赖报的，直接在宿主机的 运行pip3 install -r requirements.txt

更多参考：https://github.com/massmutual/sample-pyspark-application

实战例子

处理csv ：

csv.py

import pandas as pd

SPARK_MASTER_HOST = 'ira-r740'
SPARK_MASTER_PORT = 7077
spark = SparkSession
    .builder
    .config('spark.master', 'spark://{spark_master_host}:{spark_master_port}' 
      .format(spark_master_host=SPARK_MASTER_HOST, spark_master_port=SPARK_MASTER_PORT)) 
    .appName("PythonSmsWade")
    .getOrCreate()

def trs2(x):
    try:
        if( x and (x['smsContent'] )):
            return (str(x['smsContent']).upper(),0)
        else:
            return ('',-1)
    except Exception as e:
        return ('',-1)
    
  
#normal quick way
df1=(spark.read.format("csv").options(header="true").load("/home/klg/pyspark/py/100k.csv")) #normal
result=spark.sparkContext.parallelize(df1.collect()).map(lambda x: trs2(x)).collect()
print(len(result))    

设置环境spark 的path变量后（简化提交），
$spark-submit csv.py

注意：
官网例子的文件路径大部分是 hdfs，这里为了简化懒人，简化架构，不依赖hive 分布式存储，file 可以有这么的选项:

• copy 到各个worker 机器上
• 从db 读取（推荐）
• 放在某一台机器上后，其他worker mount 到这个目录 （推荐）

自己需要balance的点： 没有利用到分布式存储对速度上的优势，但大部分case 下面我们需要的处理时间在于多核的cpu 计算时间。

从db 读取数据

import pandas as pd
def test2( x):
    # method 1 : we reinit the var here , simple
    # method 2 : we use global share var  
    from pymongo import MongoClient
    client = MongoClient()
    client = MongoClient('10.10.20.60', 31111)
    query = {}
    docs = list(client['cash_loan']['user'].find(query).limit(200).skip(x*10))
    # do the job 
    # ...
    return docs 

SPARK_MASTER_HOST = 'ira-r740'
SPARK_MASTER_PORT = 7077
spark = SparkSession
    .builder
    .config('spark.master', 'spark://{spark_master_host}:{spark_master_port}' 
      .format(spark_master_host=SPARK_MASTER_HOST, spark_master_port=SPARK_MASTER_PORT)) 
    .appName("Pythonsmswade")
    .getOrCreate()

n=5
result=spark.sparkContext.parallelize(range(1,n)).map(lambda x: test2(x)).collect()

注意：

• return 的might be list of list if test2 return list [limit]*[n] （因为test2方法里面list 化返回值）
• 在test2 里面重新初始MongoClient 变量，以避免全局、共享变量问题

全局变量（共享变量）

To be continue...

那些年踩过的坑

To be continue...