Druid Overlord Hadoop任务中Parquet INT96时间戳列未实现问题求助

我最近在使用Druid Overlord向EMR提交MapReduce索引任务时碰到了一个棘手的问题：数据源是S3上的Parquet文件，其中包含一个INT96类型的时间戳列，而AvroSchema并不支持这种类型，导致解析时间戳时直接抛出了异常。

错误堆栈信息 #

Error: java.lang.IllegalArgumentException: INT96 not yet implemented.
at org.apache.parquet.avro.AvroSchemaConverter$1.convertINT96(AvroSchemaConverter.java:279)
at org.apache.parquet.avro.AvroSchemaConverter$1.convertINT96(AvroSchemaConverter.java:264)
at org.apache.parquet.schema.PrimitiveType$PrimitiveTypeName$7.convert(PrimitiveType.java:223)

环境信息 #

• Druid版本：0.11
• EMR版本：emr-5.11.0
• Hadoop版本：Amazon 2.7.3

提交的Druid索引任务配置JSON #

{
    "type": "index_hadoop",
    "spec": {
        "ioConfig": {
            "type": "hadoop",
            "inputSpec": {
                "type": "static",
                "inputFormat": "io.druid.data.input.parquet.DruidParquetInputFormat",
                "paths": "s3://s3_path"
            }
        },
        "dataSchema": {
            "dataSource": "parquet_test1",
            "granularitySpec": {
                "type": "uniform",
                "segmentGranularity": "DAY",
                "queryGranularity": "ALL",
                "intervals": ["2017-08-01T00:00:00/2017-08-02T00:00:00"]
            },
            "parser": {
                "type": "parquet",
                "parseSpec": {
                    "format": "timeAndDims",
                    "timestampSpec": {
                        "column": "t",
                        "format": "yyyy-MM-dd HH:mm:ss:SSS zzz"
                    },
                    "dimensionsSpec": {
                        "dimensions": [ "dim1","dim2","dim3" ],
                        "dimensionExclusions": [],
                        "spatialDimensions": []
                    }
                }
            },
            "metricsSpec": [{
                "type": "count",
                "name": "count"
            },{
                "type" : "count",
                "name" : "pid",
                "fieldName" : "pid"
            }]
        },
        "tuningConfig": {
            "type": "hadoop",
            "partitionsSpec": {
                "targetPartitionSize": 5000000
            },
            "jobProperties" : {
                "mapreduce.job.user.classpath.first": "true",
                "fs.s3.awsAccessKeyId" : "KEYID",
                "fs.s3.awsSecretAccessKey" : "AccessKey",
                "fs.s3.impl" : "org.apache.hadoop.fs.s3native.NativeS3FileSystem",
                "fs.s3n.awsAccessKeyId" : "KEYID",
                "fs.s3n.awsSecretAccessKey" : "AccessKey",
                "fs.s3n.impl" : "org.apache.hadoop.fs.s3native.NativeS3FileSystem",
                "io.compression.codecs" : "org.apache.hadoop.io.compress.GzipCodec,org.apache.hadoop.io.compress.DefaultCodec,org.apache.hadoop.io.compress.BZip2Codec,org.apache.hadoop.io.compress.SnappyCodec"
            },
            "leaveIntermediate": true
        }
    },
    "hadoopDependencyCoordinates": ["org.apache.hadoop:hadoop-client:2.7.3", "org.apache.hadoop:hadoop-aws:2.7.3", "com.hadoop.gplcompression:hadoop-lzo:0.4.20"]
}

可行的解决方案 #

经过排查，我整理出两个可行的解决方向：

方案1：绕开Avro转换，直接处理Parquet数据 #

调整数据处理链路，直接以Parquet格式进行存储和读取，完全避免将Parquet转换为Avro的步骤。这样既可以消除对AvroSchema的依赖，从根源上解决INT96类型不支持的问题，还能保留Parquet本身的高效存储优势，是比较推荐的轻量化方案。

方案2：扩展AvroSchemaConverter支持INT96类型 #

如果必须依赖Avro转换，可以修改org.apache.parquet.avro.AvroSchemaConverter类中的convertINT96方法，添加对INT96时间戳的解析逻辑——将INT96格式的时间戳转换为Avro支持的类型（比如转为long型时间戳或者格式化后的字符串）。需要注意的是，这种方式需要自定义Parquet-Avro转换的代码，并确保修改后的类能被Druid和EMR的任务环境正确加载。

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