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

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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/67f3f7d6-4e6a-c776-8729-47864844bc1c.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/dcb2c740-7e5f-cd6a-f283-0733fe9f9bc8.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)

Avro 形式のデータをデシアライズするサイトの除去するバイト数が異なることに注意してください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/06e7a6e6-7be6-ca8b-528a-1bcce54e16e0.png)

引用元：[Databricks にて Confluent から Avro 形式のデータを手動で読み込むときの MALFORMED_AVRO_MESSAGE エラーへの対応方法 #Kafka - Qiita](https://qiita.com/manabian/items/068faf8cebe3fdb51445)


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

### 1. 事前準備

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

```text
syntax = "proto3";
package ksql;

message users {
  int64 registertime = 1;
  string userid = 2;
  string regionid = 3;
  string gender = 4;
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/587f03c4-2a60-88ac-1c81-6878e4d288b5.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/fb57b8c6-4aa1-6c8c-74f2-0c34005e52c5.png)


### Databricks ノートブックと同階層のディレクトリに`user.proto`ファイルを作成

```proto
syntax = "proto3";
package ksql;

message users {
  int64 registertime = 1;
  string userid = 2;
  string regionid = 3;
  string gender = 4;
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e365ea2c-af3f-f81f-6fe0-46b4d8dc13ad.png)


### Databricks にて`user.desc`ファイルを生成

```sh
%sh sudo apt-get update -qq
```

```sh
%sh sudo apt-get install -y protobuf-compiler -qq
```

```sh
%sh protoc --include_imports \
--descriptor_set_out=user.desc \
user.proto
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c7a1046d-6274-a3f2-976c-2792e5d8b67c.png)


### Databricks にて Topic をセット

```python
topic_name = "protobu_user"
```

![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 import functions as F
from pyspark.sql.functions import col, expr, from_json, to_char
from pyspark.sql.protobuf.functions import from_protobuf

df_stripped = df.withColumn(
    "value_stripped",
    F.expr(f"substring(value, 6, length(value) - 5)"),
)

df_stripped = df_stripped.withColumn(
    "decoded_value",
    from_protobuf(
        col("value_stripped"),
        messageName="ksql.users", 
        descFilePath="./user.desc"  # Descriptor File のパス
    ),
)

df_stripped.select("value", "decoded_value").display()
```

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/19ea3070-0849-8582-5b43-3e2022e70d65.png)


## エラーの対応方法

デシリアライズする前にMagic Byte 、 Schema ID 、および、 message-indexes 分のバイト （6 バイト）を除去してから実行することを確認できます。

```python
magic_bytes = 1
schema_id_bytes = 4
message_indexes_bytes = 1
offset_bytes = str(magic_bytes + schema_id_bytes + message_indexes_bytes)
start_bytes = str(int(offset_bytes) + 1)

from pyspark.sql import functions as F
from pyspark.sql.functions import col, expr, from_json, to_char
from pyspark.sql.protobuf.functions import from_protobuf

df_stripped = df.withColumn(
    "value_stripped",
    F.expr(f"substring(value, {start_bytes}, length(value) - {offset_bytes})"),
)

df_stripped = df_stripped.withColumn(
    "decoded_value",
    from_protobuf(
        col("value_stripped"),
        messageName="ksql.users", 
        descFilePath="./user.desc"  # Descriptor File のパス
    ),
)

df_stripped.select("value", "decoded_value").display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/915eb0e1-7912-f222-dec5-ec73a7b3b0ba.png)