BigQuery Storage API 并行运行流

Question

我正在尝试使用 BigQuery Storage API 获取一个巨大的 BigQuery 表。目前，我正在使用一个流按顺序获取数据。该程序将在使用数十个虚拟 CPU 的服务器上运行，因此我想并行化表的获取以提高性能。 我正在使用的 bq 存储版本是google.cloud.bigquery.storage.v1 我在this post 中看到可以为 BALANCED 指定分片策略，以便并行计算多个流，但它看起来在 v1 中不存在。

这个选项似乎存在于 v1_beta 中，但我在 code of the repo 中找不到它。 这个选项还存在吗？或者我该如何实现并行会话？

from google.cloud.bigquery_storage import types
from google.cloud import bigquery_storage

def get_df_parallel():
    num_cores = 12
    bqclient = BigQueryClient()
    bqstorageclient = bigquery_storage.BigQueryReadClient(credentials=CREDENTIALS)

    stringify_table = f"..."
    parent = "projects/{}".format(VARIABLES['PROJECT_ID'])
    
    requested_session = types.ReadSession(
        table=stringify_table,
        data_format=types.DataFormat.ARROW,
    )
    read_session = bqstorageclient.create_read_session(
        parent=parent, 
        read_session=requested_session, 
        max_stream_count=num_cores,

        # module 'google.cloud.bigquery_storage_v1.gapic_types' has no attribute 'ShardingStrategy'
        sharding_strategy=(types.ShardingStrategy.BALANCED),
    )
    readers = []
    for stream in read_session.streams:
        position = bigquery_storage.types.StreamPosition(stream=stream)
        reader = bqstorageclient.read_rows(position)
        readers.append(reader)
    df = pd.concat([reader.to_dataframe(session) for reader in readers])
    return df

Answer 1

bigquery_storage_v1beta1 已处于弃用过程中，支持已被删除，因此不存在 python 文档。尽管如果您关注 v1beta1 RPC reference 和 ShardingStrategy 仍然可以访问它，但如果导入，它仍然存在于 v1beta1 中。但是，最好避免使用它，因为该库将在不久的将来使用。

v1beta1 API 尚未正式弃用，将通过 一个完整的弃用周期 (https://cloud.google.com/products#product-launch-stages) 之前 服务被拒绝。但是，新代码应该使用 v1 API 前进。

根据BigQuery storage v1 python docs，您可以使用multiprocessing.Process 创建并行进程。你可以尝试实现这个方法。

因为这个客户端使用了grpcio库，所以共享实例是安全的 跨线程。在多处理场景中，最佳实践是 在调用 os.fork() 之后创建客户端实例 multiprocessing.Pool 或 multiprocessing.Process.