将JSON写入BigQuery表时类型不兼容

import json
from google.cloud import bigquery

# 示例JSON数据
json_data = {
    "id": "123",
    "name": "John Doe",
    "age": "30"
}

# 将JSON数据转换为字典
data_dict = json.loads(json_data)

# 数据类型转换
data_dict["id"] = int(data_dict["id"])
data_dict["age"] = int(data_dict["age"])

# BigQuery表结构
schema = [
    bigquery.SchemaField("id", "INTEGER"),
    bigquery.SchemaField("name", "STRING"),
    bigquery.SchemaField("age", "INTEGER")
]

# 写入BigQuery表
client = bigquery.Client()
table_ref = client.dataset("your_dataset").table("your_table")
table = bigquery.Table(table_ref, schema=schema)
rows_to_insert = [data_dict]
errors = client.insert_rows(table, rows_to_insert)

if errors == []:
    print("New rows have been added.")
else:
    print("Encountered errors while inserting rows: {}".format(errors))

# 示例嵌套JSON数据
nested_json_data = {
    "id": "123",
    "name": "John Doe",
    "address": {
        "street": "123 Main St",
        "city": "Anytown"
    }
}

# 将JSON数据转换为字典
nested_data_dict = json.loads(nested_json_data)

# BigQuery表结构（嵌套）
nested_schema = [
    bigquery.SchemaField("id", "INTEGER"),
    bigquery.SchemaField("name", "STRING"),
    bigquery.SchemaField("address", "RECORD", fields=[
        bigquery.SchemaField("street", "STRING"),
        bigquery.SchemaField("city", "STRING")
    ])
]

# 写入BigQuery表
table_ref = client.dataset("your_dataset").table("your_table")
table = bigquery.Table(table_ref, schema=nested_schema)
rows_to_insert = [nested_data_dict]
errors = client.insert_rows(table, rows_to_insert)

if errors == []:
    print("New rows have been added.")
else:
    print("Encountered errors while inserting rows: {}".format(errors))

# 示例JSON数据包含空值
json_data_with_null = {
    "id": "123",
    "name": None,
    "age": "30"
}

# 将JSON数据转换为字典
data_dict_with_null = json.loads(json_data_with_null)

# 处理空值
data_dict_with_null["name"] = ""  # 使用默认值

# 写入BigQuery表
rows_to_insert = [data_dict_with_null]
errors = client.insert_rows(table, rows_to_insert)

if errors == []:
    print("New rows have been added.")
else:
    print("Encountered errors while inserting rows: {}".format(errors))