Databricks にて Confluent から Avro 形式のデータを手動で読み込むときの MALFORMED_AVRO_MESSAGE エラーへの対応方法

## 概要

Databricks で Confluent から Avro 形式のデータを読み込む際に発生する MALFORMED_AVRO_MESSAGE エラーへの対処方法を共有します。本エラーはデータ型の不一致だけでなく、Confluent が生成するバイナリ形式の仕様が原因となる場合もあります。Confluent が生成するバイナリは、先頭 5 バイトに Magic Byte と Schema ID が格納されているため、手動でバイナリをデシリアライズする際にはこの 5 バイトを除去する必要があります。本記事では、その具体的な対応手順を解説します。

> org.apache.spark.SparkException: [MALFORMED_AVRO_MESSAGE] Malformed Avro messages are detected in message deserialization. Parse Mode: FAILFAST. To process malformed Avro message as null result, try setting the option 'mode' as 'PERMISSIVE'. SQLSTATE: KD000

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b648604c-7ff3-1491-a193-4d0f2dc0afe5.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4f2a5b0c-0975-123a-8597-33a44f79ad7e.png)

引用元：[Formats, Serializers, and Deserializers for Schema Registry on Confluent Cloud | Confluent Documentation](https://docs.confluent.io/cloud/current/sr/fundamentals/serdes-develop/index.html#wire-format)

## エラーの再現方法と対応方法

### 1. 事前準備

#### Confluent にて Avro 形式の Topic を準備

```text
{
  "connect.name": "ksql.users",
  "fields": [
    {
      "name": "registertime",
      "type": "long"
    },
    {
      "name": "userid",
      "type": "string"
    },
    {
      "name": "regionid",
      "type": "string"
    },
    {
      "name": "gender",
      "type": "string"
    }
  ],
  "name": "users",
  "namespace": "ksql",
  "type": "record"
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/643f4f7c-9d69-758d-48d1-192d88862345.png)

### Databricks にて Confluent に対する認証情報をセット

```python
# Confluent に対する認証情報をセット
bootstrap_servers = "pkc-921aa.us-east-2.aws.confluent.cloud:9092"
sasl_username = "U4BML4OPRCaaaa"
sasl_password = "zzEjUzyUwYYJneNE240L7jgPVrHAtRBANo6wnnzCgKlGIJ9OxJqaJZWRvaaaaa"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/89146f4f-04dc-d41e-c4ef-4c6c2ef3af82.png)

### Databricks にて Topic とスキーマをセット

```python
topic_name = "avro_user_1"
schema = """
{
  "connect.name": "ksql.users",
  "fields": [
    {
      "name": "registertime",
      "type": "long"
    },
    {
      "name": "userid",
      "type": "string"
    },
    {
      "name": "regionid",
      "type": "string"
    },
    {
      "name": "gender",
      "type": "string"
    }
  ],
  "name": "users",
  "namespace": "ksql",
  "type": "record"
}
"""
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f4a22609-fd68-6dda-9303-036ac6569001.png)

### Confluent を参照したデータフレームを定義

```python
# Confluetn から読み込み
df = (
    spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", bootstrap_servers)
    .option("kafka.security.protocol", "SASL_SSL")
    .option(
        "kafka.sasl.jaas.config",
        f'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="{sasl_username}" password="{sasl_password}";',
    )
    .option("kafka.sasl.mechanism", "PLAIN")
    .option("subscribe", topic_name)
    .option("startingOffsets", "earliest")
    .load()
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7d0af2ac-1aac-3a97-028b-5e1191bc4092.png)


## エラーの再現方法

```python
from pyspark.sql.functions import col, expr, to_char
from pyspark.sql.avro.functions import from_avro

# 取得したデータに対する Deserialize を実施
decoded_df = df.withColumn(
    "value",
    from_avro(
        col("value_no_magic"),
        schema,
    ).alias("decoded_value"),
)

# 確認に必要なカラムのみ残して表示
selected_df = decoded_df.select("decoded_value")
display(selected_df)
```

```text
org.apache.spark.SparkException: [MALFORMED_AVRO_MESSAGE] Malformed Avro messages are detected in message deserialization. Parse Mode: FAILFAST. To process malformed Avro message as null result, try setting the option 'mode' as 'PERMISSIVE'. SQLSTATE: KD000
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a53bb4c1-230a-8180-d115-76ec184c6ddc.png)


## エラーの対応方法

デシリアライズする前に 5 バイトを除去してから実行することを確認できます。

```python
from pyspark.sql.functions import col, expr, to_char
from pyspark.sql.avro.functions import from_avro

# 先頭 5バイトをスキップしてバイト列として取り出す
no_magic_df = df.withColumn(
    "value_no_magic", expr("substring(value, 6, length(value) - 5)")
)

# 取得したデータに対する Deserialize を実施
decoded_df = no_magic_df.withColumn(
    "decoded_value",
    from_avro(
        col("value_no_magic"),
        schema,
    ).alias("decoded_value"),
)

# 確認に必要なカラムのみ残して表示
selected_df = decoded_df.select("decoded_value")
display(selected_df)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c5b132c6-bb95-e556-efc8-83335c45b012.png)