Pyspark mapPartition оценивает функцию больше раз, чем ожидалось ⇐ Python

[Anonymous]

Я работаю с PySpark для обработки больших объемов данных. Однако я заметил, что функция, вызванная mapPartitions, выполняется еще раз, чем ожидалось. Например, в следующем блоке кода функцию reformat следует вызывать четыре раза, но она вызывается пять раз: четыре раза, когда DataFrame кэшируется, и пятый раз, когда вызывается метод show.

Код: Выделить всё

from pyspark.sql import SparkSession

spark = SparkSession.builder.config("spark.driver.host", "localhost").appName('SparkByExamples.com').getOrCreate()
data = [
('James','Smith','M',3000),
('James','Smith','M',3000),
('James','Smith','M',3000),
('James','Smith','M',3000),
('Anna','Rose','F',4100),
('Anna','Rose','F',4100),
('Anna','Rose','F',4100),
('Anna','Rose','F',4100),
('Robert','Williams','M',6200),
('Robert','Williams','M',6200),
('Robert','Williams','M',6200),
('Robert','Williams','M',6200),
]

columns = ["firstname","lastname","gender","salary"]
df = spark.createDataFrame(data=data, schema = columns)
df.show()

#Example 1 mapPartitions()
def reformat(partitionData):
for row in partitionData:
yield [row.firstname+","+row.lastname,row.salary*10/100]

df2=df.repartition(4).rdd.mapPartitions(reformat).toDF(["name","bonus"])
df2.cache()
df2.show()

В этом случае время может не иметь значения. Однако для больших наборов данных это может стать серьезной проблемой. Кроме того, если функция включает вызовы внешнего API для получения данных или оценки какой-либо логики, это может привести к дублированию или избыточности запросов API.
Я тестировал приведенный выше пример с более простой логикой, но не обнаружил проблемы.

Подробнее здесь: https://stackoverflow.com/questions/793 ... n-expected