将 DataFrame 加载到 BigQuery 表时出错（pyarrow.lib.ArrowTypeError：<class 'str'> 类型的对象无法转换为 int）

Question

我在 GCS 中存储了一个 CSV 文件，我想将其加载到 BigQuery 表中。但我需要先进行一些预处理，所以我将其加载到 DataFrame，然后加载到 BigQuery 表

import pandas as pd
import json 
from google.cloud import bigquery


cols_name_list = [....]. # column name in order
uri = "gs://<bucket>/<path>/<csv_file>"
df = pd.read_csv(uri, dtype="string")
df =df.reindex(columns=cols_name_list)

client = bigquery.Client()
job_config = bigquery.LoadJobConfig(
   ... # added all schema field according to table column type
)

job = client.load_table_from_dataframe(
    df, "<bq_table_id>", job_config=job_config
)
job.result()

从上面的代码中，我对数据帧列顺序进行了重新排序，以与 BigQuery 表中的顺序相匹配（不确定这是否重要），并将所有列转换为字符串类型。

我收到此错误，如下所示

pyarrow.lib.ArrowInvalid: Could not convert '47803' with type str: tried to convert to int

我还运行了它，但没有强制 dtypes 为字符串，但出现了另一个错误

pyarrow.lib.ArrowTypeError: Expected a string or bytes dtype, got int64

代码和数据看起来都很正常。因此，我尝试降级 numpy 和 pyarrow 的版本，但仍然导致相同的错误。

更新：

我更新了代码以强制仅字符串列

df =pd.read_csv(uri, dtype={"B" : "string"})

这是我使用的 CSV 数据示例

A,B,C,D,E,F,G
47803,000000000020030263,629,,2021-01-12 23:26:37,,

其中BQ表的列类型应该是这样的

job_config = bigquery.LoadJobConfig(
    schema = [
        bigquery.SchemaField("A", "INTEGER"),
        bigquery.SchemaField("B", "STRING"),
        bigquery.SchemaField("C", "INTEGER"),
        bigquery.SchemaField("D", "INTEGER"),
        bigquery.SchemaField("E", "DATETIME"),
        bigquery.SchemaField("F", "INTEGER"),
        bigquery.SchemaField("G", "DATETIME")
    ]
)

现在，当我尝试使用load_table_from_dataframe()这些配置加载数据时，出现此错误

pyarrow.lib.ArrowTypeError: object of type <class 'str'> cannot be converted to int

所以，我打印出 dtypes

A      int64
B      string
C      int64
D      float64
E      object
F      float64
G      float64
dtype: object

现在哪一栏有问题？我该如何解决这个问题？该错误对于调试来说不太有用。因为应该是 int 的列已经是 int 了。唯一具有 string 类型的列不需要转换为 int 但会抛出这样的错误

Answer 1

您必须将source_formatLoadJobConfig 设置为源数据的格式。在这种情况下，您可以autodetect=False按照已明确指定表的架构进行设置。下面是示例，您可以根据该示例更改代码并尝试执行。

job_config = bigquery.LoadJobConfig(
 
    schema = [
        bigquery.SchemaField("A", "INTEGER"),
        bigquery.SchemaField("B", "STRING"),
        bigquery.SchemaField("C", "INTEGER"),
        bigquery.SchemaField("D", "INTEGER"),
        bigquery.SchemaField("E", "DATETIME"),
        bigquery.SchemaField("F", "INTEGER"),
        bigquery.SchemaField("G", "DATETIME")
    ],
    autodetect=False,
    source_format=bigquery.SourceFormat.CSV
)

CSV 数据：

A,B,C,D,E,F,G
47803,000000000020030263,629,785,2021-01-12 23:26:37,986,2022-01-12 23:26:37

输出：