Issue: NullPointerException When Using PermissiveRecordExceptionHandler in Structured Streaming Job
- 主要语言
- Scala
- 星标
- 242
- 派生
- 84
- 平均合并
- 10 小时 12 分钟
- 30 天内合并 PR
- 2
描述
Description
I have a Spark Structured Streaming job where I read Avro messages from Kafka, convert them to a Spark DataFrame using Abris, and then write the stream as a Delta table to S3. Below is the relevant configuration and code snippet.
Abris Configuration
```
val abrisConfig = AbrisConfig.fromConfluentAvro
.downloadReaderSchemaByVersion(1)
.andRecordNameStrategy(schemaName, schemaNamespace)
.usingSchemaRegistry(schemaRegistryUrl)
.withExceptionHandler(new PermissiveRecordExceptionHandler())
```
Code Example
Here’s an example where I write the stream to the console for demonstration purposes:
```
val df = reader
.readKafkaStream(topic, kafkaConfig, options)
.withColumnRenamed("headers", KAFKA_HEADERS)
.withColumn("kafka_meta", map_from_entries(col(KAFKA_HEADERS)))
.withColumn("kafka_meta_column", col("kafka_meta.meta"))
.withColumn(KAFKA_HEADERS_META, from_avro(col("kafka_meta_column"), abrisConfig))
df.select(KAFKA_HEADERS_META).writeStream
.outputMode("append")
.format("console")
.option("truncate", "true")
.start()
```
The kafka_meta.meta column contains the Avro message I want to deserialize using from_avro. The schema for the Avro message is as follows:
```
{
"type": "record",
"name": "Example",
"fields": [
{ "name": "id", "type": "string" },
{ "name": "timestamp", "type": "long" }
]
}
```
Problem
I encounter the following error during execution:
Caused by: java.lang.NullPointerException
Upon investigation, I concluded that this occurs because the PermissiveRecordExceptionHandler returns a null record for malformed data as expected. However, downstream transformations attempt to access the nested field meta (the column containing the Avro message), resulting in the NullPointerException.
Note: None of the columns originally contain null values, so the issue is not caused by the source data.
Question
How can I handle this scenario to avoid the NullPointerException while keeping the permissive behavior of the exception handler? Any guidance would be appreciated.
Thank you!
贡献指南
这个仓库没有索引到贡献指南
评估
这个 Issue 还没有评估数据。