0639-6.1.1-Spark读取由Impala创建的Parquet文件异常分析

作者：冉南阳

1

问题重现

• 测试环境：

1.RedHat7.4

2.CDH6.1.1

3.使用root进行操作

1.使用Impala创建Parquet表并插入数据。

create table test_parquet(id int,name string) stored as parquet;
insert into test_parquet values(1,'test'),(2,'vivi');

2.使用Impala查看数据，数据显示正常。

select * from test_parquet;

3.使用CDH6.1.1中的Spark2.4访问该数据文件。

val df=spark.read.parquet("hdfs://ip-172-31-6-83.ap-southeast-1.compute.internal:8020/user/hive/warehouse/test_parquet")
df.show()

发现name这个字段string类型显示异常，问题重现。

2

问题分析

1.直接在Spark CLI分析一下该Parquet文件的schema。

df.printSchema()

应该为String，实际为二进制binary。

2.通过CDH提供的parquet tool进行分析，参考《​​0631-6.2-如何确认一个Parquet文件是否被压缩​​》。

hadoop fs -ls /user/hive/warehouse/test_parquet
hadoop fs -get /user/hive/warehouse/test_parquet/8644d8aba29fa4cf-a9a89e1900000000_325265083_data.0.parq .
/opt/cloudera/parcels/CDH/lib/parquet/bin/parquet-tools schema -d 8644d8aba29fa4cf-a9a89e1900000000_325265083_data.0.parq

可以发现应该为String，实际为二进制binary。

这是因为Hive/Impala与Spark在Parquet的实现上不一致，Hive/Impala将string类型在Parquet文件中保存为二进制binary，它们查询的时候再进行解析。但Spark的代码查询的时候却没有这样做，而是直接将二进制值查询并展现出来，所以Spark中有一个参数spark.sql.parquet.binaryAsString，默认为false，解释如下：

由其他系统生成的Parquet文件，特别是Impala，Hive和旧版本的Spark SQL，在写Parquet文件的schema时候不区分字符串和二进制。这个参数是告诉Spark SQL将二进制数据解释为字符串，从而保证Spark与其他系统比如Hive或Impala的兼容性。

参考：

​​https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#configuration​​

3

问题解决

3.1

方法1

直接采用Spark SQL来读取，而不是Spark代码来读取Parquet文件。

1.使用以下语句直接读取Impala创建的这张表的数据。

spark.sql("select * from test_parquet").show()

发现name字段查询显示正常。

3.2

方法2

通过Spark读取Parquet文件时定义schema

1.首先在Spark Shell中定义schema

import org.apache.spark.sql.types._
val columnsList=List(
| StructField("id",IntegerType),
| StructField("name",StringType)) 
val testScheme=StructType(columnsList)

2.使用该schema再去读取之前的Parquet文件。

val df=spark.read.schema(testScheme).parquet("hdfs://ip-172-31-6-83.ap-southeast-1.compute.internal:8020/user/hive/warehouse/test_parquet")
df.show()

显示正常，问题解决。

3.3

方法3

启动spark-shell的时候带上启动参数

1.使用以下参数重新启动spark-shell

spark-shell --conf spark.sql.parquet.binaryAsString=true

2.再次用同样的代码读取之前的Parquet文件。

val df=spark.read.parquet("hdfs://ip-172-31-6-83.ap-southeast-1.compute.internal:8020/user/hive/warehouse/test_parquet")
df.show()

显示正常，问题解决。

4

问题总结

1.使用Impala创建的Parquet文件，如果包含字符串类型，由Spark代码直接读取该Parquet文件时会显示异常，将字符串的值显示为二进制binary。

2.主要原因是因为由其他系统生成的Parquet文件，特别是Impala，Hive和旧版本的Spark SQL，在写Parquet文件的schema时候不区分字符串和二进制。所以Spark提供了一个参数spark.sql.parquet.binaryAsString来解决该问题，参考：

​​https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#configuration​​

3.对于该问题的解决方案有三种，具体可以参考第三个章节：

    a)直接采用Spark SQL来读取，而不是Spark代码来读取Parquet文件。

    b)通过Spark读取Parquet文件时定义schema

    c)启动spark-shell的时候带上启动参数