Python BigQuery 存储。并行读取多个流

Question

我有以下玩具代码:

import pandas as pd
from google.cloud import bigquery_storage_v1beta1
import os
import google.auth

os.environ["GOOGLE_APPLICATION_CREDENTIALS"]='key.json'
credentials, your_project_id = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
bq_storage_client = bigquery_storage_v1beta1.BigQueryStorageClient(credentials=credentials)

table_ref = bigquery_storage_v1beta1.types.TableReference()
table_ref.project_id = "bigquery-public-data"
table_ref.dataset_id = "libraries_io"
table_ref.table_id = "versions"

parent = "projects/{}".format(your_project_id)
session = client.create_read_session(
    table_ref,
    parent,
    format_=bigquery_storage_v1beta1.enums.DataFormat.ARROW,
    sharding_strategy=(bigquery_storage_v1beta1.enums.ShardingStrategy.BALANCED),
)

reader1 = bq_storage_client.read_rows(bigquery_storage_v1beta1.types.StreamPosition(stream=session.streams[0]), timeout=100000)
reader2 = bq_storage_client.read_rows(bigquery_storage_v1beta1.types.StreamPosition(stream=session.streams[1]), timeout=100000)

df = pd.concat([reader1.to_dataframe(session),reader2.to_dataframe(session)])
df

我使用BALANCED ShardingStrategy发起了多个可以独立读取的流。

BigqueryStorage 文档说:

However, if you wanted to fan out multiple readers you could do so by having a reader process each individual stream.

我启动了两个读取器，每个读取器对应 session 中的每个流。然后，两个数据帧(每个读取器创建 1 个)被连接成一个数据帧。然而，与 LIQUID ShardingStrategy 相比，这种方法并没有提供任何加速。

我试图让两个读者并行读取行。但是，我在库文档中找不到有关并行流读取的任何内容。

问题是:

1) 如果选择 BALANCED ShardingStrategy，BugQuery Storage 是否提供同时读取多个流的任何 native 方法？

2)并行读取流的最佳方法是什么？我需要使用多处理或异步吗？

3)如果有人可以提供有关并行流红色的任何基本示例，我将不胜感激

Answer 1

BigQuery Storage API确实支持多个流，但您的执行方法不支持。您可以创建多个读取器实例，然后每个实例都可以使用单独的流来增加吞吐量。

您有很多选择要做Parallel processing in python 。然而，最容易使用的是 multiprocessing package 。

另一个选择是使用 Apache Beam默认情况下支持并行处理，但可能不适合您的用例。它有一个内置的 BigQuery IO 驱动程序，但它的 python 版本尚不支持 BigQuery Storage API，因此您可能必须为 BQ Storage API 编写自己的实现。