создать фрейм данных в цикле foreach apache spark [duplicate]

Вы не можете получить доступ к любым абстракциям «Исправлена» (RDDs, DataFrames, Datasets, SparkSession ...) Spark из функции, переданной в одну из преобразований DataFrame / RDD Spark. Вы также не можете обновлять измененные объекты на стороне драйвера из этих функций.

В вашем случае - вы пытаетесь использовать prodRows и selection (оба являются DataFrames) в функции, переданной в DataFrame.foreach. Вы также пытаетесь обновить listOfProducts (локальную драйверную переменную) из этой же функции.

Почему?

• DataFrames, RDD и SparkSession существуют только в приложении Driver. Они служат «ручкой» для доступа к данным, распределенным по кластеру рабочих машин.
• Функции, переданные в преобразования RDD / DataFrame, получают serialized и отправляются в этот кластер, которые должны выполняться на разделах данных на каждой рабочей машине. Когда сериализованные DataFrames / RDD получаются десериализованными на этих машинах - они бесполезны, они не могут представлять данные в кластере, поскольку они представляют собой только полные копии тех, которые созданы в приложении драйвера, что фактически поддерживает соединение для кластерных машин
• По той же причине попытка обновления драйверов переменных не будет выполнена: переменные (начинающиеся как пустые, в большинстве случаев) будут сериализованы, десериализованы на каждом из рабочие, обновляются локально на рабочих и остаются там ... исходная переменная со стороны водителя останется неизменной

Как вы можете это решить? При работе с Spark, особенно с DataFrames, вы должны стараться избегать «итерации» над данными и вместо этого использовать декларативные операции DataFrame. В большинстве случаев, когда вы хотите ссылаться на данные другого DataFrame для каждой записи в вашем DataFrame, вы хотите использовать join для создания нового DataFrame с записями, объединяющими данные из двух DataFrames.

В этом конкретном случае, это примерно эквивалентное решение, которое делает то, что вы пытаетесь сделать, если мне удастся это сделать правильно. Попытайтесь использовать это и прочитайте документацию DataFrame, чтобы выяснить детали:

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._
import spark.implicits._

val numRecProducts = 10

val result = prodRows.as("left")
  // self-join by gender:
  .join(prodRows.as("right"), $"left.gender_PK" === $"right.gender_PK" || $"right.gender_PK" === "UNISEX")
  // limit to 10 results per record:
  .withColumn("rn", row_number().over(Window.partitionBy($"left.product_PK").orderBy($"right.product_PK")))
  .filter($"rn" <= numRecProducts).drop($"rn")
  // group and collect_list to create products column:
  .groupBy($"left.product_PK" as "product_PK")
  .agg(collect_list(struct($"right.product_PK", lit(1))) as "products")