最近用了一个RowNumber() over()函数 进行三张4000万数据的关联筛选，建表语句如下:
create table CiCustomerPortrait2 as SELECT ROW_NUMBER() OVER() as id,* from (select t_7.phone_no,t_7.L1301,t_7.L1431,t_7.L1449,t_7.L1489,t_8.L1713,t_92.L1879,t_92.L1907 from DW_COC_LABEL_INTERNET_D_20151123 t_7 inner join DW_COC_LABEL_INTERNET_M_201510 t_8 on t_7.phone_no = t_8.phone_no inner join DW_COC_LABEL_BITEMP_M_201510 t_92 on t_7.phone_no = t_92.phone_no ) a 一方面由于多表关联(每个表4000W以上数据)一方面窗口函数的原因异常吃内存。
spark-env.sh参数如下:(default我就不贴了)
![]()
给了18个executor,每个executor有12G内存，每台服务器启动3个core,那么每个core就是4G内存。但执行过程中，一直卡在这里：
![]()
查询相关资料。
Shuffle的数据如何拉取过来
作业提交的时候，DAGScheduler会把Shuffle的过程切分成map和reduce两个Stage（之前一直被我叫做shuffle前和shuffle后）,map的中间结果是写入到本地硬盘的，而不是内存,所以对磁盘的读写要求非常高，(最好是固态硬盘比较快，本人亲自尝试，同样的性能参数下，固态硬盘会比普通磁盘快10倍。)默认是一个map的中间结果文件是M*R（M=map数量，R=reduce的数量），设置了spark.shuffle.consolidateFiles为true之后是R个文件，根据bucketId把要分到同一个reduce的结果写入到一个文件中。MapOutputTrackerWorker向MapOutputTrackerMaster获取shuffle相关的map结果信息。把map结果信息构造成BlockManagerId --> Array(BlockId, size)的映射关系,通过BlockManager的getMultiple批量拉取block。
当过了N久执行过去了后，将生成好的文件拷贝到hdfs相应路径下
![]()
最终将生成好的文件拷贝到目录下，整个耗时10多分钟。Spark beeline方式连接有个缺陷，如果你一个job执行的时间过长，就会卡在那里，即便执行完也卡在那里，这样项目中用jdbc连接的时候，程序也不会退出，一直等待着结束，造成程序无法继续向下执行。这个还要调整各方面参数想办法优化执行效率。
