Следующий код Spark правильно демонстрирует, что я хочу сделать, и генерирует правильный вывод с небольшим набором демонстрационных данных.

Когда я запускаю этот же общий тип кода для большого объема производственных данных, у меня возникают проблемы во время выполнения. Задание Spark выполняется в моем кластере в течение ~ 12 часов и не выполняется.

Просто взглянув на приведенный ниже код, кажется неэффективным разбивать каждую строку, просто чтобы объединить ее обратно. В данном наборе тестовых данных четвертая строка с тремя значениями в array_value_1 и тремя значениями в array_value_2, которые будут разнесены до 3 * 3 или девяти разнесенных строк.

Таким образом, в большом наборе данных строка с пятью такими столбцами массива и десятью значениями в каждом столбце взорвалась бы до 10 ^ 5 разнесенных строк?

Глядя на предоставляемые функции Spark, нет готовых функций, которые бы делали то, что я хочу. Я мог бы предоставить пользовательскую функцию. Есть ли какие-либо недостатки в скорости?

val sparkSession = SparkSession.builder.
  master("local")
  .appName("merge list test")
  .getOrCreate()

val schema = StructType(
  StructField("category", IntegerType) ::
    StructField("array_value_1", ArrayType(StringType)) ::
    StructField("array_value_2", ArrayType(StringType)) ::
    Nil)

val rows = List(
  Row(1, List("a", "b"), List("u", "v")),
  Row(1, List("b", "c"), List("v", "w")),
  Row(2, List("c", "d"), List("w")),
  Row(2, List("c", "d", "e"), List("x", "y", "z"))
)

val df = sparkSession.createDataFrame(rows.asJava, schema)

val dfExploded = df.
  withColumn("scalar_1", explode(col("array_value_1"))).
  withColumn("scalar_2", explode(col("array_value_2")))

// This will output 19. 2*2 + 2*2 + 2*1 + 3*3 = 19
logger.info(s"dfExploded.count()=${dfExploded.count()}")

val dfOutput = dfExploded.groupBy("category").agg(
  collect_set("scalar_1").alias("combined_values_2"),
  collect_set("scalar_2").alias("combined_values_2"))

dfOutput.show()