将时间戳从 Dataframe 加载到 BigQuery 数据集

Posted 2023-03-25

【中文标题】将时间戳从 Dataframe 加载到 BigQuery 数据集

【英文标题】：Load timestamp from Dataframe to BigQuery dataset

【发布时间】：2019-09-15 15:13:27

【问题描述】：

我的 BigQuery 表 result_data 中有一个时间戳字段 loaded_at，它与纪元等效 loaded_at_epoch。我正在使用 Python 定期从外部源获取新数据，将这两个字段添加到数据框中，并将此数据框加载到我的 BigQuery 表中。

    loaded_at = datetime.utcnow()
    loaded_at_epoch = int((loaded_at - datetime(1970, 1, 1)).total_seconds()) 
    df['loaded_at'] = pd.Series(loaded_at, index=df.index)
    df['loaded_at_epoch'] = pd.Series(loaded_at_epoch, index=df.index)

    job_config = bigquery.LoadJobConfig()
    job_config.write_disposition = bigquery.WriteDisposition.WRITE_APPEND
    job_config.schema_update_options = [bigquery.SchemaUpdateOption.ALLOW_FIELD_ADDITION]

    bq_client.load_table_from_dataframe(df, result_data, location="US", job_config=job_config,)

它曾经可以工作，但几周后，loaded_at 的值错误，例如 1970-01-19 03:32:09.693 UTC，而 loaded_at_epoch 的时间戳值正确。看起来时间戳以秒为单位，但从数据帧加载时被解释为以毫秒为单位。

我不确定如何进行这项工作。我一直在尝试将 loaded_at 作为字符串，但随后出现错误： google.api_core.exceptions.BadRequest: 400 Provided Schema does not match Table <project_id>:<dataset_id>.result_data. Field loaded_at has changed type from TIMESTAMP to STRING

我也尝试在作业配置中添加job_config.autodetect = False，但也没有解决问题。

知道如何让日期始终有效吗？

谢谢！

【参考方案1】：

您能否尝试将 loaded_at 硬编码以将样本运行到虚拟表中，看看会发生什么？您的代码看起来不错，所以我确信这与基于 pandas 的加载有关。

或者，如果您想避免 pandas 将数据加载到 BigQuery 中，您可以使用 bq cli 为您完成这项工作：

import subprocess
#--you compose your df in this block
# df = ...

loaded_at = datetime.utcnow()
loaded_at_epoch = int((loaded_at - datetime(1970, 1, 1)).total_seconds()) 
df['loaded_at'] = pd.Series(loaded_at, index=df.index)
df['loaded_at_epoch'] = pd.Series(loaded_at_epoch, index=df.index)

#--write the file locally
df.to_csv('temp-data.csv', sep=',', index=False, header=False)

#--load via bq cli
cmd = '''bq --location=US load yourdataset.yourtable temp-data.csv col:type,col:type...'''
subprocess.call(cmd, shell=True)

【参考方案2】：

感谢 Khan，这实际上帮助我弄清楚了如何解决它。我首先尝试使用字符串格式的硬编码时间戳，但遇到了同样的问题。然后我尝试了一个硬编码的熊猫时间戳，它起作用了。下面的代码现在可以工作了。

df['loaded_at'] = pd.Series(pd.Timestamp(loaded_at_epoch, unit='s', tz='UTC'), index=df.index)