PySpark：重新分区vs partitionBy
rdd = sc.parallelize([('a', 1), ('a', 2), ('b', 1), ('b', 3), ('c',1), ('ef',5)])
rdd1 = rdd.repartition(4)
rdd2 = rdd.partitionBy(4)
rdd1.glom().collect()
[[('b', 1), ('ef', 5)], [], [], [('a', 1), ('a', 2), ('b', 3), ('c', 1)]]
rdd2.glom().collect()
[[('a', 1), ('a', 2)], [], [('c', 1)], [('b', 1), ('b', 3), ('ef', 5)]]
既然这样，为什么还会有人使用重新分区呢？我想我唯一能看到它被使用的时候是如果我没有使用PairRDD，或者我有很大的数据偏差？
我是否遗漏了什么，或者有人能从不同的角度为我照亮吗？
共有1个答案
repartition()用于根据核心数和数据量指定分区数。
PartitionBy()用于提高洗牌函数的效率，例如ReduceByKey()、Join()、cogroup()等。它只在多次使用RDD的情况下有用，所以通常在后面跟persist()。
两者在行动上的差异：
pairs = sc.parallelize([1, 2, 3, 4, 2, 4, 1, 5, 6, 7, 7, 5, 5, 6, 4]).map(lambda x: (x, x))
pairs.partitionBy(3).glom().collect()
[[(3, 3), (6, 6), (6, 6)],
[(1, 1), (4, 4), (4, 4), (1, 1), (7, 7), (7, 7), (4, 4)],
[(2, 2), (2, 2), (5, 5), (5, 5), (5, 5)]]
pairs.repartition(3).glom().collect()
[[(4, 4), (2, 2), (6, 6), (7, 7), (5, 5), (5, 5)],
[(1, 1), (4, 4), (6, 6), (4, 4)],
[(2, 2), (3, 3), (1, 1), (5, 5), (7, 7)]]
-
步骤3我通过for循环加载每个分区，执行聚合，并以追加模式将其保存为文件夹，这样我就有9个模块作为文件夹:、等。它们不按模块分区，只是保存为文件夹。由于我的默认spark numpartitions是，每个模块文件夹都有文件，因此总共有文件 步骤4到目前为止还不错，但是我需要按把它分区回来。因此，我循环遍历每个分区，并将文件保存为一个没有任何分区的parquet文件。这导致总共有文件。我不知道这是
-
我有一个具有如下模式的dataframe：
-
我在这里浏览了文档:https://spark . Apache . org/docs/latest/API/python/py spark . SQL . html 它说： 重新分区：生成的DataFrame是哈希分区的 对于repartitionByRange：结果DataFrame是范围分区的 而且之前的一个问题也提到了。然而，我仍然不明白它们到底有什么不同，当选择一个而不是另一个时会有什么
-
根据Spark 1.6.3的文档，应该保留结果数据表中的分区数： 返回由给定分区表达式分区的新DataFrame，保留现有的分区数 Edit：这个问题并不涉及在Apache Spark中删除空DataFrame分区的问题（例如，如何在不产生空分区的情况下沿列重新分区），而是为什么文档所说的内容与我在示例中观察到的内容不同
-
我有一个像下面的输入数据一样的pyspark数据框架。我想在空白上拆分productname列中的值。然后我想用前3个值创建新列。我在下面有示例输入和输出数据。有人能建议如何使用pyspark做到这一点吗？ 输入数据： 输出：
-
假设我有一个1.2 GB的文件，那么考虑到128 MB的块大小，它将创建10个分区。现在，如果我将其重新分区(或合并)为4个分区，这意味着每个分区肯定会超过128 MB。在这种情况下，每个分区必须容纳320 MB的数据，但块大小是128 MB。我有点糊涂了。这怎么可能？我们如何创建一个大于块大小的分区？