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

Question

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

示例：我的数据由以下列组成：parent_id、nationality & child_name。输入数据是新出生的孩子。在更新插入数据存储之前，我想获取父母现有的孩子并附加新值。

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，它将是您工作中最慢的部分。