Получить идентификатор задачи исполнителя в pyspark

Question

У меня есть действие rdd.foreachPartition(some_function) в pyspark. Функция some_function записывает данные текущей задачи в исполнителе в файл, который является общим для всех исполнителей (например, hdfs или s3 bucket).

Теперь, если я использую одно и то же имя файла для всех исполнителей, файл заменяется и остается только последний записанный файл. Поэтому я ищу уникальный идентификатор для представления каждой задачи и, следовательно, для каждого имени файла.

Меня заинтересовал идентификатор задачи, поскольку он уникален. Но нигде не смог найти, как получить идентификатор задачи в pyspark. Я нашел что-то подобное в Scala / Java, но не в Pyspark.

Обновление: Как и предполагалось, я проверил это . Тем не менее, это дает stageID, в то время как я был заинтересован в TaskID отдельных задач внутри каждого этапа. Ответ stageId также является хорошей информацией, которую нужно знать.

Answer 1

У меня была такая же проблема некоторое время назад.

Я решил это, используя datetime.now() в имени файла, и чтобы быть абсолютно уверенным, я не получу те же имена файлов, которые я также использовал:

rdd.mapPartitionsWithIndex(lambda x,y: (x,y), preservesPartitioning = True).foreachPartition(lambda x: some_function(x))

Это даст вам уникальный идентификатор раздела в x [0], который вы можете добавить к любому имени файла, чтобы обеспечить уникальность