将CSV上传至BigQuery时遇时间解析错误，求解决方法

问题原因 #

BigQuery的TIME类型仅支持00:00:00到23:59:59的合法时间范围，而你的CSV中global_time_for_first_response_goal字段存在80:00:00这类超过24小时的时长值，无法被解析为TIME类型，导致上传失败。

解决方案 #

方案1：修改目标字段类型（推荐） #

将BigQuery表中global_time_for_first_response_goal字段的类型改为STRING（直接存储原始字符串）或INT64（存储总秒数，更适合时长统计），同时在加载代码中指定正确的Schema，关闭自动检测：

def upload_csv_bigquery_dataset():
    client      = bigquery.Client()
    table_id    = "myproject-dev.tickets.ticket"
    # 手动指定Schema，修改目标字段类型
    job_config  = bigquery.LoadJobConfig(
        write_disposition     = bigquery.WriteDisposition.WRITE_TRUNCATE,
        source_format         = bigquery.SourceFormat.CSV,
        schema                = [
            # 示例：保留其他字段，修改目标字段为INT64（存储总秒数）
            # 替换为你实际的表字段定义
            bigquery.SchemaField("ticket_id", "STRING"),
            bigquery.SchemaField("global_time_for_first_response_goal", "INT64"),
            # ... 其他字段依次定义
        ],
        skip_leading_rows     = 1,
        autodetect            = False,  # 关闭自动检测，使用指定Schema
        allow_quoted_newlines = True
    )
    uri = "gs://mybucket/mytickets/2023-02-1309:58:11:865588.csv"
    load_job = client.load_table_from_uri(uri, table_id, job_config=job_config)
    load_job.result()

    destination_table = client.get_table(table_id)
    print(">>> Loaded {} rows.".format(destination_table.num_rows))

如果选择INT64类型，后续可以通过BigQuery函数将秒数转换为可读时长：

-- 将总秒数转换为HH:MM:SS格式
FORMAT_TIMESTAMP('%H:%M:%S', TIMESTAMP_SECONDS(global_time_for_first_response_goal))

方案2：预处理CSV转换非法格式 #

在上传前用Python处理CSV，将超过24小时的时长转换为总秒数，再上传到BigQuery的INT64字段：

import csv
from google.cloud import storage

def preprocess_csv():
    storage_client = storage.Client()
    bucket = storage_client.get_bucket("mybucket")
    source_blob = bucket.blob("mytickets/2023-02-1309:58:11:865588.csv")
    # 读取CSV内容
    lines = source_blob.download_as_text().splitlines()
    reader = csv.reader(lines)
    header = next(reader)
    processed_rows = [header]

    # 处理每一行，假设目标字段是第36位（索引35，从0开始）
    for row in reader:
        time_str = row[35]
        if ':' in time_str:
            parts = time_str.split(':')
            if len(parts) == 3:
                hours, mins, secs = map(int, parts)
                total_seconds = hours * 3600 + mins * 60 + secs
                row[35] = str(total_seconds)
        processed_rows.append(row)
    
    # 上传处理后的文件到GCS
    processed_blob_path = "mytickets/processed_2023-02-1309:58:11:865588.csv"
    processed_blob = bucket.blob(processed_blob_path)
    processed_blob.upload_from_string('\n'.join([','.join(row) for row in processed_rows]))
    return f"gs://mybucket/{processed_blob_path}"

# 调用预处理后再上传
def upload_csv_bigquery_dataset():
    processed_uri = preprocess_csv()
    client = bigquery.Client()
    table_id = "myproject-dev.tickets.ticket"
    job_config = bigquery.LoadJobConfig(
        write_disposition     = bigquery.WriteDisposition.WRITE_TRUNCATE,
        source_format         = bigquery.SourceFormat.CSV,
        schema                = [
            # 目标字段设为INT64
            bigquery.SchemaField("global_time_for_first_response_goal", "INT64"),
            # ... 其他字段定义
        ],
        skip_leading_rows     = 1,
        allow_quoted_newlines = True
    )
    load_job = client.load_table_from_uri(processed_uri, table_id, job_config=job_config)
    load_job.result()

    destination_table = client.get_table(table_id)
    print(">>> Loaded {} rows.".format(destination_table.num_rows))

方案3：临时忽略错误行（不推荐） #

如果允许丢失少量错误数据，可以在加载配置中设置允许的错误记录数，跳过无法解析的行：

job_config  = bigquery.LoadJobConfig(
    # 其他配置不变
    max_bad_records=10,  # 设置允许跳过的错误行数
    autodetect            = True
)

内容的提问来源于stack exchange，提问作者xis10z