如何将 Csv 文件流式传输到 BigQuery？

Question

到目前为止，我发现的示例是将 json 流式传输到 BQ，例如https://cloud.google.com/bigquery/streaming-data-into-bigquery

如何将 Csv 或任何文件类型流式传输到 BQ？下面是流式传输的代码块，似乎“问题”在 insert_all_data 中，其中“行”定义为 json .. 谢谢

# [START stream_row_to_bigquery]
def stream_row_to_bigquery(bigquery, project_id, dataset_id, table_name, row,
                           num_retries=5):
    insert_all_data = {
        'rows': [{
            'json': row,
            # Generate a unique id for each row so retries don't accidentally
            # duplicate insert
            'insertId': str(uuid.uuid4()),
        }]
    }
    return bigquery.tabledata().insertAll(
        projectId=project_id,
        datasetId=dataset_id,
        tableId=table_name,
        body=insert_all_data).execute(num_retries=num_retries)
    # [END stream_row_to_bigquery]

Answer 1

这就是我wrote 使用bigquery-python 库的方式非常容易。

def insert_data(datasetname,table_name,DataObject):
          client = get_client(project_id, service_account=service_account,
                            private_key_file=key, readonly=False, swallow_results=False)

          insertObject = DataObject
          try:
              result  = client.push_rows(datasetname,table_name,insertObject)
          except Exception, err:
              print err
              raise
          return result

这里的 insertObject 是一个字典列表，其中一个字典包含一行。

例如：[{field1:value1, field2:value2},{field1:value3, field2:value4}]

csv可以读成如下，

import pandas as pd
fileCsv = pd.read_csv(file_path+'/'+filename, parse_dates=C, infer_datetime_format=True)
data = []
for row_x in range(len(fileCsv.index)):
    i = 0
    row = {}
    for col_y in schema:
        row[col_y['name']] = _sorted_list[i]['col_data'][row_x]
        i += 1
    data.append(row)
insert_data(datasetname,table_name,data)

数据列表可以发送到insert_data

这样做可以，但我已经提出了here 的限制。