如何在spark数据框中强制重新分区？

2019-06-08 08:45:52 阅读：249 来源： 互联网

标签：python apache-spark distributed pyspark

我有许多火花数据帧,我需要在其中执行以下操作：

1) load a single spark dataframe
2) select rows from it
3) merge it with all of the previous spark dataframes

现在,上述每个操作都需要不同数量的分区.选择行需要许多分区,例如100个分区.合并需要很少的分区,比如10个分区.

所以,我真的希望它像这样工作：

1) load a single spark dataframe
1.5) repartition into 100 partitions
2) select rows from it
2.5) repartition into 10 partitions
3) merge it with all of the previous spark dataframes

现在,如何强制它在步骤1和2之间以及2到3之间重新分配？

我知道当我调用data = data.repartition(7)时,它被懒惰地评估,因此它只在实际保存时重新分区.

所以,我一直这样做：

1) load a single spark dataframe
1.5) repartition into 100 partitions
1.75) `df.count()` *just* to force materialization
2) select rows from it
2.5) repartition into 10 partitions
2.75) `df.count()` *just* to force materialization
3) merge it with all of the previous spark dataframes

有没有更好的方法强迫它在这里重新分配？有没有比在数据帧上运行count()更好的方法？

解决方法:

由于对spark中数据帧的所有转换都进行了懒惰评估,因此您需要执行一个操作来实际执行转换.目前没有其他方法可以强制进行转换.

可以在documentation中找到所有可用的数据帧操作(查看操作).在你的情况下,而不是使用count()来强制转换,你可以使用first(),这应该是显着更便宜.

在步骤2.5中,您可以使用coalesce()替换repartition(),因为它将避免完全shuffle.当新的分区数量少于之前时,这通常是有利的,因为它将最小化数据移动.

编辑：

如果您不使用任何操作,只需执行以下操作即可回答您的问题：1)重新分区,2)火花数据帧转换,3)重新分区.由于优化火花对变换执行,似乎并不总是遵循该顺序.我制作了一个小测试程序来测试它：

val df = spark.sparkContext.parallelize(Array((1.0,"a"),(2.0,"b"),(3.0,"c"),(1.0,"d"),(2.0,"e"),(3.0,"f"))).toDF("x", "y")
val df1 = df.repartition(10).filter($"x" =!= 1.0).repartition(5).filter($"y" =!= "b")
df1.explain(true)

这将返回有关如何计算数据帧的信息.

== Parsed Logical Plan ==
'Filter NOT ('y = b)
+- Repartition 5, true
   +- Filter NOT (x#5 = 1.0)
      +- Repartition 10, true
         +- Project [_1#2 AS x#5, _2#3 AS y#6]
            +- LogicalRDD [_1#2, _2#3]

== Analyzed Logical Plan ==
x: double, y: string
Filter NOT (y#6 = b)
+- Repartition 5, true
   +- Filter NOT (x#5 = 1.0)
      +- Repartition 10, true
         +- Project [_1#2 AS x#5, _2#3 AS y#6]
            +- LogicalRDD [_1#2, _2#3]

== Optimized Logical Plan ==
Repartition 5, true
+- Project [_1#2 AS x#5, _2#3 AS y#6]
   +- Filter ((NOT (_1#2 = 1.0) && isnotnull(_2#3)) && NOT (_2#3 = b))
      +- LogicalRDD [_1#2, _2#3]

== Physical Plan ==
Exchange RoundRobinPartitioning(5)
+- *Project [_1#2 AS x#5, _2#3 AS y#6]
   +- *Filter ((NOT (_1#2 = 1.0) && isnotnull(_2#3)) && NOT (_2#3 = b))
      +- Scan ExistingRDD[_1#2,_2#3]

从这里可以看出,重新分区(10)步骤不包括在内,似乎在优化过程中已被删除.

标签：python,apache-spark,distributed,pyspark
来源： https://codeday.me/bug/20190608/1196402.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

如何在spark数据框中强制重新分区？