将实现某业务逻辑的pyspark代码翻译成sparksql,基于sparksql补充过去半年的历史数据(按天跑);
1)将pyspark翻译成sparksql;
2)基于sparksql,补充过去半年的历史数据(按天跑);
1)首先,pyspark翻译成spark sql,大部分直接翻译;基于原来共同的地方,可以缓冲一个cache表,后续不用多次计算;
2)其次,翻译完sparksql之后,需要补充跑过去半年的数据;通常有几种方式:方式1,一天一天的轮询循环跑;方式2,并行跑;
我尝试选择的是并行跑,将业务逻辑表进行梳理:一张输入表;一张输出表;n张中间表;
并行跑代码框架:
from multiprocessing import Process from pyspark.sql import SparkSession sc = SparkSession \ .builder \ .appName("Python Spark SQL basic example") \ .config("spark.some.config.option", "some-value") \ .enableHiveSupport() \ .getOrCreate() date_range_list = ['20210215', '20201130', '20201214'] def run_sql_process(date): run_sql = " xxx ... " run_sql = run_sql.replace('${UOW_TO_DT}', date) run_sql_list = run_sql.split(';') for run_sql in run_sql_list: if run_sql: sc.sql(run_sql) def mutil_process(): for date in date_range_list: # 创建进程 run_sql_process(date) if __name__ == '__main__': mutil_process() print("程序运行成功!!!!!")
并行跑策略1:
通过python的多进程调用,每次一个spark session去同时运行多天的spark sql,发现这个跑不起来(从并行数5,设置到并行数2),报OOM资源不足;
猜想:是由于 一个spark session本身申请的资源已定,如果同时运行多个spark sql,资源占用较大,所以内存溢出;
在上述代码中,同一个session下,同时运行多个run_sql(run_sql就是纯的像insert overwrite table...)
并行跑策略2:
猛然惊醒,我的这个并行有点不太合理,总资源没变,但是一下多并行跑了几个spark sql的任务,很显然没有达到并行提速的效果,只可能导致大家都跑不起,因为总资源是有限的;因此,正确的方式是,应该用并行提交spark_submit的方式,并行资源运行任务;
在上述代码中,同一个session下,同时运行多个run_sql(像insert overwrite table...) 并行任务1: /apache/spark2.3/bin/spark-submit --master yarn --deploy-mode client --queue xxx --conf spark.pyspark.driver.python=/usr/share/anaconda2/bin/python --conf spark.pyspark.python=/usr/share/anaconda2/bin/python --executor-cores 4 --executor-memory 20g --driver-memory 10g /apache/releases/spark-2.3.1.1.1.6-bin-ebay/python/pyspark/tests.py 并行任务2: /apache/spark2.3/bin/spark-submit --master yarn --deploy-mode client --queue xxx --conf spark.pyspark.driver.python=/usr/share/anaconda2/bin/python --conf spark.pyspark.python=/usr/share/anaconda2/bin/python --executor-cores 4 --executor-memory 20g --driver-memory 10g /apache/releases/spark-2.3.1.1.1.6-bin-ebay/python/pyspark/tests2.py
但是,这个方式理论上可行,但是也遇到了问题,就是sparksql并行插入同一张表的不同分区的时候,会报lock的错误,因此中间的表全都设置为临时表的带上日期的表,这样每次跑就不会冲突;最终采取的方式是在公司的可视化提交平台上一天一天线性跑,因为只是补一次历史数据;后续就只需要每天定时跑就可以了;