BigQuery 的 to_arrow_iterable 每次只返回 8 行，如何调整？

2026-11-04数据阿明

page_size=10000 不能保证 to_arrow_iterable() 每次返回 10000 行，也不能用来设置每批至少返回 1000 行。先通过 batch.num_rows 确认实际行数；如果批次确实很小，可以在客户端合并后再交给 Polars 处理。仅凭现有代码，无法认定原因是 BigQuery Storage API 限流。

为什么 page_size 没有达到预期？

page_size 和 Arrow 批次的行数受不同机制控制：

page_size 指定分页结果的每页最大行数，不保证每页填满。
使用 BigQuery Storage API 时，数据以 Arrow 批次返回，REST 分页参数不能指定这些批次的行数。

to_arrow_iterable() 没有用于设置目标批次行数的 batch_size 参数。它提供的 max_queue_size 控制预取队列大小，max_stream_count 控制最大并行下载流数，都不表示每批的行数。参见 RowIterator 官方文档。

先确认“8 行”是否为实际行数

如果“8 行”来自打印出的 DataFrame，先排除显示截断。Polars 可以限制表格展示的行数，屏幕上可见的行数不一定等于 DataFrame 的实际行数。参见 Polars 表格显示配置。

在原来的循环中临时记录前几个批次的统计信息：

for index, batch in enumerate(self._batch_arrow_iterator):
    chunk_df = pl.from_arrow(batch)
    print(
        "batch_index=", index,
        "arrow_rows=", batch.num_rows,
        "arrow_bytes=", batch.nbytes,
        "polars_shape=", chunk_df.shape,
    )
    if index >= 4:
        break

重点看 batch.num_rows 和 chunk_df.height。如果两者都远大于 8，应继续排查转换和后续处理的耗时；如果两者都是 8，说明实际批次确实很小。这段检查会提前终止迭代，正式处理时需要重新创建结果迭代器。

如何按 1000 行交给 Polars 处理？

可以继续使用 Storage API 读取数据，在应用层累积小批次并拆分过大的批次。下面的生成器每次产出一个包含 1000 行的 Arrow Table，最后一批允许不足 1000 行：

import pyarrow as pa
import polars as pl
from google.cloud import bigquery_storage


def iter_arrow_tables(batches, target_rows=1000):
    if target_rows <= 0:
        raise ValueError("target_rows must be positive")

    pending = []
    pending_rows = 0

    for batch in batches:
        offset = 0
        while offset < batch.num_rows:
            take = min(
                target_rows - pending_rows,
                batch.num_rows - offset,
            )
            pending.append(batch.slice(offset, take))
            pending_rows += take
            offset += take

            if pending_rows == target_rows:
                yield pa.Table.from_batches(pending)
                pending = []
                pending_rows = 0

    if pending:
        yield pa.Table.from_batches(pending)


# client 和 query 沿用现有代码中的对象。
query_job = client.query(query)
rows = query_job.result()
storage_client = bigquery_storage.BigQueryReadClient()

try:
    batches = rows.to_arrow_iterable(
        bqstorage_client=storage_client,
        max_queue_size=2,
        max_stream_count=1,
    )

    for table in iter_arrow_tables(batches, target_rows=1000):
        chunk_df = pl.from_arrow(table)
        # 在这里处理或写出 chunk_df，避免累计保存所有批次。
finally:
    storage_client.close()

RecordBatch.slice() 支持零拷贝切片，Table.from_batches() 可以把相同 schema 的多个批次组织成一个 Table。这里每次只将当前处理批次传给 from_batches()，不会收集整个查询结果。参见 Arrow RecordBatch 文档和 Arrow Table 文档。

这种方式调整了应用层的处理粒度，可以减少频繁创建小 DataFrame 的开销。它不会让服务器改为每次发送 1000 行，也不能保证网络下载提速。

实际批次仍很小时，继续检查什么？

检查实际 schema 和批次字节数。“约 40 个 float32 列”是对数据的描述，诊断时应以 batch.schema、batch.num_rows 和 batch.nbytes 为准。
确认实际读取路径。传入客户端后，仍应结合客户端日志或 Cloud Monitoring 的 ReadRows 调用，确认是否使用了 Storage API，避免仅根据调用参数推断。Google 提供了 ReadRows 延迟监控说明。
区分下载、转换和业务处理耗时。先测量只消费 Arrow 批次的速度，再与加入 pl.from_arrow() 和业务操作后的速度比较，避免把转换或逐批写出的开销归因于下载。
记录依赖版本和可复现查询。如果统计确认大量批次确实只有 8 行，保留 google-cloud-bigquery、google-cloud-bigquery-storage、pyarrow 和 polars 的版本，以及脱敏 SQL、schema 和批次统计，以便进一步判断原因。

示例中的队列大小和单流设置用于限制预取资源，并非通用的性能最优值，可以根据内存和吞吐量调整。如果查询通过 ORDER BY 确定结果顺序，并行流数还会受到限制。

1000 行只是输出批次大小，不是进程的严格内存上限。预取队列、当前原始批次和 Polars 转换都可能占用额外内存，Arrow 切片也可能继续引用较大的原始缓冲区。