api - 使用 BigQuery Storage API(测试版)启动和读取多个流

Question

BigQuery Storage API ( https://googleapis.github.io/google-cloud-python/latest/bigquery_storage/gapic/v1beta1/api.html ) 对于从 BigQuery 表读取数据非常有用，其速度几乎是标准 BigQuery API 的 10 倍。为了使其更快，它支持多个读取流，每个读取流从相关表中读取一组动态分配的行。

我的问题是:虽然您可能请求多个流，但请求后分配的流不在您的控制范围内。因此，我无法启动超过 1 个流。

我正在阅读的数据由 3 列和 600 万行组成，如下所示。我将创建的流总数打印到控制台。

from google.cloud import bigquery_storage_v1beta1

project_id = 'myproject'
client = bigquery_storage_v1beta1.BigQueryStorageClient()

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

read_options = bigquery_storage_v1beta1.types.TableReadOptions()
read_options.selected_fields.append("year")
read_options.selected_fields.append("name")
read_options.selected_fields.append("number")

# I request 3 streams to be created!
requested_streams = 3  

parent = "projects/{}".format(project_id)
session = client.create_read_session(
    table_ref, parent, table_modifiers=modifiers, read_options=read_options, 
    requested_streams=requested_streams
)  

response = client.batch_create_read_session_streams(session, requested_streams)

# I see only 1 stream being created.
print("Streams created: " + str(len(session.streams)))
print("Stream names array: " + str(session.streams))


reader = client.read_rows(
    bigquery_storage_v1beta1.types.StreamPosition(stream=session.streams[0])
)

rows = reader.rows(session)

names = set()

import time
start = time.time()
#---------------------------------------------------
i=0
for row in rows:
    i += 1
    names.add(row["name"])
    if i > 6000000:
        break
#---------------------------------------------------    
end = time.time()
print(end - start)
print("Got {} unique names and {} total rows.".format(len(names), i))

我有几个问题:

1) 我是否只看到 1 个流，因为多流实现不完整(API 处于 Beta 版)？

2) 我是否只看到 1 个流，因为对于流分配算法来说数据相对“小”？ 6m 行已经相当可观了。

3) 如果我开始看到创建的多个流，API 文档没有描述如何并行读取这些流。关于如何做到这一点的任何想法？

Answer 1

问题是您正在读取的表只有一个可用的输入文件。虽然它有 600 万行，但数据是高度可压缩的，因此只有一个支持数据的柱状文件。目前，存储 API 不会比这更精细地拆分数据。

如果您检查从该表中选择的查询计划，您将看到相同的内容(只有一个输入)。