Spark 1.6中Avro读取器分区发现功能失效问题排查

我之前也踩过Spark 1.6里Avro分区发现的坑，确实和Parquet的体验差很多。核心原因是Spark 1.6的Avro数据源（当时主要依赖Databricks的spark-avro包）并没有实现和Parquet一样的自动分区列解析逻辑——Parquet是Spark原生深度支持的格式，分区发现是内置功能，但Avro在这个版本里还没跟上这个特性。

下面是几种可行的解决办法：

1. 手动提取分区列（快速临时方案） #

利用input_file_name()函数获取每个数据文件的路径，再通过正则表达式提取出year（以及month、date）的值，转换成对应的列类型：

Scala版本 #

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.IntegerType

// 读取Avro文件
val avroDF = sqlContext.read.format("com.databricks.spark.avro").load("/your/data/path")

// 从文件路径中提取year列
val dfWithYear = avroDF.withColumn(
  "year",
  regexp_extract(input_file_name(), "year=(\\d+)", 1).cast(IntegerType)
)
// 同理提取month和date
val finalDF = dfWithYear.withColumn(
  "month",
  regexp_extract(input_file_name(), "month=(\\d+)", 1).cast(IntegerType)
).withColumn(
  "date",
  regexp_extract(input_file_name(), "date=(\\d+)", 1).cast(IntegerType)
)

Python版本 #

from pyspark.sql.functions import regexp_extract, input_file_name

avro_df = sqlContext.read.format("com.databricks.spark.avro").load("/your/data/path")

# 提取year、month、date列
final_df = avro_df.withColumn("year", regexp_extract(input_file_name(), r"year=(\d+)", 1).cast("int")) \
                  .withColumn("month", regexp_extract(input_file_name(), r"month=(\d+)", 1).cast("int")) \
                  .withColumn("date", regexp_extract(input_file_name(), r"date=(\d+)", 1).cast("int"))

2. 通过Hive表映射（长期规范方案） #

如果数据是长期存储的，推荐用Hive外部表来管理分区，Spark通过HiveContext读取时会自动加载分区列：

第一步：在Hive中创建外部表 #

CREATE EXTERNAL TABLE avro_partitioned_table (
  -- 填写你的Avro数据列定义，示例：
  id int,
  content string
)
PARTITIONED BY (year int, month int, date int)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.avro.AvroSerDe'
STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat'
LOCATION '/your/data/path';

第二步：同步Hive分区（匹配目录中的分区） #

MSCK REPAIR TABLE avro_partitioned_table;

第三步：在Spark中读取Hive表 #

// 确保使用HiveContext
val hiveCtx = new org.apache.spark.sql.hive.HiveContext(sc)
val df = hiveCtx.table("avro_partitioned_table")

这样year、month、date列就会自动出现在DataFrame中，和Parquet的体验完全一致。

3. 升级Spark版本（彻底解决） #

如果业务允许的话，升级到Spark 2.0及以上版本是最彻底的方案——后续版本的spark-avro包已经完善了分区自动发现功能，和Parquet的行为完全一致，无需额外操作就能自动识别year=xxx/month=xxx/date=xxx格式的分区列。

内容的提问来源于stack exchange，提问作者Shankar