如何在Python中解码Kafka Topic里Oracle NUMBER(38,10)类型ID字段

• 场景：通过Debezium Oracle连接器将源库DEBEZIUM.LOG表（ID字段为NUMBER类型）同步至Kafka Topic，PySpark消费数据时发现ID字段是Base64编码格式（如fXZlaCccAA==），写入MySQL整数类型字段时触发错误：

  java.sql.BatchUpdateException: Incorrect integer value: 'VMgnfsfUAA==' for column 'ID' at row 1
  

相关配置与代码 #

Debezium配置 #

name=inventory-connector
connector.class=io.debezium.connector.oracle.OracleConnector
tasks.max=1
database.server.name=server1
database.hostname=***.**.*.*
database.port=1521
database.user=USER
database.password=PASS
database.dbname=ORCLCDB
database.pdb.name=ORCLPDB1
database.connection.adapter=logminer
database.history.kafka.bootstrap.servers=localhost:9092
database.history.kafka.topic=schema-changes.inventory
table.include.list=DEBEZIUM.LOGS
column.include.list=DEBEZIUM.LOG.ID,DEBEZIUM.LOG.SERVICENAME
time.precision.mode=connect
transforms=filter,route
transforms.filter.type=io.debezium.transforms.Filter
transforms.filter.language=jsr223.groovy
transforms.filter.condition=value.source.table == 'LOG'
transforms.filter.topic.regex=server1.DEBEZIUM.*
transforms.route.type=org.apache.kafka.connect.transforms.RegexRouter
transforms.route.regex=([^.]+)\\.([^.]+)\\.([^.]+)
transforms.route.replacement=$3

Oracle表结构 #

CREATE TABLE log (
ID NUMBER NOT NULL PRIMARY KEY,
SERVICENAME VARCHAR2(50),

原始PySpark消费代码 #

def deserialize_avro(serialized_msg):
   data1 = json.loads(serialized_msg)
   dfMessage = data1["payload"]['after']
   
   return (
       dfMessage['ID'],
       dfMessage['SERVICENAME'],
       dfMessage['NAYAPAYID'],
       dfMessage['ACCOUNTNUMBER']
   )

def saveToMysql(currentDf, epocId):
    dbCredentials = {"user": mysql_user_name,
                     "password":mysql_password,
                     "driver":mysql_driver_class}

    print("Printing Epoc Id")
    print(epocId)

    processedAt = time.strftime("%Y-%m-%d %H:%M:%S")

    currentDfFinal = currentDf\
        .withColumn("processed_at",lit(processedAt))\
        .withColumn("batch_id", lit(epocId))

    print("Printing Before MySql Table Save: " + str(epocId))

    currentDfFinal \
        .write \
        .jdbc(url=mysqlJdbcUrl,
              table = mysql_table_name,
              mode = "append",
              properties = dbCredentials)

    print("Printing After MySql table Save " + str(epocId))

orders_df = spark \
        .readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
        .option("subscribe", KAFKA_TOPIC_NAME) \
        .option("startingOffsets", "latest") \
        .load()

orders_schema = StructType()\
        .add("ID", StringType()) \
        .add("SERVICENAME", StringType()) \
       
avro_deserialize_udf = psf.udf(deserialize_avro, returnType=orders_schema)

df = orders_df.withColumn("avro", avro_deserialize_udf(psf.col("value"))).select("avro.*")

df.writeStream \
        .trigger(processingTime='15 seconds') \
        .outputMode("update") \
        .foreachBatch(saveToMysql) \
        .start()

trans_detail_write_stream.awaitTermination()

Kafka消息示例 #

"payload":{"before":null,"after":{"ID":"fXZlaCccAA==","SERVICENAME":"ECommerce"},"source":{"version":"1.8.1.Final","connector":"oracle","name":"server1","ts_ms":1686053890489,"snapshot":"last","db":"ORCLPDB1","sequence":null,"schema":"DEBEZIUM","table":"WSLOG","txId":null,"scn":"2589081","commit_scn":null,"lcr_position":null},"op":"r","ts_ms":1686053890489,"transaction":null}}

解决方法 #

方法一：修改Debezium配置（推荐，从源头避免编码） #

Debezium Oracle连接器默认对NUMBER类型采用二进制序列化（Base64编码），可以通过添加配置参数让连接器直接输出可读的数值：
在Debezium配置中追加以下参数：

# 输出为字符串格式的数值
decimal.handling.mode=string
oracle.number.handling.mode=string

# 或者输出为高精度数值类型（适合需要保留精度的场景）
# decimal.handling.mode=precise
# oracle.number.handling.mode=precise

配置生效后，Kafka消息中的ID字段会直接显示为数字字符串或数值，无需后续解码。

方法二：在PySpark中解码Base64为数值 #

若无法修改Debezium配置，可在Python中解析Oracle NUMBER类型的二进制编码（Base64解码后的字节流），转换为整数。

1. 实现解码函数

添加以下解码函数到代码中：

import base64

def decode_oracle_number(base64_str):
    if not base64_str:
        return None
    # 解码Base64字符串为字节数据
    byte_data = base64.b64decode(base64_str)
    # 解析Oracle NUMBER二进制格式：第一个字节是符号+指数
    sign_byte = byte_data[0]
    # 符号位：bit7为0是正数，1是负数
    sign = 1 if (sign_byte & 0x80) == 0 else -1
    # 指数：bit0-6的值减去65
    exponent = (sign_byte & 0x7F) - 65
    # 后续每个字节存储两位BCD码，高位在前
    digits = []
    for b in byte_data[1:]:
        digits.append(str((b >> 4) & 0x0F))
        digits.append(str(b & 0x0F))
    # 移除末尾无效的0
    while digits and digits[-1] == '0':
        digits.pop()
    if not digits:
        return 0
    # 拼接数字字符串并计算最终数值
    num_str = ''.join(digits)
    num = int(num_str) * (10 ** (exponent - (len(num_str)-1)))
    return num * sign

2. 修改反序列化函数

更新deserialize_avro函数，对ID字段进行解码：

def deserialize_avro(serialized_msg):
    data1 = json.loads(serialized_msg)
    dfMessage = data1["payload"]['after']
    
    # 解码ID字段
    decoded_id = decode_oracle_number(dfMessage['ID'])
    
    return (
        decoded_id,
        dfMessage['SERVICENAME'],
        dfMessage.get('NAYAPAYID'),  # 使用get避免KeyError
        dfMessage.get('ACCOUNTNUMBER')
    )

3. 更新Schema定义

将ID字段的类型从StringType()改为LongType()（或IntegerType()，根据实际数值范围调整）：

orders_schema = StructType()\
    .add("ID", LongType()) \
    .add("SERVICENAME", StringType()) \
    .add("NAYAPAYID", StringType()) \
    .add("ACCOUNTNUMBER", StringType())

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