我一直在尝试用Scala2.11从spark structured streaming（2.4.4）读取kafka的avro序列化消息。为此，我使用了spark avro（依赖关系如下）。我使用合流kafka库从python生成kafka消息。spark streaming能够使用模式来消费消息，但是它不能正确读取字段的值。我准备了一个简单的例子来说明这个问题，代码在这里可用：https://github.com/anigmo97/simpleexamples/tree/master/spark_streaming_kafka_avro_scala
我用python创建记录，记录的模式是：

{
    "type": "record",
    "namespace": "example",
    "name": "RawRecord",
    "fields": [
        {"name": "int_field","type": "int"},
        {"name": "string_field","type": "string"}
    ]
}

它们是这样产生的：

from time import sleep
from confluent_kafka.avro import AvroProducer, load, loads

def generate_records():
    avro_producer_settings = {
        'bootstrap.servers': "localhost:19092",
        'group.id': 'groupid',
        'schema.registry.url': "http://127.0.0.1:8081"
    }
    producer = AvroProducer(avro_producer_settings)
    key_schema = loads('"string"')
    value_schema = load("schema.avsc")
    i = 1
    while True:
        row = {"int_field": int(i), "string_field": str(i)}
        producer.produce(topic="avro_topic", key="key-{}".format(i), 
                         value=row, key_schema=key_schema, value_schema=value_schema)
        print(row)
        sleep(1)
        i+=1

spark结构化流媒体（scala）的消耗如下：

import org.apache.spark.sql.{ Dataset, Row}
import org.apache.spark.sql.streaming.{ OutputMode, StreamingQuery}
import org.apache.spark.sql.avro._
...
        try {

            log.info("----- reading schema")
            val jsonFormatSchema = new String(Files.readAllBytes(
                                                    Paths.get("./src/main/resources/schema.avsc")))

            val ds:Dataset[Row] = sparkSession
                .readStream
                .format("kafka")
                .option("kafka.bootstrap.servers", kafkaServers)
                .option("subscribe", topic)
                .load()

            val output:Dataset[Row] = ds
                .select(from_avro(ds.col("value"), jsonFormatSchema) as "record")
                .select("record.*")

            output.printSchema()

            var query: StreamingQuery = output.writeStream.format("console")
                .option("truncate", "false").outputMode(OutputMode.Append()).start();

            query.awaitTermination();

        } catch {
            case e: Exception => log.error("onApplicationEvent error: ", e)
            //case _: Throwable => log.error("onApplicationEvent error:")
        }
...

在spark中打印模式时，奇怪的是字段可以为null，尽管avro模式不允许这样。spark展示了这一点：

root
 |-- int_field: integer (nullable = true)
 |-- string_field: string (nullable = true)

我已经用python检查了另一个消费者的消息，这些消息很好，但与spark显示的消息内容无关。

+---------+------------+
|int_field|string_field|
+---------+------------+
|0        |            |
+---------+------------+

使用的主要依赖项有：

<properties>
    <spark.version>2.4.4</spark.version>
    <scala.version>2.11</scala.version>
</properties>

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_${scala.version}</artifactId>
    <version>${spark.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_${scala.version}</artifactId>
    <version>${spark.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-avro_${scala.version}</artifactId>
    <version>${spark.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_${scala.version}</artifactId>
    <version>${spark.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql-kafka-0-10_${scala.version}</artifactId>
    <version>${spark.version}</version>
</dependency>

有人知道为什么会这样吗？
提前谢谢。重现错误的代码如下：
https://github.com/anigmo97/simpleexamples/tree/master/spark_streaming_kafka_avro_scala

解决方案

问题是，我使用的是python中的合流Kafka库，我使用spark avro库读取spark结构化流中的avro消息。
confluent的Kafka库使用confluent的avro格式，spark的avro读取使用标准的avro格式。
区别在于，为了使用schema registry，confluent avro在消息前面加上四个字节，指示应该使用哪个schema。
资料来源：https://www.confluent.io/blog/kafka-connect-tutorial-transfer-avro-schemas-across-schema-registry-clusters/
为了能够使用合流avro并从spark结构化流媒体读取它，我将spark avro库替换为abris（abris允许将avro和合流avro与spark集成）。https://github.com/absaoss/abris