apache-spark - 如何在Apache Spark中处理更改 Parquet 架构

Question

我遇到了一个问题，我在S3中将Parquet数据作为日常数据块(以s3://bucketName/prefix/YYYY/MM/DD/的形式)，但是由于某些列类型不匹配，并且无法获得AWS EMR Spark中的数据，因此出现了许多异常之一， 例如:

java.lang.ClassCastException: optional binary element (UTF8) is not a group

当在某些文件中存在具有值的数组类型，但在其他文件中同一列可能具有 null值，然后将其推断为字符串类型时，会出现。

要么

org.apache.spark.SparkException: Job aborted due to stage failure: Task 23 in stage 42.0 failed 4 times, most recent failure: Lost task 23.3 in stage 42.0 (TID 2189, ip-172-31-9-27.eu-west-1.compute.internal):
org.apache.spark.SparkException: Failed to merge incompatible data types ArrayType(StructType(StructField(Id,LongType,true), StructField(Name,StringType,true), StructField(Type,StringType,true)),true)

我有JSON格式的S3原始数据，最初的计划是创建一个自 Action 业，该作业将启动EMR集群，读取上一个日期的JSON数据，然后将其作为拼写形式写回到S3。

JSON数据也分为日期，即键具有日期前缀。读取JSON效果很好。无论当前正在读取多少数据，都可以从数据推断模式。

但是，当写入 Parquet 文件时，问题就增加了。据我了解，当我用元数据文件编写 Parquet 时，这些文件包含 Parquet 文件所有部分/分区的架构。在我看来，也可以使用不同的模式。当我禁用编写元数据时，据说Spark从给定的Parquet路径中的第一个文件推断出整个架构，并假设其他文件中的架构保持不变。

当某些列(应为 double类型)在给定日期仅具有整数值时，从JSON中读取这些值(这些数字为整数，没有浮点数)，Spark会认为它是 long类型的列。即使我可以在编写Parquet文件之前将这些列转换为双倍，但这也不是一件好事，因为架构可能会更改，可以添加新列，并且无法对此进行跟踪。

我已经看到有些人有同样的问题，但是我还没有找到足够好的解决方案。

有哪些最佳实践或解决方案？

Answer 1

这些是我用来将 Parquet 写到S3的选项。关闭架构合并可提高回写性能-它也可以解决您的问题

val PARQUET_OPTIONS = Map(
 "spark.sql.parquet.mergeSchema" -> "false",
 "spark.sql.parquet.filterPushdown" -> "true")