我们有很多parquet数据集,按年/月/日/小时划分。
只有一个_SUCCESS文件,其中一些时间是空的。
我们实现迭代所有分区并执行工作的作业。打开空数据集时遇到问题。org.apache.spark.sql.AnalysisException: Unable to infer schema for Parquet. It must be specified manually.
(使用EMR5.3.0 - 我们正在努力摆脱依赖,以便我们可以升级到更高版本)
Spark-shell示例:
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/_,_/_/ /_/_ version 2.1.0
/_/
Using Scala version 2.11.8 (OpenJDK 64-Bit Server VM, Java 1.8.0_121)
Type in expressions to have them evaluated.
Type :help for more information.
scala> val df = spark.read.parquet("s3://some-bucket/empty-dataset/")
org.apache.spark.sql.AnalysisException: Unable to infer schema for Parquet. It must be specified manually.;
at org.apache.spark.sql.execution.datasources.DataSource
$$ anonfun$8.apply(DataSource.scala:189) at org.apache.spark.sql.execution.datasources.DataSource $$
anonfun$8.apply(DataSource.scala:189)
正如spark所说,我可以指定一个模式,但这对我来说是不切实际的,因为模式很大并且每个分区不同(不要问)。
我首选的解决方案是,如果可以将spark配置为将空数据集推断为空模式。
添加schema就行了。没那么事多; 如: StructType schema = DataTypes .createStructType(Arrays.asList( DataTypes.createStructField("msisdn", DataTypes.StringType, true), DataTypes.createStructField("eid", DataTypes.IntegerType, true) )); sc.read().schema(schema).parquet("/test");
理想情况下spark.read.parquet应该采取param ignoreEmptyFiles=true,但这不受支持。
可以在读取之前过滤空目录
版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。