java - 无法通过Spark 1.6从Parquet Hive表中读取数据

Question

我正在尝试从以Parquet格式存储的配置单元表中读取数据。我正在使用MapR发行版。读取数据后，当我尝试执行任何操作时，例如df.show(3)，它将引发java.lang.ArrayIndexOutOfBoundsException:7.如果表存储更改为ORC，则它将起作用。

而且，我正在尝试从共享集群中的表中读取。因此，我无法在源表中进行任何更改。

Hive表结构，

CREATE TABLE employee_p(
  emp_id bigint,
  f_name string,
  l_name string,
  sal double)
ROW FORMAT DELIMITED
  FIELDS TERMINATED BY '\u0001'
STORED AS INPUTFORMAT
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
OUTPUTFORMAT
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
  'maprfs:/user/hive/warehouse/sptest.db/employee_p'
TBLPROPERTIES (
  'COLUMN_STATS_ACCURATE'='true',
  'numFiles'='1',
  'numRows'='4',
  'rawDataSize'='16',
  'totalSize'='699',
  'transient_lastDdlTime'='1550203019')

Java代码

    String warehouseLocation = args[0];
    String query1 = "select emp_id, f_name, l_name, sal from sptest.employee_p";

    SparkConf conf = new SparkConf().setAppName("Parquet Table");
    JavaSparkContext jsc = new JavaSparkContext(conf);
    HiveContext hc = new HiveContext(jsc);

    DataFrame df = hc.sql(query1);

    df.printSchema();
    df.show(10);

作业提交命令，

    $SPARK_HOME/bin/spark-submit --class com.app.hive.FetchFromParquetTable \
    ${APP_HOME}/SparkTest-0.0.1-SNAPSHOT.jar maprfs:/user/hive/warehouse \
    --master yarn --deploy-mode cluster \
    --conf "spark.sql.parquet.writeLegacyFormat=true" \
    --conf "spark.sql.parquet.filterPushdown=false" \
    --queue myqueue

异常(exception)，

19/02/14 21:08:23 WARN TaskSetManager: Lost task 0.0 in stage 1.0 (TID 2, lgpbd1523.gso.aexp.com): java.lang.ArrayIndexOutOfBoundsException: 7
        at org.apache.parquet.bytes.BytesUtils.bytesToLong(BytesUtils.java:250)
        at org.apache.parquet.column.statistics.LongStatistics.setMinMaxFromBytes(LongStatistics.java:50)
        at org.apache.parquet.format.converter.ParquetMetadataConverter.fromParquetStatistics(ParquetMetadataConverter.java:255)
        at org.apache.parquet.format.converter.ParquetMetadataConverter.fromParquetMetadata(ParquetMetadataConverter.java:550)
        at org.apache.parquet.format.converter.ParquetMetadataConverter.readParquetMetadata(ParquetMetadataConverter.java:527)
        at org.apache.parquet.hadoop.ParquetFileReader.readFooter(ParquetFileReader.java:430)
        at org.apache.parquet.hadoop.ParquetFileReader.readFooter(ParquetFileReader.java:385)
        at org.apache.parquet.hadoop.ParquetRecordReader.initializeInternalReader(ParquetRecordReader.java:157)
        at org.apache.parquet.hadoop.ParquetRecordReader.initialize(ParquetRecordReader.java:140)
        at org.apache.spark.rdd.SqlNewHadoopRDD$$anon$1.<init>(SqlNewHadoopRDD.scala:180)
        at org.apache.spark.rdd.SqlNewHadoopRDD.compute(SqlNewHadoopRDD.scala:126)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:270)

Answer 1

我猜您在读取表时遇到了这个问题，您可以从spark UI进行确认，因为当您触发一个实际上开始读取表的操作时spark是惰性的，并且遇到了此问题。

我在阅读配置单元和带有MapR分布的spark 2.1.0版本的 hive ， Parquet 和活泼的压缩时创建的表时遇到了同样的问题。

您可以尝试将emp_id数据类型从bigint更改为String吗？