Spark Scala读取Kafka Avro主题报错及相关技术问题咨询

1. 解决「Failed to find data source: kafka」报错 #

这个错误是因为Spark未加载Kafka数据源依赖，需根据运行方式引入对应包：

• spark-shell启动时：添加--packages参数指定依赖包（注意Scala和Spark版本需与环境匹配，示例为Scala 2.12、Spark 3.5.0）：

  spark-shell --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0
  

• 打包Jar提交时：在build.sbt中添加依赖（Provided表示依赖由集群环境提供，避免打包冲突）：

  libraryDependencies += "org.apache.spark" %% "spark-sql-kafka-0-10" % "3.5.0" % Provided
  

2. 更实用的Scala程序运行方式 #

生产环境：打包Jar提交（推荐） #

将代码打包成Jar后，用spark-submit提交到集群，稳定性和可维护性更强：

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --class KafkaConsumerAvro \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0,org.apache.spark:spark-avro_2.12:3.5.0 \
  your-application.jar

开发调试：sbt run #

本地开发时，直接在项目目录执行sbt run，sbt会自动管理依赖，适合快速调试代码。

集群交互式调试：spark-shell带依赖启动 #

如果需要交互式调试，启动时带上Kafka和Avro依赖，避免手动导入：

spark-shell --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0,org.apache.spark:spark-avro_2.12:3.5.0 -i path/file.scala

3. 动态获取Avro Schema、解码并写入HUE表 #

HUE依赖Hive metastore，写入HUE表本质是写入Hive表，步骤如下：

3.1 动态获取Avro Schema #

从Confluent Schema Registry拉取（常用场景）

如果Kafka的Avro数据关联Schema Registry，直接通过客户端拉取：

import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient
import org.apache.avro.Schema

// 初始化Registry客户端
val schemaRegistryUrl = "http://your-registry-host:8081"
val client = new CachedSchemaRegistryClient(schemaRegistryUrl, 100)
// 方式1：通过Schema ID获取
val schema = client.getSchemaById(123) // 替换为实际的Schema ID
// 方式2：通过主题获取最新版本Schema
val schema = client.getLatestSchemaVersion("your-topic-value").getSchema

从二进制Avro数据中解析（内嵌Schema场景）

如果Avro二进制数据包含Schema，直接从字节流中提取：

import org.apache.avro.io.DecoderFactory
import org.apache.avro.generic.GenericDatumReader

def extractSchemaFromBytes(avroBytes: Array[Byte]): Schema = {
  val decoder = DecoderFactory.get().binaryDecoder(avroBytes, null)
  val reader = new GenericDatumReader[Any]()
  reader.getSchema // 从二进制中解析Schema
}

3.2 解码二进制Avro数据 #

推荐使用Spark SQL内置的Avro函数，简洁高效：

import org.apache.spark.sql.avro.functions._

// 读取Kafka原始数据
val kafkaRawDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "your-broker:9092")
  .option("subscribe", "your-topic")
  .load()

// 将Schema转为字符串，用于解码
val schemaStr = schema.toString
// 解码value字段为结构化数据
val decodedDF = kafkaRawDF.select(from_avro($"value", schemaStr).as("avro_data"))
  .select("avro_data.*") // 展开Avro的所有字段为DataFrame列

3.3 写入HUE（Hive表） #

确保Spark配置了Hive metastore地址，然后根据处理类型（流/批）写入：

流处理场景

decodedDF.writeStream
  .format("hive")
  .option("database", "your_hive_database")
  .option("table", "your_hive_table")
  .option("checkpointLocation", "/hdfs/path/to/checkpoint") // 必须指定，用于故障恢复
  .start()
  .awaitTermination()

批处理场景

decodedDF.write
  .mode("append") // 追加模式，避免覆盖已有数据
  .format("hive")
  .saveAsTable("your_hive_database.your_hive_table")

如果Hive表未提前创建，Spark会根据DataFrame结构自动生成表，也可以在HUE中手动创建表结构后再写入。

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