【发布时间】：2022-08-19 07:56:17
【问题描述】：
我有一个大表（BigQuery 外部，因为数据在 Google Cloud Storage 中）。我想使用 BigQuery 将表扫描到客户端计算机。对于吞吐量，我在多个线程中同时获取多个流。
据我所知，并发不起作用。使用多个线程时实际上会有一些惩罚。
import concurrent.futures
import logging
import queue
import threading
import time
from google.cloud.bigquery_storage import types
from google.cloud import bigquery_storage
PROJECT_ID = \'abc\'
CREDENTIALS = {....}
def main():
table = \"projects/{}/datasets/{}/tables/{}\".format(PROJECT_ID, \'db\', \'tb\')
requested_session = types.ReadSession()
requested_session.table = table
requested_session.data_format = types.DataFormat.AVRO
requested_session.read_options.selected_fields = [\"a\", \"b\"]
requested_session.read_options
client = bigquery_storage.BigQueryReadClient(credentials=CREDENTIALS)
session = client.create_read_session(
parent=\"projects/{}\".format(PROJECT_ID),
read_session=requested_session,
max_stream_count=0,
)
if not session.streams:
return
n_streams = len(session.streams)
print(\"Total streams\", n_streams) # this prints 1000
q_out = queue.Queue(1024)
concurrency = 4
with concurrent.futures.ThreadPoolExecutor(concurrency) as pool:
tasks = [
pool.submit(download_row,
client._transport.__class__,
client._transport._grpc_channel,
s.name,
q_out)
for s in session.streams
]
t0 = time.perf_counter()
ntotal = 0
ndone = 0
while True:
page = q_out.get()
if page is None:
ndone += 1
if ndone == len(tasks):
break
else:
for row in page:
ntotal += 1
if ntotal % 10000 == 0:
qps = int(ntotal / (time.perf_counter() - t0))
print(f\'QPS so far: {qps}\')
for t in tasks:
t.result()
def download_row(transport_cls, channel, stream_name, q_out):
try:
transport = transport_cls(channel=channel)
client = bigquery_storage.BigQueryReadClient(
transport=transport,
)
reader = client.read_rows(stream_name)
for page in reader.rows().pages:
q_out.put(page)
finally:
q_out.put(None)
if __name__ == \'__main__\':
main()
Google BigQuery Storage API 文档和多个来源声称可以同时获取多个“流”以获得更高的吞吐量，但我没有找到任何功能示例。我已按照建议在线程之间共享 GRPC“通道”。
数据项很大。我得到的QPS大概是
150, concurrency=1
120, concurrency=2
140, concurrency=4
每个“页面”包含大约 200 行。
想法：
-
BigQuery 配额？我只看到请求速率限制，没有看到数量限制 每秒数据流量。配额似乎并没有限制我的情况。
-
BigQuery 服务器端选项？似乎不相关。 BigQuery 应该接受 具有足够能力的并发请求。
-
GPRC 用法？我认为这是挖掘的主要方向。但我不知道是什么 我的代码错了。
任何人都可以对此有所了解吗？谢谢。
-
您的 CPU 或网络利用率指标是否会随着并发的变化而变化？他们中的任何一个都达到极限了吗？考虑多处理的原因之一是为服务器打开更多通道。上次我experimented with this more channels helped。我猜你的主要问题是下面提到的 gil/python 缓慢。锁定 GIL 并且通常很慢的部分是“for row in page”。最好尝试通过页面大小来衡量吞吐量。或者至少批量转换为箭头并以这种方式测量页面大小
-
我确实使用了带有多个通道的生成的 mp，并且也使用了异步，并且得到了类似的结果。我也跳过了解包页面到行，并得到了稍微不同的结果。
-
网络最大化确实是我应该检查的。我没有仅仅因为我不熟悉那个。另外，我并不急于检查，因为另一个应用程序在整个过程中变得更高（不完全可比，所以我也应该进一步检查）。
-
我还尝试获取更小的字段，并在整个过程中获得更高的计数。并发在那里也没有帮助。它伤害了。难题是为什么并发没有帮助。一个自然的假设是并发不会发生。很难相信这种 grpc 单线程顺序使用会最大化我的网络容量，因此并发无济于事。
-
我认为顺序提取不能最大化网络容量，几乎“根据定义”。
标签： python multithreading google-bigquery grpc google-bigquery-storage-api