如何使用 Java 在 Spark 1.3.1 中读取 AVRO 数据？答案

【问题标题】：How do I use Java to read AVRO data in Spark 1.3.1?如何使用 Java 在 Spark 1.3.1 中读取 AVRO 数据？
【发布时间】：2015-12-22 08:30:38
【问题描述】：

我正在尝试开发一个 Java Spark 应用程序，该应用程序通过一种称为 Gobblin (https://github.com/linkedin/gobblin/wiki) 的技术从 HDFS 读取 AVRO 记录 (https://avro.apache.org/)。

一个示例 HDFS AVRO 数据文件：

/gobblin/work/job-output/KAFKA/kafka-gobblin-hdfs-test/20150910213846_append/part.task_kafka-gobblin-hdfs-test_1441921123461_0.avro

不幸的是，我发现用 Java 编写的示例非常有限。

我发现最好的东西是用 Scala 编写的（使用 Hadoop 版本 1 库）。

https://gist.github.com/MLnick/5864741781b9340cb211

任何帮助将不胜感激。

目前我正在考虑使用以下代码，但我不确定如何从我的 AVRO 数据中提取值的 HashMap：

JavaPairRDD avroRDD = sc.newAPIHadoopFile( 
    path, 
    AvroKeyInputFormat.class, 
    AvroKey.class, 
    NullWritable.class, 
    new Configuration() );

// JavaPairRDD avroRDD = sc.newAPIHadoopFile( 
//    path, 
//    AvroKeyValueInputFormat.class, 
//    AvroKey.class, 
//    AvroValue.class, 
//    new Configuration() );

我当前的 Maven 依赖项：

<dependencies>

    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.10</artifactId>
        <version>1.3.1</version>
    </dependency>

    <dependency>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro</artifactId>
        <version>1.7.6</version>
    </dependency>
    <dependency>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro-mapred</artifactId>
        <version>1.7.6</version>
        <classifier>hadoop2</classifier>
    </dependency>
    <dependency>
      <groupId>com.fasterxml.jackson.core</groupId>
      <artifactId>jackson-annotations</artifactId>
      <version>2.4.3</version>
    </dependency>


    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <scope>provided</scope>
    </dependency>

    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <scope>provided</scope>
    </dependency>

    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <scope>test</scope>
    </dependency>

</dependencies>

【问题讨论】：

为什么不使用spark-avro (v. 1.0.0)？像这样的东西应该可以工作：HashMap<String, String> options = new HashMap<String, String>(); options.put("path", path); DataFrame df = sqlContext.load("com.databricks.spark.avro", options);
理想情况下只想针对标准 Spark Java API 编写代码。所以除了 Avro 依赖之外，我不想使用任何外部库。
在名为“Hadoop 应用程序架构”Hadoop Application Architectures JavaSessionize Example 的书中的另一个有用示例，但是，我宁愿不必编写“*.avsc”文件并依赖 avro-maven-plugin 来生成所需的类文件。

标签： java apache-spark hdfs avro gobblin

【解决方案1】：

我编写了一个小型原型，它能够读取我的示例 Gobblin Avro 记录作为输入，并使用 Spark 输出相关结果 (spark-hdfs-avro-test)。值得一提的是，我需要解决几个问题。 任何 cmets 或反馈将不胜感激。

问题 1：当前 Avro 版本 (1.7.7) 和 Java 序列化存在问题：

引用：

Spark 依赖于 Java 的 Serializable 接口来序列化对象。 Avro 对象不实现 Serializable。因此，要在 Spark 中使用 Avro 对象，您需要对 Avro 生成的类进行子类化并实现 Serializable，例如https://github.com/massie/spark-parquet-example/blob/master/src/main/scala/com/zenfractal/SerializableAminoAcid.java.

为了解决这个问题，我编写了自己的 Serializable 包装类：

问题 2：我的 Avro 消息不包含“密钥”值。

很遗憾，我无法使用任何开箱即用的输入格式，不得不自己编写：AvroValueInputFormat

public class AvroValueInputFormat<T> extends FileInputFormat<NullWritable, AvroValue<T>> {

我无法使用以下内容：

# org.apache.avro.mapreduce.AvroKeyInputFormat
public class AvroKeyInputFormat<T> extends FileInputFormat<AvroKey<T>, NullWritable> {

# org.apache.avro.mapreduce.AvroKeyValueInputFormat
public class AvroKeyValueInputFormat<K, V> extends FileInputFormat<AvroKey<K>, AvroValue<V>> {

问题 3：我无法使用 AvroJob 类设置器来设置架构值，我必须手动执行此操作。

    hadoopConf.set( "avro.schema.input.key", Schema.create( org.apache.avro.Schema.Type.NULL ).toString() ); //$NON-NLS-1$
    hadoopConf.set( "avro.schema.input.value", Event.SCHEMA$.toString() ); //$NON-NLS-1$
    hadoopConf.set( "avro.schema.output.key", Schema.create( org.apache.avro.Schema.Type.NULL ).toString() ); //$NON-NLS-1$
    hadoopConf.set( "avro.schema.output.value", SeverityEventCount.SCHEMA$.toString() ); //$NON-NLS-1$

【讨论】：