在Apache Beam与Dataflow中使用ReadFromKafka时遇属性错误

问题场景 #

尝试使用Apache Beam Python SDK连接Kafka Topic，并将管道提交为Dataflow任务，运行时出现AttributeError: 'bool' object has no attribute 'encode'错误。

代码片段 #

import sys
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
import argparse
import os

from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io.kafka import ReadFromKafka
from apache_beam.coders.coders import Coder

os.environ["GOOGLE_APPLICATION_CREDENTIALS"]="/path/to/gcp_auth.json"

parser = argparse.ArgumentParser()
args, beam_args = parser.parse_known_args()
print(beam_args)

beam_options = PipelineOptions(
    beam_args,
    runner='DataflowRunner',
    project='my_project',
    job_name='my_job_name',
    temp_location='my_bucket/temp',
    region='my_region',
    network='my_network',
    subnetwork='my_sub_network',
    service_account_email = 'my_service_account'
)

with beam.Pipeline(options=beam_options) as pipeline:
  msg_kv_bytes = pipeline | "read file" >> ReadFromKafka(
     consumer_config={
      "bootstrap.servers": "ip:port",
      "group.id": "my_group_id",
      "security.protocol": "SSL",
        "ssl.ca.location": "bucket/cacerts.cer",
        "ssl.key.location":"bucket/clientcert.key",
        "ssl.certificate.location":"bucket/clientcert.cer",
        "ssl.keystore.password":"mypass",
        "ssl.key.password": "mypass",
        "enable.auto.commit": True,
        "auto.offset.reset": "earliest"
    },
    topics=["my_topic"]
  )

运行命令 #

python3 readfromkafka.py --worker_region my_region --runner DataflowRunner --project my_project --network my_network --subnetwork my_subnet --temp_location gcs_bucket/temp  --no_use_public_ips --streaming

错误日志 #

Traceback (most recent call last):
  File "/Users/myuser/Documents/dev_box/apache_beam/network-outage/readfromkafka.py", line 36, in <module>
    msg_kv_bytes = pipeline | "read file" >> ReadFromKafka(
                                             ^^^^^^^^^^^^^^
  File "/Users/myuser/Documents/dev_box/apache_beam/beam_env/lib/python3.11/site-packages/apache_beam/io/kafka.py", line 177, in __init__
    super().__init__(
  File "/Users/myuser/Documents/dev_box/apache_beam/beam_env/lib/python3.11/site-packages/apache_beam/transforms/external.py", line 652, in __init__
    payload.payload() if isinstance(payload, PayloadBuilder) else payload)
    ^^^^^^^^^^^^^^^^^
  File "/Users/myuser/Documents/dev_box/apache_beam/beam_env/lib/python3.11/site-packages/apache_beam/transforms/external.py", line 110, in payload
    return self.build().SerializeToString()
           ^^^^^^^^^^^^
  File "/Users/myuser/Documents/dev_box/apache_beam/beam_env/lib/python3.11/site-packages/apache_beam/transforms/external.py", line 147, in build
    schema=schema, payload=RowCoder(schema).encode(row))
                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/prabhan/Documents/dev_box/apache_beam/beam_env/lib/python3.11/site-packages/apache_beam/coders/coders.py", line 448, in encode
    return self.get_impl().encode(value)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "apache_beam/coders/coder_impl.py", line 233, in apache_beam.coders.coder_impl.StreamCoderImpl.encode
  File "apache_beam/coders/coder_impl.py", line 236, in apache_beam.coders.coder_impl.StreamCoderImpl.encode
  File "apache_beam/coders/coder_impl.py", line 1769, in apache_beam.coders.coder_impl.RowCoderImpl.encode_to_stream
  File "apache_beam/coders/coder_impl.py", line 701, in apache_beam.coders.coder_impl.MapCoderImpl.encode_to_stream
  File "apache_beam/coders/coder_impl.py", line 270, in apache_beam.coders.coder_impl.CallbackCoderImpl.encode_to_stream
  File "/Users/myuser/Documents/dev_box/apache_beam/beam_env/lib/python3.11/site-packages/apache_beam/coders/coders.py", line 406, in encode
    return value.encode('utf-8')
           ^^^^^^^^^^^^
AttributeError: 'bool' object has no attribute 'encode'

错误原因与解决方案 #

原因 #

错误根源在于Kafka consumer配置中的布尔值参数（如enable.auto.commit）使用了Python原生布尔类型True，而Beam在序列化这些配置时会默认尝试对所有值调用字符串的encode('utf-8')方法，但布尔对象没有该方法，因此触发AttributeError。

Beam的Kafka连接器底层依赖Java客户端，而Java客户端的所有配置参数值均为字符串类型，即使是布尔、数值类参数也需要以字符串形式传递。

解决方案 #

将consumer_config中的布尔值参数改为字符串格式的"true"或"false"，修改后的配置如下：

consumer_config={
  "bootstrap.servers": "ip:port",
  "group.id": "my_group_id",
  "security.protocol": "SSL",
  "ssl.ca.location": "bucket/cacerts.cer",
  "ssl.key.location":"bucket/clientcert.key",
  "ssl.certificate.location":"bucket/clientcert.cer",
  "ssl.keystore.password":"mypass",
  "ssl.key.password": "mypass",
  "enable.auto.commit": "true",  # 替换True为字符串"true"
  "auto.offset.reset": "earliest"
}

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