查询 Google 数据存储的 Apache Beam DoFn 在云中很慢 -python

Question

我想使用 apache beam 将实体更新到数据存储，但在执行 WriteToDatastore 之前，我创建了一个自定义 DoFN - 从导入步骤中获取实体， - 检查数据存储中是否存在实体 - 如果是，提取属性的值并将该值与新实体连接起来。 - 输出实体

示例：我的数据由以下列组成：parent_id、国籍和 child_name。输入数据是新 children 诞生的。在更新到数据存储之前，我想获取 parent 的现有子项并附加新值。

with beam.Pipeline(options=options) as p:

    (p | 'Reading input file' >> beam.io.ReadFromText(input_file_path)
     | 'Converting from csv to dict' >> beam.ParDo(CSVtoDict())
     | 'Create entities' >> beam.ParDo(CreateEntities())
     | 'Update entities' >> beam.ParDo(UpdateEntities())
     | 'Write entities into Datastore' >> WriteToDatastore(PROJECT)
     )

花费最多时间的 Pardo 是更新实体：

class UpdateEntities(beam.DoFn):
"""Updates Datastore entity"""
def process(self, element):
    query = query_pb2.Query()
    parent_key = entity_pb2.Key()
    parent = datastore_helper.get_value(element.properties['parent_id'])
    datastore_helper.add_key_path(parent_key, kind, parent)
    parent_key.partition_id.namespace_id = datastore_helper.get_value(element.properties['nationality'])
    query.kind.add().name = kind
    datastore_helper.set_property_filter(query.filter, '__key__', PropertyFilter.EQUAL, parent_key)


    req = helper.make_request(project=PROJECT, namespace=parent_key.partition_id.namespace_id,query=query)
    resp = helper.get_datastore(PROJECT).run_query(req)

    if len(resp.batch.entity_results) > 0:
        existing_entity = resp.batch.entity_results[0].entity
        existing_child_name_v = datastore_helper.get_value(existing_entity.properties['child_name'])
        new_child_names = existing_child_name_v + ';' + datastore_helper.get_value(element.properties['child_name'])
        datastore_helper.set_value(element.properties['child_name'],new_child_names)
        return [element]
    else:
        return [element]

Answer 1

毫不奇怪，UpdateEntities 是光束流中最慢的部分。您在每次调用 UpdateEntities 时执行 RPC（您应该使用 get/lookup 而不是查询键，因为键查询最终是一致的）。只要您在 UpdateEntities 中执行 RPC，这将是您工作中最慢的部分。

查询 Google 数据存储的 Apache Beam DoFn 在云中很慢 -python

Apache Beam DoFn with query to Google Datastore is slow in cloud -python

python

google-cloud-datastore

google-cloud-dataflow

apache-beam