apache-spark - Spark RDD 对于每个分区中的元素集是否具有确定性？

Question

我找不到太多关于确保分区顺序的文档 - 我只是想确保给定一组确定性转换(输出行始终相同)，如果基础数据集不接收相同的元素集，则分区始终接收相同的元素集改变。那可能吗？

它不需要排序:一个例子是在 RDD 上应用一组转换后，现在看起来像这样 -> (A, B, C, D, E, F, G)

如果我的 Spark.default.parallelism 是 2 或 3，则元素集将始终是:分别为(A、B、C、D)、(E、F、G)或(A、B)、(C、D)、(E、F、G)。

这是因为我必须让我的执行器根据其正在操作的分区/元素集产生一些副作用，并且我想确保 Spark 应用程序是幂等的。 (如果重新启动，也会产生相同的副作用)

编辑:显然，DF 重新分区是确定性的，但 RDD 分区不是(Spark 2.4.4)。

def f1(rdds):
    rows = list(rdds)
    stats_summary = [{
        'origin': str(row['origin']),
        'dest': str(row['dest']),
        'start_time': analysis_date.isoformat(),
        'value': row['count']
    } for row in rows]

    stats_summary.sort(key=lambda t: (t['start_time'], t['origin'], t['dest']))

    rtn = "partition size: {}, first: ({}, {}), last: ({}, {})".format(
        len(rows), 
        stats_summary[0]["origin"], stats_summary[0]["dest"],
        stats_summary[-1]["origin"], stats_summary[-1]["dest"])
    return [rtn]

repartition_rdd_res = unq_statistics.rdd \
                                    .repartition(10) \
                                    .mapPartitions(f1) \
                                    .collect()

repartition_df_res = unq_statistics.repartition(10) \
                                   .rdd \
                                   .mapPartitions(f1) \
                                   .collect()

repartition_rdd_res4 = ['partition size: 131200, first: (-1, -1), last: (999, -1)',
 'partition size: 131209, first: (-1, 1014), last: (996, 996)',
 'partition size: 131216, first: (-1, 1021), last: (999, 667)',
 'partition size: 131218, first: (-1, 1008), last: (991, 1240)',
 'partition size: 131222, first: (-1, 1001), last: (994, 992)',
 'partition size: 131229, first: (-1, 1007), last: (994, 890)',
 'partition size: 131233, first: (-1, 1004), last: (991, -1)',
 'partition size: 131235, first: (-1, 1005), last: (999, 1197)',
 'partition size: 131237, first: (-1, 100), last: (999, 997)',
 'partition size: 131240, first: (-1, 1010), last: (994, -1)']

repartition_rdd_res3 = ['partition size: 131200, first: (-1, -1), last: (999, -1)',
 'partition size: 131209, first: (-1, 1006), last: (994, 2048)',
 'partition size: 131216, first: (-1, 1002), last: (996, 996)',
 'partition size: 131218, first: (-1, 1017), last: (999, 667)',
 'partition size: 131222, first: (-1, 1008), last: (994, 890)',
 'partition size: 131229, first: (-1, 1000), last: (99, 96)',
 'partition size: 131233, first: (-1, 1001), last: (994, 992)',
 'partition size: 131235, first: (-1, 1009), last: (990, 1601)',
 'partition size: 131237, first: (-1, 1004), last: (994, -1)',
 'partition size: 131240, first: (-1, 1003), last: (999, 997)']

repartition_rdd_res2 = ['partition size: 131200, first: (-1, 1013), last: (991, 2248)',
 'partition size: 131209, first: (-1, 1007), last: (999, 667)',
 'partition size: 131216, first: (-1, 100), last: (99, 963)',
 'partition size: 131218, first: (-1, 1002), last: (999, 997)',
 'partition size: 131222, first: (-1, 101), last: (996, 996)',
 'partition size: 131229, first: (-1, -1), last: (991, 1240)',
 'partition size: 131233, first: (-1, 1006), last: (999, 1197)',
 'partition size: 131235, first: (-1, 1001), last: (994, 992)',
 'partition size: 131237, first: (-1, 1019), last: (999, -1)',
 'partition size: 131240, first: (-1, 1017), last: (991, -1)']

repartition_df_res2 = ['partition size: 131222, first: (-1, 1023), last: (996, 996)',
 'partition size: 131223, first: (-1, 1003), last: (999, 667)',
 'partition size: 131223, first: (-1, 1012), last: (990, 990)',
 'partition size: 131224, first: (-1, -1), last: (999, 1558)',
 'partition size: 131224, first: (-1, 100), last: (99, 98)',
 'partition size: 131224, first: (-1, 1008), last: (99, 968)',
 'partition size: 131224, first: (-1, 1018), last: (999, 997)',
 'partition size: 131225, first: (-1, 1006), last: (994, 992)',
 'partition size: 131225, first: (-1, 101), last: (990, 935)',
 'partition size: 131225, first: (-1, 1013), last: (999, 1197)']

Answer 1

让我们看看 source ，特别是它的随机播放部分:

...
if (shuffle) {
  /** Distributes elements evenly across output partitions, starting from a random partition. */
  val distributePartition = (index: Int, items: Iterator[T]) => {
    var position = new Random(hashing.byteswap32(index)).nextInt(numPartitions)
    items.map { t =>
      // Note that the hash code of the key will just be the key itself. The HashPartitioner
      // will mod it with the number of total partitions.
      position = position + 1
      (position, t)
    }
  } : Iterator[(Int, T)]
  ...

正如您所看到的，元素从给定源分区 N 到 X 目标分区的分布是一个简单的增量(稍后按 X) 从某个仅取决于 N 的数字开始，因此是预先确定的。因此，如果您的源 RDD 未更改，repartition(X) 的结果每次也应该相同。