我在 pyspark 中有一个 rdd.foreachPartition(some_function) Action 。 some_function 函数将执行器中当前任务的数据写入所有执行器(如 hdfs 或 s3 存储桶)共有位置的文件中。
现在，如果我在所有执行程序中使用相同的文件名，该文件将被替换并只留下最后写入的文件。所以我正在寻找一个唯一的标识符来表示每个任务，从而表示每个文件名。
我对任务 ID 很感兴趣，因为它是独一无二的。但是找不到任何地方，如何在 pyspark 中获取任务 ID。我发现了一些东西 similar在 scala/java 中但不在 pyspark 中。
更新:按照建议，我检查了this .但是，这给出了 stageID，而我对每个阶段内各个任务的 taskID 感兴趣。不过，stageId 答案也是一个很好的信息。
请您参考如下方法:
我前段时间遇到了同样的问题。
我通过在文件名中使用 datetime.now() 解决了这个问题，并且绝对确定我不会得到我也使用过的相同文件名:
rdd.mapPartitionsWithIndex(lambda x,y: (x,y), preservesPartitioning = True).foreachPartition(lambda x: some_function(x))
这将在 x[0] 处为您提供一个唯一的分区 ID，您可以将其添加到任何文件名以确保唯一性
