Kafka AVRO - 从长到日期时间的转换答案

【问题标题】：Kafka AVRO - conversion from long to datetimeKafka AVRO - 从长到日期时间的转换
【发布时间】：2018-05-31 18:15:24
【问题描述】：

当我想发送包含 long 类型字段的 AVRO 消息时，出现以下错误：

Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 61
Caused by: java.lang.ClassCastException: java.lang.Long cannot be cast to org.joda.time.DateTime

我使用 Confluent 3.2.0 和 Apache Spark 2.2.0。在处理 AVRO 消息并在控制台中打印它们的 Spark 作业中会引发此错误。在 AVRO 模式中，对应的字段是这样定义的：

{\"name\": \"event_time\", \"type\": { \"type\" : \"long\", \"logicalType\": \"timestamp-millis\"}}

在.avsc文件生成的Java类中，字段定义如下：

private DateTime event_time;

【问题讨论】：

标签： apache-kafka deserialization avro confluent-platform

【解决方案1】：

我在使用 Confluent 4.0.0 和 Avro 1.8.2 时遇到了类似的问题。我有一个流处理器试图将 long 转换为 DateTime。我通过添加正确的转换克服了这个问题。在开始任何处理逻辑之前，我使用了Specific Data 静态实用程序类并添加了正确的逻辑类型转换。

SpecificData.get().addLogicalTypeConversion(new TimeConversions.TimestampConversion());

【讨论】：

你能证明你用过这个吗？
它需要在 Main 对象中使用，无论应用程序从哪里开始。

【解决方案2】：

在 Avro 1.9.X 及更高版本中，符号 TimestampConversion 不再存在。用 TimestampMillisConversion 替换上面的 @user3222582 为我修复了 avro 1.9.2 的编译错误

【讨论】：

【解决方案3】：

尝试将avro依赖中的avro版本升级到1.9.X，pom.xml中的avro-maven-plugin升级

<dependency>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro</artifactId>
    <version>1.9.1</version>
</dependency>

<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <version>1.9.1</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals>
                <goal>schema</goal>
                <!--<goal>idl-protocol</goal>-->
            </goals>
            <configuration>
                <sourceDirectory>${project.basedir}/src/main/avro</sourceDirectory>
                <outputDirectory>${project.basedir}/src/main/java</outputDirectory>
                <enableDecimalLogicalType>true</enableDecimalLogicalType>
                <stringType>String</stringType>
            </configuration>
        </execution>
    </executions>
</plugin>

还要确保从 AVRO 架构中删除以前生成的类并执行 mvn 编译。

【讨论】：