如何加载混合Parquet模式到DataFrame使用Apache火花？

提问者：小点点

如何加载混合Parquet模式到DataFrame使用Apache火花？

我有一个火花作业不断上传Parquet文件到S3（分区）。
这些文件都有相同的拼花模式。

其中一种字段类型最近已更改（从String更改为long），因此某些分区的parquet模式是混合的。

虽然我似乎可以执行：sqlContext.read. load（path）
当尝试在DataFrame上应用任何获取操作时（例如收集），该操作失败，ParquetDecodingException

我打算迁移数据并重新格式化它，但无法将混合内容读取到DataFrame中。
如何使用Apache Spark将混合分区加载到DataFrames或任何其他Spark构造中？

以下是ParquetDecodingException跟踪：

scala> df.collect
[Stage 1:==============>        (1 + 3) / 4]
WARN TaskSetManager: Lost task 1.0 in stage 1.0 (TID 2, 172.1.1.1, executor 0): org.apache.parquet.io.ParquetDecodingException: 
Can not read value at 1 in block 0 in file 
s3a://data/parquet/partition_by_day=20180620/partition_by_hour=10/part-00000-6e4f07e4-3d89-4fad-acdf-37054107dc39.snappy.parquet
    at org.apache.parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:243)
    at org.apache.parquet.hadoop.ParquetRecordReader.nextKeyValue(ParquetRecordReader.java:227)
    at org.apache.spark.sql.execution.datasources.RecordReaderIterator.hasNext(RecordReaderIterator.scala:39)
    at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:102)
    at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.nextIterator(FileScanRDD.scala:166)
    at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:102)
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
    at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
    at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:377)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:231)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:225)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:826)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:826)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
    at org.apache.spark.scheduler.Task.run(Task.scala:99)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:282)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ClassCastException: [B cannot be cast to java.lang.Long
    at scala.runtime.BoxesRunTime.unboxToLong(BoxesRunTime.java:105)

共2个答案

匿名用户

据我所知，您不能将具有相同字段和不同类型的2个模式混合在一起。因此，我能想到的唯一解决方案是：

列出分区的文件

将每个文件重写到新位置并将数据转换到右侧schame

匿名用户

还有另一个想法：不是改变现有字段的类型（field_string），而是添加一个长类型的新字段（field_long），并将读取数据的代码更新为类似这样的代码（在伪代码中）并启用模式合并。我相信它默认启用，但这是一个明确说明它的好案例：

sqlContext.read.option("mergeSchema", "true").parquet(<parquet_file>)

...

if isNull(field_long) 
  field_value_long = field_string.value.to_long
else
  field_value_long = field_long.value