You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Spark中Schema Registry最新Schema缺失字段时解析Confluent Avro Kafka消息

avatar
安全阿岳
2026-8-14
Spark Streaming + Abris反序列化Kafka消息的问题及解决方案

问题场景

在Spark Streaming中使用Abris的from_confluent_avro(za.co.absa.abris.avro.functions.from_confluent_avro)函数读取并反序列化Kafka消息时，遇到两个问题：

  1. 当Schema Registry中的最新Schema缺失部分字段时，用该Schema反序列化会将消息中存在的有效值字段设为null；
  2. 尝试使用本地静态Avro Schema文件反序列化时，抛出ArrayIndexOutOfBoundsException错误。

Abris原配置 

SchemaManager.PARAM_SCHEMA_REGISTRY_TOPIC -> topicNm,
SchemaManager.PARAM_SCHEMA_REGISTRY_URL -> schemaRegUrl,
SchemaManager.PARAM_VALUE_SCHEMA_NAMING_STRATEGY -> "topic.name",
SchemaManager.PARAM_VALUE_SCHEMA_ID -> "latest"

Schema Registry中的Schema 

{
  "type": "record",
  "name": "topic_test",
  "fields": [
    {
      "name": "actions",
      "type": ["null", "int"],
      "default": null
    },
    {
      "name": "clicks",
      "type": ["null", "int"],
      "default": null
    },
    {
      "name": "deal_id",
      "type": ["null", "string"],
      "default": null
    },
    {
      "name": "deal_line_id",
      "type": ["null", "string"],
      "default": null
    },
    {
      "name": "delivery_date",
      "type": ["null", "long"],
      "default": null
    },
    {
      "name": "delivery_source",
      "type": ["null", "string"],
      "default": null
    },
    {
      "name": "delivery_type",
      "type": ["null", "string"],
      "default": null
    },
    {
      "name": "external_ps_ad_id",
      "type": ["null", "string"],
      "default": null
    },
    {
      "name": "external_ps_ad_name",
      "type": ["null", "string"],
      "default": null
    },
    {
      "name": "gross_revenue",
      "type": ["null", "int"],
      "default": null
    },
    {
      "name": "index",
      "type": ["null", "string"],
      "default": null
    },
    {
      "name": "job_id",
      "type": ["null", "string"],
      "default": null
    },
    {
      "name": "metric2",
      "type": ["null", "int"],
      "default": null
    },
    {
      "name": "metric3",
      "type": ["null", "int"],
      "default": null
    },
    {
      "name": "net_revenue",
      "type": ["null", "int"],
      "default": null
    },
    {
      "name": "order_id",
      "type": ["null", "string"],
      "default": null
    },
    {
      "name": "order_line_item_id",
      "type": ["null", "string"],
      "default": null
    },
    {
      "name": "parent_deal_line_id",
      "type": ["null", "string"],
      "default": null
    },
    {
      "name": "production_system_id",
      "type": ["null", "string"],
      "default": null
    },
    {
      "name": "source",
      "type": {
        "type": "record",
        "name": "source",
        "fields": [
          {
            "name": "destinationTenantName",
            "type": ["null", "string"],
            "default": null
          },
          {
            "name": "endOfJob",
            "type": ["null", "boolean"],
            "default": null
          },
          {
            "name": "entityType",
            "type": ["null", "string"],
            "default": null
          },
          {
            "name": "externalSystemName",
            "type": ["null", "string"],
            "default": null
          },
          {
            "name": "jobid",
            "type": ["null", "string"],
            "default": null
          },
          {
            "name": "parentJobId",
            "type": ["null", "string"],
            "default": null
          },
          {
            "name": "recordKey",
            "type": ["null", "string"],
            "default": null
          },
          {
            "name": "sourceSystemName",
            "type": ["null", "string"],
            "default": null
          },
          {
            "name": "sourceTenantName",
            "type": ["null", "string"],
            "default": null
          },
          {
            "name": "tenant_name",
            "type": ["null", "string"],
            "default": null
          },
          {
            "name": "topicPrefix",
            "type": ["null", "string"],
            "default": null
          }
        ]
      }
    },
    {
      "name": "write_identifier",
      "type": ["null", "string"],
      "default": null
    }
  ]
}

Kafka消息内容 

{
  "actions": 200,
  "clicks": 100,
  "deal_id": "6362359ce058c0124d04e14d",
  "deal_line_id": "636235b0e058c0124d04e16e",
  "delivery_date": 1667606400000,
  "delivery_source": "s4jY-JDRS8atYVX6VrvaqA",
  "delivery_type": "test primary",
  "external_ps_ad_id": "6147090412",
  "external_ps_ad_name": "ank_deal_1234",
  "gross_revenue": 0,
  "impressions": 8000,
  "index": "",
  "job_id": "e5d66396-b9e5-4662-a574-a342d517566a",
  "metric1": 8000,
  "metric2": 100,
  "metric3": 200,
  "net_revenue": 0,
  "order_id": "test order",
  "order_line_item_id": "test olid",
  "parent_deal_line_id": "",
  "production_system_id": "test production_system_id",
  "source": {
    "destinationTenantName": "testtenant",
    "endOfJob": false,
    "entityType": "testenitty",
    "externalSystemName": "server",
    "jobid": "d357c9d5-c004-4555-a581-864ab2f2ebfb",
    "parentJobId": "e5d66396-b9e5-4662-a574-a342d517566a",
    "recordKey": "6147090412_6McFTGZvQxuwJxAwUFRmmA_2022-11-05_s4jY-JDRS8atYVX6VrvaqA",
    "sourceSystemName": "test",
    "sourceTenantName": "test",
    "tenant_name": "test",
    "topicPrefix": "test"
  },
  "write_identifier": ""
}

静态Schema反序列化尝试及错误

尝试代码 

val jsonFormatSchema = new String(
  Files.readAllBytes(Paths.get("C:\\Users\\leo_t" +
    "\\src\\main\\resources\\ddavro.avsc")))

val deserDF = result.select(za.co.absa.abris.avro.functions.from_avro(col("value"),
  jsonFormatSchema).as("convertedMsg"))
  .select("convertedMsg.*")
deserDF.show(10,false)

错误日志 

2022-11-07 11:13:05,355  [ERROR] org.apache.spark.executor.Executor - Exception in task 0.0 in stage 3.0 (TID 14)
org.apache.spark.SparkException: Malformed records are detected in record parsing.
    at za.co.absa.abris.avro.sql.AvroDataToCatalyst.nullSafeEval(AvroDataToCatalyst.scala:68)
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.serializefromobject_doConsume_0$(Unknown Source)
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
    at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
    at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$11$$anon$1.hasNext(WholeStageCodegenExec.scala:619)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:255)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:247)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:836)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:836)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
    at org.apache.spark.scheduler.Task.run(Task.scala:121)
    at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:402)
    at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:408)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ArrayIndexOutOfBoundsException: 12
    at org.apache.avro.io.parsing.Symbol$Alternative.getSymbol(Symbol.java:460)
    at org.apache.avro.io.ResolvingDecoder.readIndex(ResolvingDecoder.java:284)
    at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:187)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:160)
    at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:259)
    at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:247)
    at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:179)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:160)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:153)
    at za.co.absa.abris.avro.sql.AvroDataToCatalyst.decodeVanillaAvro(AvroDataToCatalyst.scala:137)
    at za.co.absa.abris.avro.sql.AvroDataToCatalyst.decode(AvroDataToCatalyst.scala:103)
    at za.co.absa.abris.avro.sql.AvroDataToCatalyst.nullSafeEval(AvroDataToCatalyst.scala:59)
    ... 22 more
2022-11-07 11:13:05,407  [ERROR] org.apache.spark.scheduler.TaskSetManager - Task 0 in stage 3.0 failed 1 times; aborting job

解决方案

方案1：使用消息自带的Schema ID反序列化（推荐）

Confluent Avro格式的Kafka消息头部会包含序列化时使用的Schema ID，Abris可以自动提取该ID并从Schema Registry拉取对应的Schema进行反序列化，精准匹配消息的原始Schema，避免字段丢失。

修改Abris配置，移除PARAM_VALUE_SCHEMA_ID -> "latest"：

val abrisConfig = Map(
  SchemaManager.PARAM_SCHEMA_REGISTRY_TOPIC -> topicNm,
  SchemaManager.PARAM_SCHEMA_REGISTRY_URL -> schemaRegUrl,
  SchemaManager.PARAM_VALUE_SCHEMA_NAMING_STRATEGY -> "topic.name"
)

val deserDF = result.select(from_confluent_avro(col("value"), abrisConfig).as("convertedMsg"))
  .select("convertedMsg.*")
deserDF.show(10, false)

方案2：修复静态Schema反序列化错误

静态Schema反序列化报错的核心原因是本地Schema与消息序列化时使用的Schema不兼容（比如本地Schema缺失impressions、metric1字段，或字段顺序、类型不匹配）。

解决步骤：

  1. 通过消息头部的Schema ID，从Schema Registry查询到消息对应的原始Schema；
  2. 将该Schema保存为本地静态文件，确保包含消息中的所有字段；
  3. 重新执行静态Schema反序列化代码。

示例补充字段后的静态Schema片段：

{
  "type": "record",
  "name": "topic_test",
  "fields": [
    // 保留原有所有字段
    {
      "name": "impressions",
      "type": ["null", "int"],
      "default": null
    },
    {
      "name": "metric1",
      "type": ["null", "int"],
      "default": null
    }
    // 其他原有字段...
  ]
}

方案3：配置Schema Registry的兼容性模式

将Schema Registry的兼容性模式设置为backward兼容，确保新增字段不会影响旧消息的反序列化，同时旧Schema也能正确解析包含新字段的消息。但仍建议优先使用方案1，保证用消息对应的Schema反序列化。

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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
火山引擎 最新活动
全面开放
Doubao-Seedance-2.5全新模型升级

模型再升级,30秒超长叙事, 模态参考扩容

HOT
方舟Codingplan

模型自由，工具不限，最新支持 Deepseek-V4 系列、GLM-5.3 系列

HOT
方舟Agent plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列

HOT
新用户特惠专场

大模型19元起，Al应用9.9元畅享，新人首购爆款尽享优惠

最近更新时间：2026.08.14 16:05:34