对于每个分区中的元素集,Spark RDD 是否具有确定性?
Is a Spark RDD deterministic for the set of elements in each partition?
我找不到太多关于确保分区顺序的文档 - 我只是想确保给定一组确定性转换(输出行始终相同),如果基础数据集没有,分区总是接收相同的元素集'改变。那可能吗?
它不需要排序:一个例子是在一个 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) 分别。
这是因为我必须让我的执行程序根据它所操作的元素的 partition/set 产生一些副作用,并且我想确保 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)']
关于容错,如果一个分区失败,它将被重新提交给另一个执行者,这样分区就不会被排序。
Spark 有一些内部机制来计算关于您拥有的数据的最佳执行计划。所以你无法预测分区的顺序和内容。
在内部,Spark使用一个默认分区器(HashPartitioner取决于数据)对数据进行分区,它使用散列来标识哪个该项目所属的分区。因此,你可以说数据项将始终进入同一个分区,因为分区数是相同的,因为如果分区数发生变化,它也会影响散列。
跨分区的记录分布不需要均匀。保证分区数,每个分区的记录数大致相同。对于任何操作都没有关系。如果由于任何原因发生随机播放,则将重新创建新分区。
假设 (A, B, C, D, E, F, G) 被划分为 2 为 (A, B, C, D) 和 (E, F, G)。如果执行者处理(E,F,G)死亡,那么 Spark 将重新启动它并尝试重新处理(E,F,G)。如果这个执行器是不可恢复的,那么整个作业将失败,它将重新开始(A、B、C、D、E、F、G)分成两部分并重新开始处理。在第二次尝试中,它可能分裂为(A,B,C)和(D,E,F,G)。处理的最终结果将相同。
让我们看看 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)
的结果也应该每次都相同。
除了大家说的:
- Dataframe 基于 RDD。
- 你说:Dataframe 重新分区是确定性的。
- 如果 RDD 重新分区不是确定性的,那么 Dataframe 重新分区也不是。
我找不到太多关于确保分区顺序的文档 - 我只是想确保给定一组确定性转换(输出行始终相同),如果基础数据集没有,分区总是接收相同的元素集'改变。那可能吗?
它不需要排序:一个例子是在一个 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) 分别。
这是因为我必须让我的执行程序根据它所操作的元素的 partition/set 产生一些副作用,并且我想确保 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)']
关于容错,如果一个分区失败,它将被重新提交给另一个执行者,这样分区就不会被排序。
Spark 有一些内部机制来计算关于您拥有的数据的最佳执行计划。所以你无法预测分区的顺序和内容。
在内部,Spark使用一个默认分区器(HashPartitioner取决于数据)对数据进行分区,它使用散列来标识哪个该项目所属的分区。因此,你可以说数据项将始终进入同一个分区,因为分区数是相同的,因为如果分区数发生变化,它也会影响散列。
跨分区的记录分布不需要均匀。保证分区数,每个分区的记录数大致相同。对于任何操作都没有关系。如果由于任何原因发生随机播放,则将重新创建新分区。
假设 (A, B, C, D, E, F, G) 被划分为 2 为 (A, B, C, D) 和 (E, F, G)。如果执行者处理(E,F,G)死亡,那么 Spark 将重新启动它并尝试重新处理(E,F,G)。如果这个执行器是不可恢复的,那么整个作业将失败,它将重新开始(A、B、C、D、E、F、G)分成两部分并重新开始处理。在第二次尝试中,它可能分裂为(A,B,C)和(D,E,F,G)。处理的最终结果将相同。
让我们看看 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)
的结果也应该每次都相同。
除了大家说的:
- Dataframe 基于 RDD。
- 你说:Dataframe 重新分区是确定性的。
- 如果 RDD 重新分区不是确定性的,那么 Dataframe 重新分区也不是。