将 csv 文件加载到 BIGquery 中并在插入数据时添加日期列

Posted 2023-03-25

【中文标题】将 csv 文件加载到 BIGquery 中并在插入数据时添加日期列

【英文标题】：Load csv files into BIGquery and adding a date column while insert data

【发布时间】：2021-08-04 17:49:26

【问题描述】：

我正在编写一个 python 云函数，在添加新列 creation_date 后将 csv 文件加载到 BigQuery 中。到现在都没有成功。有没有办法使用云功能或熊猫来实现这一点。 任何帮助将不胜感激。 在添加日期列后，我已经浏览了生成 csv 文件并保存在 GCS 中的其他链接。我的要求是不要创建任何额外的文件。你认为 pandas 会是一个不错的选择吗？请建议。

谢谢 丽图

【参考方案1】：

是的，可以使用 CloudFunction 来实现。

您可以做什么，将 csv 文件下载到 Cloud Function docker 实例（/tmp 目录），加载到 pandas 数据框，然后您可以根据需要操作数据（添加/删除列/行等） .

一旦数据准备好加载到 BQ 中，您可以使用以下方法：

load_job = client.load_table_from_dataframe(
             dataframe, table_id, job_config=job_config
           )

更新：

我看到 Pandas 现在支持 gs:// 直接从 GCS 加载。

df = pd.read_csv('gs://bucket/your_path.csv')

参考：https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.read_csv.html

【讨论】：

我们可以传入blob文件而不是/tmp

是的，你可以。我刚刚对其进行了测试并检查了文档。我在回答中举了一个例子。

【参考方案2】：

您必须在 Big Query 中创建数据集和表，以及每列的类型。然后，在数据框中定义一个“创建时间”列，creation_date 变量：

import pandas as pd

creation_date = pd.Timestamp.now() # for each entry in the table

然后，将您的数据框保存到 Big Query，same names of pandas columns，带有特定的列名称和df，your_pandas_dataframe：

from google.cloud import bigquery

client = bigquery.Client()

job_config = bigquery.LoadJobConfig(
    schema=[
        bigquery.SchemaField("DATE", bigquery.enums.SqlTypeNames.DATE), # create each column in Big Query along with types
        bigquery.SchemaField("NAME_COLUMN_2", bigquery.enums.SqlTypeNames.STRING),
        bigquery.SchemaField("NAME_COLUMN_3", bigquery.enums.SqlTypeNames.STRING),
        bigquery.SchemaField("NAME_COLUMN_4", bigquery.enums.SqlTypeNames.INTEGER),
    ],
    write_disposition="WRITE_APPEND",
)

job = client.load_table_from_dataframe(
    your_pandas_dataframe, 'project.dataset.table', job_config=job_config
) 

【参考方案3】：

是的，您绝对可以为此使用 pandas。这是我经过测试的有效示例

import pandas as pd
from datetime import datetime

# df = pd.from_csv('ex.csv')
df = pd.DataFrame('test': ['one', 'two'])
data = []
for i in range(0, df.shape[0]):
       if i == 0:
              data.append(str(datetime. today()).split(".")[0])
       else:
              data.append("")
df['creation_date'] = data
print(df)
# df.to_csv('temp/save.csv')

【讨论】：

我厌倦了你的选择，但我可以看到 creation_date 没有被添加到所有行，而只是添加到第一行