通过跨多个源的连接来激发ETL

Question

我有一个要求，加入一组3键使用火花数据帧50余个文件。我有驱动程序表，每天有100000条记录。我使用如下数据帧将这个表与53个其他文件连接起来。

val df1 = spark.read.parquet(<driver file>)
val df2 = spark.read.parquet(<right side file1>)
.
.
val df52 = spark.read.parquet(<right side file 52>)
//join
val refinedDF1 = df1.join(df2,Seq("key1","key2","key3"),"leftouter")).select(<some from left table>, <some from right table>)
val refinedDF2 = refinedDF1.join(df3,Seq("key1","key2","key3"),"leftouter")).select(<some from left table>, <some from right table>)
 .
 .
 so on for all 50 odd files
 refinedFinalDF.write.parquet(<s3 location>)

执行失败，出现错误

容器退出，退出代码为非零52

这基本上是一个内存不足的例外。我有一个相当大的集群，包含100000条记录的数据集。我有一个EMR，有12个执行器，每个16G，驱动器内存20G。

我尝试用df手动将数据帧划分为200个分区。以循环方式重新分区（200），这根本没有帮助。在连接键中，只有key1对于所有记录是不同的，key2和key3对于所有记录都是相同的值。是否有任何优化可以使其工作？我试图保存的最后一个数据帧中有140列。如果驱动程序表有n条记录，那么在每个左外之后，我只会有n条。

更新：我已经尝试创建一个较小的数据框的驱动程序表与限制（100），我仍然得到内存溢出异常。

Answer 1

我有一个类似的情况，我有多个联接，最后我必须将最终数据帧写入 HDFS/Hive 表（拼花格式）。

Spark基于惰性执行机制工作，这意味着，当您的第53个数据帧被操作(保存/写入为Parquet)时，Spark会返回到所有的连接并执行它们，这会导致数据的巨大混乱，最终您的作业的容器会失败并抛出内存不足错误。

建议：您可以先将每个加入的数据帧写入HDFS，我的意思是一旦您加入了2个（可以超过2个，但保持限制）数据帧，将加入的数据帧写入HDFS/Hive并使用选择*'hive拼花表

val refinedDF1 = df1.join(df2 ,condition,'join_type')
refinedDF1.write.parquet("location") or refinedDF1.write.mode("overwrite").saveAsTable("dbname.refine1")
val refinedDF1 = hc.sql("select * from dbname.refine1")

val refinedDF2 = refinedDF1.join(df3)
refinedDF2.write.parquet("location") or refinedDF1.write.mode("overwrite").saveAsTable("dbname.refine2")
val refinedDF2 = hc.sql("select * from dbname.refine2")

现在你经常将连接写入 hdfs，这意味着 Spark 在调用最终连接时不必执行它们，它只需要使用你以表形式保存的 52'nd 连接输出。

通过使用这种方法，我的脚本从22小时（包括容器内存错误）下降到15到30分钟（没有内存异常/错误）。

小提示：

1） 排除join＜code＞键＜/code＞为空的记录，spark与具有＜code＞null＝null＜/code>条件的join相比性能不佳，因此在连接数据帧之前将其删除

2)当左侧数据帧是多行，右侧数据帧是一个查找或几行时，使用广播连接。

3)在脚本执行后，您必须清理保存在Hive/Hdfs中的中间数据帧。

Answer 2

碰巧的是，我用来创建数据框的s3存储桶中的底层数据有多个文件夹，我正在过滤特定文件夹作为过滤器的一部分。例如：spark.read.parquet（s3存储桶）。过滤器（'folder_name="val"）。看起来火花正在将s3存储桶中的所有数据加载到执行器内存中，然后运行过滤器。这就是为什么它被轰炸了，因为相同的逻辑运行在蜂巢外部表上运行，指向s3位置的文件夹，因为分区列工作得很好。我不得不删除过滤器并读取特定文件夹来解决问题…spark.read.parquet（s3存储桶/文件夹=值）…

Answer 3

你们的桌子是一对一还是一对一？如果它们是一对多，那么您的联接将产生比您可能想要的更多的行。如果是这种情况，一个选项是首先对要加入的每个表执行groupBy。考虑这个例子：

val df1 = Seq(1, 2).toDF("id")
val df2 = Seq(
  (1, "a", true),
  (1, "b", false),
  (2, "c", true)
).toDF("id", "C2", "B2")

val df3 = Seq(
  (1, "x", false),
  (1, "y", true),
  (2, "z", false)
).toDF("id", "C3", "B3")

// Left outer join without accounting for 1-Many relationship.  Results in cartesian
// joining on each ID value!
df1.
  join(df2, Seq("id"), "left_outer").
  join(df3, Seq("id"), "left_outer").show()

+---+---+-----+---+-----+
| id| C2|   B2| C3|   B3|
+---+---+-----+---+-----+
|  1|  b|false|  y| true|
|  1|  b|false|  x|false|
|  1|  a| true|  y| true|
|  1|  a| true|  x|false|
|  2|  c| true|  z|false|
+---+---+-----+---+-----+

或者，如果在连接之前对行进行分组，使关系始终为1-1，则不会添加记录

val df2Grouped = df2.groupBy("id").agg(collect_list(struct($"C2", $"B2")) as "df2")
val df3Grouped = df3.groupBy("id").agg(collect_list(struct($"C3", $"B3")) as "df3")

val result = df1.
  join(df2Grouped, Seq("id"), "left_outer").
  join(df3Grouped, Seq("id"), "left_outer")
result.printSchema
result.show(10, false)

scala> result.printSchema
root
 |-- id: integer (nullable = false)
 |-- df2: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- C2: string (nullable = true)
 |    |    |-- B2: boolean (nullable = false)
 |-- df3: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- C3: string (nullable = true)
 |    |    |-- B3: boolean (nullable = false)


scala> result.show(10, false)
+---+-----------------------+-----------------------+
|id |df2                    |df3                    |
+---+-----------------------+-----------------------+
|1  |[[a, true], [b, false]]|[[x, false], [y, true]]|
|2  |[[c, true]]            |[[z, false]]           |
+---+-----------------------+-----------------------+