Распределяет ли Spark фрейм данных по узлам внутри?

Question

Я пытаюсь использовать Spark для обработки CSV-файла в кластере. Я хочу понять, нужно ли мне явно читать файл на каждом из рабочих узлов для параллельной обработки, или будет ли узел драйвера читать файл и распределять данные по кластеру для внутренней обработки? (Я работаю с Spark 2.3.2 и Python)

Я знаю, что RDD можно распараллелить с помощью SparkContext.parallelize (), но что в случае Spark DataFrames?

if __name__=="__main__":
     spark=SparkSession.builder.appName('myApp').getOrCreate()
     df=spark.read.csv('dataFile.csv',header=True)
     df=df.filter("date>'2010-12-01' AND date<='2010-12-02' AND town=='Madrid'")

Итак, если я запускаю приведенный выше код в кластере, будет ли вся операция выполняться узлом драйвера или он будет распределять df по кластеру, и каждый работник будет выполнять обработку на своем разделе данных?

Answer 1

Строго говоря, если вы запустите приведенный выше код, он не будет читать или обрабатывать какие-либо данные. Фреймы данных - это в основном абстракция, реализованная поверх RDD. Как и в случае RDD, вы должны различать преобразования и действия . Поскольку ваш код состоит только из одного преобразования filter(...), это будет происходить с точки зрения повторного считывания или обработки данных. Spark создаст только DataFrame, который является планом выполнения. Вы должны выполнить действие типа count() или write.csv(...), чтобы фактически запустить обработку файла CSV.

Если вы сделаете это, данные будут затем прочитаны и обработаны 1..n рабочими узлами. Он никогда не читается и не обрабатывается узлом драйвера. Сколько или ваши рабочие узлы на самом деле задействованы, зависит - в вашем коде - от количества разделов вашего исходного файла. Каждый раздел исходного файла может обрабатываться параллельно одним рабочим узлом. В вашем примере это, вероятно, один CSV-файл, поэтому, когда вы вызываете df.rdd.getNumPartitions() после прочтения файла, он должен возвращать 1. Следовательно, только один рабочий узел будет читать данные. То же самое верно, если вы проверите количество разделов после операции filter(...).

Вот два способа параллельной обработки вашего отдельного файла CSV:

1. Вы можете вручную перераспределить исходный DataFrame, вызвав df.repartition(n) с n числом разделов, которое вы хотите иметь. Но - и это немаловажно, но - это означает, что все данные потенциально могут передаваться по сети (иначе это случайный порядок)!

2. Вы выполняете агрегации или объединения в DataFrame. Эти операции имеют , чтобы вызвать случайное перемешивание. Затем Spark использует количество разделов, указанное в spark.sql.shuffle.partitions (по умолчанию: 200), для разбиения результирующего кадра данных.