如何在不进行写入的情况下评估累加器？

Question

我想在编写数据框之前执行轻量级验证。在写入之前，我必须通过“foo”序列化数据帧。我在“foo”中增加一个累加器：

acc = sc.accumulator(0)
output = df.map(foo)
if acc.value < THRESHOLD:
 raise ValueError(f"Failed validation: {acc.value} < {THRESHOLD}")
output.write(path)

问题是 acc.value == 0，因为显然直到 output.write() 才评估累加器，我想避免这种情况，因为数据验证失败。什么是正确的设计模式？

Answer 1

如果您的目标是在将数据发布到某个输出路径之前验证计数，只需将数据写入中间路径即可。然后评估累加器计数器，如果计数有效 - 将中间路径重命名为实际输出目的地。

acc = sc.accumulator(0)
output = df.map(foo)
output.write(tmp_path)
if acc.value < THRESHOLD:
 # fs.delete(tmp_path)
 raise ValueError(f"Failed validation: {acc.value} < {THRESHOLD}")
else fs.rename(tmp_path, path)

如何在不进行写入的情况下评估累加器？

How to evaluate an accumulator without incurring a write?

counter

apache-spark

pyspark