如何修复 Apache Beam 的 AttributeError：对象“BmsSchema”没有属性。 “元素类型”

使用架构处理解决 Apache Beam 中的属性错误

使用 Apache Beam 的 Python 脚本 - 解决方案 1：使用 NamedTuple 定义架构

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from apache_beam.io.gcp.bigquery import WriteToBigQuery
from apache_beam.dataframe.convert import to_dataframe
import os
import typing
import json
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "path/to/your-credentials.json"
# Define schema using NamedTuple for type enforcement
class BmsSchema(typing.NamedTuple):
    ident: str
beam.coders.registry.register_coder(BmsSchema, beam.coders.RowCoder)
# Parses Pub/Sub messages
class ParsePubSubMessage(beam.DoFn):
    def process(self, message):
        all_columns = ['ident']
        main_dict = dict(zip(all_columns, [None] * len(all_columns)))
        record = json.loads(message.decode('utf-8'))
        main_dict.update(record)
        yield {all_columns[0]: main_dict[all_columns[0]]}
# Transforms data with Pandas integration
class PandasTransform(beam.DoFn):
    def process(self, element):
        df = to_dataframe([element])
        for _, row in df.iterrows():
            yield row.to_dict()
def run():
    options = PipelineOptions(
        project='your-project-id',
        runner='DirectRunner',
        streaming=True,
        temp_location='gs://your-temp-location',
        region='your-region')
    options.view_as(StandardOptions).streaming = True
    input_subscription = 'projects/your-project/subscriptions/your-subscription'
    table_schema = {"fields": [{"name": "ident", "type": "STRING", "mode": "ABLE"}]}
    with beam.Pipeline(options=options) as p:
        messages = (
            p | 'Read from PubSub' >> beam.io.ReadFromPubSub(subscription=input_subscription)
              | 'Parse PubSub Message' >> beam.ParDo(ParsePubSubMessage())
              | 'Attach Schema' >> beam.Map(lambda x: BmsSchema(x)).with_output_types(BmsSchema)
              | 'Transform with Pandas' >> beam.ParDo(PandasTransform())
        )
        messages | 'Write to BigQuery' >> WriteToBigQuery(
            table='your_dataset.your_table',
            schema=table_schema,
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
            custom_gcs_temp_location='gs://your-temp-location'
        )
if __name__ == '__main__':
    run()

替代解决方案：使用基于类的架构处理 Apache Beam 中的架构属性

使用 Apache Beam 的 Python 脚本 - 解决方案 2：带有类型检查的基于类的架构

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from apache_beam.io.gcp.bigquery import WriteToBigQuery
from apache_beam.dataframe.convert import to_dataframe
import os
import json
# Define a class-based schema with validation method
class BmsSchema:
    def __init__(self, ident):
        self.ident = ident
    def validate(self):
        if not isinstance(self.ident, str):
            raise TypeError("Expected 'ident' to be a string")
class ParsePubSubMessage(beam.DoFn):
    def process(self, message):
        record = json.loads(message.decode('utf-8'))
        ident = record.get('ident', None)
        yield BmsSchema(ident=ident)
class PandasTransform(beam.DoFn):
    def process(self, element):
        if hasattr(element, 'validate'):
            element.validate()
        df = to_dataframe([{'ident': element.ident}])
        for _, row in df.iterrows():
            yield row.to_dict()
def run_pipeline():
    options = PipelineOptions(
        project='your-project-id',
        runner='DirectRunner',
        streaming=True,
        temp_location='gs://your-temp-location',
        region='your-region')
    options.view_as(StandardOptions).streaming = True
    input_subscription = 'projects/your-project/subscriptions/your-subscription'
    table_schema = {"fields": [{"name": "ident", "type": "STRING", "mode": "ABLE"}]}
    with beam.Pipeline(options=options) as p:
        messages = (
            p | 'Read from PubSub' >> beam.io.ReadFromPubSub(subscription=input_subscription)
              | 'Parse Message' >> beam.ParDo(ParsePubSubMessage())
              | 'Transform Columns' >> beam.ParDo(PandasTransform())
        )
        messages | 'Write to BigQuery' >> WriteToBigQuery(
            table='your_dataset.your_table',
            schema=table_schema,
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
            custom_gcs_temp_location='gs://your-temp-location'
        )
if __name__ == '__main__':
    run_pipeline()

有关 Apache Beam 架构错误的常见问题

1. 是什么原因导致“AttributeError：'MySchemaClassName'对象没有属性”错误？
2. 当为对象定义的架构与正在处理的数据不匹配时，Apache Beam 中经常会出现此错误。确保使用显式注册架构 beam.coders.registry.register_coder。
3. 如何在 Apache Beam 中注册自定义架构？
4. 在 Apache Beam 中，您可以使用定义自定义架构 typing.NamedTuple 对于结构化数据，然后将其注册到 beam.coders.RowCoder 来管理序列化。
5. 使用目的是什么 to_dataframe 在 Beam 管道中？
6. to_dataframe 将 Beam PCollection 转换为 Pandas DataFrame，允许您使用 Pandas 函数进行转换。确保数据架构兼容以避免属性错误。
7. 如何处理 Beam 和 BigQuery 之间的类型不匹配？
8. 确保 BigQuery 架构与 Beam 中定义的数据架构匹配。使用 WriteToBigQuery 通过架构强制执行，并在管道的早期验证数据类型。
9. 我可以在运行管道之前捕获架构错误吗？
10. 是的，通过在每个中添加自定义验证器 DoFn 类，您可以在数据格式导致管道错误之前检查它们。
11. 正在使用 beam.Map 比 beam.DoFn 用于转换？
12. 这取决于。 beam.Map 对于直接转换来说很简单，但是 beam.DoFn 为复杂逻辑提供更大的灵活性，特别是在需要架构调整时。
13. 为什么 Beam 管道需要显式 with_output_types 声明？
14. Apache Beam 强制执行类型安全以维护跨转换的架构完整性。使用 with_output_types 有助于强制执行预期类型并防止运行时错误。
15. 怎么样 ParsePubSubMessage 在示例中工作？
16. ParsePubSubMessage 是一个 DoFn 解码 JSON 消息、应用预期的模式格式并生成它以便在管道中进一步处理的函数。
17. 我可以在 Beam 中使用带有嵌套对象的架构吗？
18. 是的，Apache Beam 支持复杂的模式。使用 NamedTuple 对于嵌套模式并将它们注册为 17 号 为了正确的序列化。
19. 有什么区别 DirectRunner 和 Beam 的其他跑步者？
20. DirectRunner 主要用于本地测试。对于生产，使用像这样的跑步者 DataflowRunner 在 Google Cloud 上部署管道。

总结：解决 Apache Beam 属性错误

了解属性错误的根本原因 阿帕奇光束（通常是由于架构未对齐）可以防止未来出现问题并提高数据处理的可靠性。通过注册模式、确保类型兼容性和使用结构化转换，本指南提供了解决“AttributeError”问题的实用步骤。

借助这些解决方案，您可以自信地构建处理从 Pub/Sub 到 BigQuery 的实时数据的管道，同时保持架构完整性。无论是在单个项目上工作还是在生产环境中扩展，这些技术都有助于使数据管道更加高效、强大且易于管理。 🚀