C#环境下高效可扩展地将BigQuery数据迁移至SQL Server的实现方案问询

我现在在开发一款ETL工具，核心需求是从多数据源（比如Snowflake、各类SQL数据库等）抽取数据，然后插入到SQL Server数据库中。针对SQL或Snowflake这类数据源，我可以直接用IDataReader配合SqlBulkCopy来高效写入目标库，但在处理BigQuery（不管是REST还是gRPC客户端）时，遇到了内存占用的瓶颈——因为BQ的原生客户端没有提供开箱即用的IDataReader实现，我只能自己写，但目前的实现逻辑会把全量数据集加载到内存里，这在处理超大规模数据时完全不现实。

我翻遍了GCP/BQ的官方文档，却没找到能解决这个问题的明确方案，所以想请教各位大佬，有没有更高效、可扩展的方式，能把BigQuery的数据迁移到SQL Server，同时避免全量数据驻留内存？

现有实现的核心问题 #

不管是用BQ REST还是gRPC客户端，我目前的做法都是先把所有查询结果全量加载到内存，再封装成自定义的IDataReader实现供SqlBulkCopy使用。但面对TB级别的数据时，这种方式会直接导致内存溢出，完全不具备扩展性。

1. BigQuery REST客户端实现示例 #

var bigQueryClient = await bigQueryClientProvider.GetBigQueryRestClientAsync(sourceCredentialId);
var results = await bigQueryClient.ExecuteQueryAsync(
    selectQuery,
    parameters: [],
    queryOptions: new QueryOptions { UseQueryCache = true },
    cancellationToken: cancellationToken);
// 这里的results会直接加载所有查询结果到内存中

var fieldsMetadata = new List<BigQueryFieldMetadata>();
var columnIndex = 0;
foreach (var field in results.Schema.Fields)
{
    fieldsMetadata.Add(new BigQueryFieldMetadata(field.Name, columnIndex++, field.Type, field.Mode));
}

return (new BigQueryQueryDataReader(results), fieldsMetadata);

// 自定义IDataReader实现，内部持有全量results数据
class BigQueryQueryDataReader : IDataReader 
{ 
    // 具体实现逻辑
    .... 
}

2. BigQuery gRPC客户端实现示例 #

gRPC客户端虽然原生支持流读取，但我目前的实现还是把所有流的行都收集到内存List中，再生成IDataReader，依然没解决内存问题：

var bigQueryGrpcClient = await bigQueryClientProvider.GetBigQueryGrpcClientAsync(sourceCredentialId);
var projectId = await bigQueryClientProvider.GetProjectIdAsync(sourceCredentialId);
var tableNameRef = TableName.FromProjectDatasetTable(projectId, sourceDataset, sourceTableName);

var readSession = bigQueryGrpcClient.CreateReadSession(new CreateReadSessionRequest
{
    Parent = $"projects/{tableNameRef}",
    ReadSession = new ReadSession
    {
        Table = tableNameRef.ToString(),
        DataFormat = DataFormat.Arrow,
        ReadOptions = new ReadSession.Types.TableReadOptions(),
    },
    MaxStreamCount = 1
});

if (sourceSelectedColumns?.Count > 0)
{
    readSession.ReadOptions.SelectedFields.AddRange(sourceSelectedColumns);
    readSession.ReadOptions.RowRestriction = sourceFilterBy;
}

var results = new List<Dictionary<string, object>>();
var fieldsMetadata = new List<BigQueryFieldMetadata>();

foreach (var stream in readSession.Streams)
{
    cancellationToken.ThrowIfCancellationRequested();
    var streamRows = await this.ReadFromStreamAsync(bigQueryGrpcClient, stream.Name, fieldsMetadata, cancellationToken);
    results.AddRange(streamRows); // 把所有流的行全量加载到内存List中
}

return (new BigQueryTableScanDataReader(results, results.FirstOrDefault()?.Keys?.ToList() ?? []), fieldsMetadata);

// 自定义IDataReader实现
class BigQueryTableScanDataReader : IDataReader 
{ 
    // 具体实现逻辑
    .... 
}

我的核心诉求 #

我需要一种不需要把全量BigQuery数据加载到内存的方式，能让SqlBulkCopy可以流式读取BQ的数据，逐批写入SQL Server，同时保持IDataReader的兼容（因为ETL工具的其他逻辑都是基于IDataReader抽象的）。有没有大佬做过类似的实现，或者知道BQ客户端里有什么我没注意到的流式API可以利用？

内容来源于stack exchange