我有一个名为“table1”的表，我正在根据一个条件拆分它，然后在for循环中逐个连接拆分的部分。
当我加入他们的时候，加入的时间呈指数级增加。
0.7423694133758545
join
0.4046192169189453
join
0.5775985717773438
join
5.664674758911133
join
1.0985417366027832
join
2.2664384841918945
join
3.833379030227661
join
12.762675762176514
join
44.14520192146301
join
124.86295890808105
join
389.46189188957214。以下是我的参数
spark = SparkSession.builder.appName("xyz").getOrCreate()
sqlContext = HiveContext(spark)
sqlContext.setConf("spark.sql.join.preferSortMergeJoin", "true")
sqlContext.setConf("spark.serializer","org.apache.spark.serializer.KryoSerializer")
sqlContext.setConf("spark.sql.shuffle.partitions", "48")
sqlContext.setConf("spark.sql.shuffle.partitions", "48")和
--executor-memory 16G --num-executors 8 --executor-cores 8 --driver-memory 32G在join迭代中，我也将分区数增加到2000，然后再减少到4，并通过df.cached（）缓存DF Dataframe ，但是没有任何效果。我知道我做错了什么，但我不知道是什么。请您指导我如何纠正这个错误。
我真的很感激任何帮助：）
1条答案
按热度按时间vh0rcniy1#
我有同样的问题前一段时间。如果你检查你的pyspark的web用户界面，并进入阶段部分，并检查你的任务的DAG可视化，你可以看到DAG是指数增长，等待时间，你看到的是使这个DAG不做任务acutally。我不知道为什么，但它接缝时，你加入表的 Dataframe 与它自己pyspark不能处理分区，它变得更大。如何以往的解决方案，我发现在那一刻是保存每个连接结果在单独的文件，并在结束后，重新启动内核加载，并再次连接所有的文件。它似乎如果 Dataframe ，你想连接不是从对方，你不会看到这个问题。