如何在 Apache Spark 中处理不断变化的镶木地板架构答案

【问题标题】：How to handle changing parquet schema in Apache Spark如何在 Apache Spark 中处理不断变化的镶木地板架构
【发布时间】：2017-04-17 01:43:03
【问题描述】：

我遇到了一个问题，我在 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 原始数据，我最初的计划是创建一个自动作业，它启动一个 EMR 集群，读取上一个日期的 JSON 数据，然后简单地将其作为 parquet 写回 S3。

JSON 数据也分为日期，即键具有日期前缀。读取 JSON 工作正常。无论当前正在读取多少数据，都会从数据中推断出架构。

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

当某些应该是double 类型的列在给定日期只有整数值时，从 JSON 中读取它们（这些数字是整数，没有浮点数）使 Spark 认为它是一个具有类型的列long。即使我可以在编写 Parquet 文件之前将这些列加倍，这仍然不好，因为架构可能会更改，可以添加新列，并且无法跟踪。

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

对此有哪些最佳做法或解决方案？

【问题讨论】：

标签： apache-spark apache-spark-sql spark-dataframe emr parquet

【解决方案1】：

这些是我用于将 parquet 写入 S3 的选项；关闭模式合并可以提高写回性能 - 它也可以解决您的问题

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

【讨论】：

嗨，我会试试这个。但我想知道，在编写镶木地板时，您是否将这个 PARQUET_OPTIONS 映射包含在 .option() 函数中？但是读书呢？我只使用了sqlContext.read.option("mergeSchema", true).parquet("path")，但仍然使用不带选项的常规写入。
好吧，我尝试使用.option("mergeSchema", "false").option("filterPushdown","true") 进行读写，但并没有改变任何事情。当 mergeSchema 为真时，我得到Failed to merge incompatible data types DoubleType and LongType，当它为假时，读取数据有效。打印模式显示该列是双精度类型，show() 命令显示前 20 行，但对该列的过滤和分组失败：Cost's declared type (java.lang.Double) does not match the schema found in file metadata.
听起来您的某些文件的架构不一致
Slack 最近发布了on this topic。他们的一个关键主题是“从不删除字段，只在末尾添加它们”。毕竟，如果你标记一个字段是可选的，你可以省略数据。就像我说的，这不是我工作的领域。我所知道的是模式合并需要 parquet 来读取每个文件末尾的模式，这很昂贵，尤其是在 Hadoop
是的，当前合并没有帮助，因为它无法选择一种类型并使用它。它唯一做的就是在读取架构不匹配时抛出异常。如果没有它，我们稍后会在执行 groupBy 或类似的操作时遇到错误。我们的 GoLang 服务器中有数据结构，这些数据结构最终在 S3 中以 JSON 文件的形式出现。添加新字段是最常见的选项（但不是最后，因为可以在嵌套结构等中添加一些字段），但我认为也可以删除一些字段。类型可能没有改变，但如果数据丢失，spark 会以不同的方式解释类型

【解决方案2】：

当我从 JSON 中读取每日块中的数据并在每日 S3 文件夹中写入 Parquet 时，在读取 JSON 或在写入 Parquet 之前将容易出错的列转换为正确类型时没有指定我自己的架构，Spark 可能会推断出不同的架构不同天的数据取决于数据实例中的值，并写入具有冲突模式的 Parquet 文件。

这可能不是完美的解决方案，但我发现使用不断发展的架构解决问题的唯一方法如下：

在我每天（更具体地说是每晚）批处理前一天数据的 cron 作业之前，我正在创建一个大部分为空值的虚拟对象。

我确保 ID 是可识别的，例如，由于真实数据具有唯一的 ID，我将“虚拟”字符串作为 ID 添加到虚拟数据对象中。

然后我将为容易出错的类型的属性提供预期值，例如我将提供非零值的浮点数/双精度值，因此当编组为 JSON 时，它们肯定会有小数分隔符，例如“0.2”而不是“ 0”（编组为 JSON 时，具有 0 值的双精度/浮点数显示为“0”而不是“0.0”）。

字符串、布尔值和整数工作正常，但除了双精度/浮点数之外，我还需要将数组实例化为空数组和其他类/结构的对象以及相应的空对象，这样它们就不会是“null”-s，因为 Spark 将 null-s 作为字符串读取。

然后，如果我填写了所有必需的字段，我会将对象编组为 JSON 并将文件写入 S3。

然后我会在我的 Scala 批处理脚本中使用这些文件来读取它们，将模式保存到一个变量中，并在我读入真实的 JSON 数据时将此模式作为参数提供，以避免 Spark 进行自己的模式推断。

这样我就知道所有字段始终属于同一类型，并且模式合并仅在添加新字段时才需要加入模式。

当然，当添加易出错类型的新字段时，它会增加手动更新虚拟对象创建的缺点，但目前这是一个小缺点，因为它是我发现的唯一可行的解决方案。

【讨论】：

这比在 Spark 处理期间自定义解析数据（即读取字符串、使用 json4s 解析和构建您需要的案例类）更好吗？
@Sim 好问题。我想尽可能地自动化解决方案。由于模式几乎每天都在发展，因此构建案例类效率不高。原始数据结构在 Go 应用程序中定义，涉及至少 200-300 个或更多数据属性。向原始结构添加新字段时，也应将新字段添加到案例类中。但如果是由不同的人来做，就很难跟踪和维护了。
我无法理解您如何将数据写入 s3。您是否正在与虚拟对象和传入数据进行 rdd union 并存储它？
@Avishek 我的真实数据来自 Firehose 并存储在 S3 中。我的虚拟数据刚刚写入/上传到另一个位置的 S3。然后在读取所有数据时，我将虚拟文件的路径与真实数据的路径一起包括在内。然后我只有一行虚拟数据，它为整个数据集提供了正确的模式。后来我只需要过滤掉它或以其他方式删除它。
@Avishek 我的解决方案是为了解决不兼容的类型问题。我们的数据类型是在 Go 结构中定义的。所以最新的虚拟数据只有在有字符串类型属性的地方才有字符串，对于数组，虚拟数据有预定义的虚拟数组。因此，实际数据可能具有字符串值和实际数组类型的数组值，但没有字符串类型的属性的数组值。实际上唯一的选择是向结构中添加了一个新的数组类型属性，但不是虚拟对象的正确值，那么它会失败，是的，没有检查。

【解决方案3】：

只需创建一个 rdd[String]，其中每个字符串都是一个 json，当将 rdd 作为数据框时，使用primitiveAsString 选项将所有数据类型设置为字符串

 val binary_zip_RDD = sc.binaryFiles(batchHolder.get(i), minPartitions = 50000)
 // rdd[String]  each string is a json ,lowercased json
    val TransformedRDD = binary_zip_RDD.flatMap(kv => ZipDecompressor.Zip_open_hybrid(kv._1, kv._2, proccessingtimestamp))
 // now the schema of dataframe would be consolidate schema of all json strings
    val jsonDataframe_stream = sparkSession.read.option("primitivesAsString", true).json(TransformedRDD)

    println(jsonDataframe_stream.printSchema())


    jsonDataframe_stream.write.mode(SaveMode.Append).partitionBy(GetConstantValue.DEVICEDATE).parquet(ApplicationProperties.OUTPUT_DIRECTORY)

【讨论】：