ETL 从谷歌云存储加载到 bigquery答案

【问题标题】：ETL load from google cloud storage to biqueryETL 从谷歌云存储加载到 bigquery
【发布时间】：2018-10-30 11:36:26
【问题描述】：

我想从谷歌云存储上的数百个 CSV 文件中加载数据，并每天使用云数据流（最好使用 python SDK）将它们附加到 Bigquery 上的单个表中。你能告诉我我是怎么做到的吗？

谢谢

【问题讨论】：

问题到底是什么？是的，您可以使用 Python SDK for Dataflow 加载到 BigQuery 表。或者（我的偏好）只是加载到 BigQuery，然后运行查询进行转换。
嗨艾略特！我想运行一个每日批处理作业，将云存储中的 csv 文件中的数据加载到 bigquery 表中。我同意你的观点，但即使是直接加载，数据流代码也需要将 csv 行数据转换为 JSON（键值对）要写入 bigquery..bigquery 仅在您使用 apache Beam（数据流 SDK）时接受键值对
听起来很像this answer。
是的，但它在 java..我更熟悉 python。 python 将 pcollection 转换为 tablerow 的任何资源？
stackoverflow.com/a/47379508/6253347

标签： python google-cloud-platform google-bigquery google-cloud-storage dataflow

【解决方案1】：

我们也可以通过 Python 来实现。请找到以下代码 sn-p。

def format_output_json(element):
    """
    :param element: is the row data in the csv
    :return: a dictionary with key as column name and value as real data in a row of the csv.

    :row_indices: I have hard-coded here, but can get it at the run time.
    """
    row_indices = ['time_stamp', 'product_name', 'units_sold', 'retail_price']
    row_data = element.split(',')
    dict1 = dict()
    for i in range(len(row_data)):
        dict1[row_indices[i]] = row_data[i]

    return [dict1]

【讨论】：