Spark Streaming使用fileStream处理内嵌Schema的Avro数据参数咨询

嘿，我来帮你理清用fileStream()处理带内嵌Schema的Avro数据时，三个参数该怎么选！你之前看到的Parquet相关方案和这个逻辑类似，但Avro有专门的InputFormat实现，所以参数选择上有区别，咱们一步步说清楚~

1. KeyClass：NullWritable #

• 对于Avro文件来说，我们通常不需要用到Hadoop InputFormat中的key字段（Avro的结构化数据都存在value里），所以直接用org.apache.hadoop.io.NullWritable就好，它表示这个位置没有实际有用的数据。

2. ValueClass：GenericRecord（或自定义Avro生成类） #

• 因为你的Avro数据内嵌了Schema，不需要提前编译生成对应的Java/Scala类，所以用org.apache.avro.generic.GenericRecord最合适——它可以动态读取内嵌的Schema，并让你通过字段名获取对应的值。
• 如果你已经用Avro工具（比如avro-tools）生成了对应数据结构的Java/Scala类，那也可以直接把这个类作为ValueClass，这样操作数据时能获得类型安全的支持。

3. InputFormatClass：AvroInputFormat #

• 必须选择org.apache.avro.mapreduce.AvroInputFormat（注意是mapreduce包下的，不是旧的mapred包版本），这个InputFormat专门用于读取Avro格式的文件，并且能自动解析内嵌的Schema。
• 如果用自定义生成的Avro类，要把泛型指定为你的类，比如AvroInputFormat[YourAvroGeneratedClass]。

示例代码（Scala） #

import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.avro.mapreduce.AvroInputFormat
import org.apache.hadoop.io.NullWritable
import org.apache.avro.generic.GenericRecord

// 假设你已经初始化好SparkConf和StreamingContext
val ssc = new StreamingContext(sparkConf, Seconds(5))

// 创建Avro数据流
val avroDStream = ssc.fileStream[NullWritable, GenericRecord, AvroInputFormat[GenericRecord]](
  "/path/to/your/avro/data/directory"
)

// 处理数据：提取value部分（GenericRecord）并操作
avroDStream.map(_._2).foreachRDD { rdd =>
  rdd.foreach { record =>
    // 根据内嵌Schema的字段名获取值（注意类型转换要匹配实际数据类型）
    val userId = record.get("user_id").asInstanceOf[Int]
    val userName = record.get("user_name").asInstanceOf[String]
    println(s"User ID: $userId, Name: $userName")
  }
}

ssc.start()
ssc.awaitTermination()

补充提醒 #

• 确保你的项目依赖中包含了Avro和Spark Streaming的相关包，比如Maven中要添加：

  <dependency>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro</artifactId>
    <version>1.11.0</version>
  </dependency>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.12</artifactId>
    <version>3.3.0</version>
    <scope>provided</scope>
  </dependency>
  

• 如果用Java开发，代码逻辑完全一致，只需要调整为Java的语法写法即可。

内容的提问来源于stack exchange，提问作者Nk.Pl