'PBegin' 对象在运行光束管道时没有属性'windowing'答案

【问题标题】：'PBegin' object has no attribute 'windowing' while running beam pipeline'PBegin' 对象在运行光束管道时没有属性'windowing'
【发布时间】：2019-11-05 03:07:45
【问题描述】：

在运行数据流作业时，我得到的“PBegin”对象没有“windowing”属性。我在 pardo 函数中调用 connectclass 类。

我正在尝试从 Beam python SDK 连接 NOSQL 数据库并运行 sql 以从表中提取数据。然后使用另一个 pardo 将输出写入单独的文件。

class Connector(beam.DoFn):
    def __init__(self,username,seeds,keyspace,password,datacenter=None):
    self.username = username
    self.password = password
    self.seeds = seeds
    self.keyspace = keyspace
    self.datacenter = datacenter
    super(self.__class__, self).__init__()

    def process(self, element):

    if datacenter:
        load_balancing_policy = DCAwareRoundRobinPolicy(local_dc=self.datacenter)
    auth_provider = PlainTextAuthProvider(username=self.username, password=self.password)
    cluster = Cluster(contact_points=self.seeds,
                      load_balancing_policy=load_balancing_policy,
                      auth_provider=auth_provider)
    session=cluster.connect(self.seeds,self.keyspace,self.username, self.password, self.datacenter)
    rows = session.execute(SQL Query)
    yield rows

【问题讨论】：

标签： python-3.x google-cloud-dataflow apache-beam apache-beam-io

【解决方案1】：

偶然发现了同样的问题。尝试连接到 RDBMS 源，但我猜在实现设计方面，NoSQL 和 SQL 数据库之间没有区别。

除了 Jayadeep Jayaraman 的建议之外，这可以通过使用 ParDo 来实现。实际上，使用 ParDo 进行连接是 beam documentation 推荐的，如果这样做的限制对于您的用例是可以接受的：

对于有界（批处理）源，目前有两种创建 Beam 源的选项：

使用 ParDo 和 GroupByKey。

使用 Source 接口并扩展 BoundedSource 抽象子类。

ParDo 是推荐的选项，因为实现 Source 可能很棘手。请参阅何时使用 > 源接口以获取您可能希望使用源的一些用例列表 > >（例如动态工作重新平衡）。

你没有展示你如何使用你的 DoFn。对我来说，记住 DoFn 作用于已经存在的 PCollection 的元素是很有帮助的。它不能自己从头开始创建你的 DoFn。因此，要解决您提到的问题，您可能需要从内存中创建一个 PCollection，其中包含一个用于从源检索数据的查询元素。然后将从您的源读取的 ParDo 应用到此 PCollection。

顺便说一句：我想出了每个分区的一个元素，我想从我的 Pcollection 中的 RDBMS 中读取数据 - 因此可以从我的 SQL 数据库中并行读取数据。

解决方案可能如下所示：

p | beam.Create(["Your Query / source object qualifier goes here"]) 
  | "Read from Database" >> beam.ParDo(YourConnector())

我还要提一下使用 DoFn 的 start_bundle 和 finish_bundle 方法来设置/断开连接可能是个好主意。

【讨论】：

【解决方案2】：

您需要为此使用 Beam IO。这里有一个关于如何在 Python 中构建自定义 IO 的指南 [1]。

ParDo 通常用于在 PCollection 上运行转换。您还可以查看 SplittableDoFn 来构建类似的东西。参考这里[2]

1 - https://beam.apache.org/documentation/io/developing-io-python/

2 - https://beam.apache.org/blog/2017/08/16/splittable-do-fn.html

【讨论】：