最近有个需求需要union 上千甚至更多的dataset数据，然后cache(),然后count(),在执行count(）这个action操作的时候，spark程序报错，如下：

org.apache.spark.SparkException: Job aborted due to stage failure: Total size of serialized results of 16092 tasks (16.0 GB) is bigger than spark.driver.maxResultSize (16.0 GB) ? ? ? ? at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1890) ? ? ? ? at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1878) ? ? ? ? at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1877)

网上一百度都说要调大 spark.driver.maxResultSize， 但是已经从8G调大到16G依然报错，而且，此配置不能无限调大把。

后来，查看spark? ui中的stage,发现这个count()产生了2万多条task，虽然最终数据很少，但是这里的task数据量很大，聚合到driver肯定OOM。 因此决定调小partition数量。

把每个Dataset在经过spark sql执行之后需要coalesce(1),然后经过union之后，会有1000个分区数据，再重分区20。而且要用sparkContext （sc）里面的union算子，而不能使用dataset的union算子，那样会有无数次的union，而用sc会一次union起来。

这样最终count()执行时，大概有2千多task, 这时还需要调节executor堆外内存，配置如下，默认只有300M,如果不设置会有警告。

--conf spark.executor.memoryOverhead=5g

for(int i=0;i<1000;i++) { Dataset<Row> rowSeqSet = sparkSession.sql(rowSeq).coalesce(1); if(i==0){ first = colSet.toJavaRDD(); } else{ list.add(colSet.toJavaRDD()); } } JavaRDD<Row> allRDD = sc.union(first, list).repartition(20); StructType type1 = new StructType(new StructField[]{ createStructField("A", IntegerType, true), createStructField("B", StringType, true), createStructField("C", IntegerType, true), createStructField("D", IntegerType, true), createStructField("E", StringType, true), createStructField("F", StringType, true) }); allSet = sparkSession.createDataFrame(allRDD,type1); allSet = allSet.persist(StorageLevel.MEMORY_AND_DISK_SER()); System.out.println("total allSet count="+allSet.count()); ?

---

1.本站遵循行业规范，任何转载的稿件都会明确标注作者和来源；2.本站的原创文章，会注明原创字样,如未注明都非原创，如有侵权请联系删除！；3.作者投稿可能会经我们编辑修改或补充；4.本站不提供任何储存功能只提供收集或者投稿人的网盘链接。