小一時間悩んだ
環境
- google-cloud-bigquery 2.30.1 (Python Client for Google BigQuery)
- Python 3.9.0
現象
下記のようなコードで、BigQuery に insert しようとしたところ
foo.py
import json
from google.cloud import bigquery
from google.cloud.bigquery.job import LoadJob
from google.cloud.bigquery.table import Table
if __name__ == '__main__':
a = '{"foo":"123","bar": "123"}'
row = json.loads(a)
project_id = 'foo_project'
dataset_id = 'bar_dataset'
table_id = 'test'
client = bigquery.Client(project = project_id)
dataset = client.dataset(dataset_id)
table = dataset.table(table_id)
table_result: Table = client.get_table(table)
job_config = bigquery.LoadJobConfig()
job_config.source_format = bigquery.SourceFormat.NEWLINE_DELIMITED_JSON
job_config.schema = table_result.schema
table_id_constructor = ".".join([project_id,dataset_id,table_id])
job: LoadJob = client.load_table_from_json(json_rows=row, destination=table_id_constructor, job_config=job_config)
print(job.result())
下記のようなエラーになる(そっけない...)
google.api_core.exceptions.BadRequest: 400 Error while reading data, error message: JSON table encountered too many errors, giving up. Rows: 1; errors: 1. Please look into the errors[] collection for more details.
原因
あれこれ見たりしたがエラーからはよくわからず、結果 load_table_from_json() の json_rows に渡すのは、list型(配列)じゃないとあかん ということがわかった。
というわけで以下のようにすることで insert できましたとさ。
(JSONLを期待してるだろうから、まぁ、そうなんだろうなという気もするが)
foo2.py
import json
from typing import List
from google.cloud import bigquery
from google.cloud.bigquery.job import LoadJob
from google.cloud.bigquery.table import Table
if __name__ == '__main__':
a = '{"foo":"123","bar": "123"}'
row = json.loads(a)
rows: List[dict] = []
rows.append(row) # 変更点: list に append してあげる
project_id = 'foo_project'
dataset_id = 'bar_dataset'
table_id = 'test'
client = bigquery.Client(project = project_id)
dataset = client.dataset(dataset_id)
table = dataset.table(table_id)
table_result: Table = client.get_table(table)
job_config = bigquery.LoadJobConfig()
job_config.source_format = bigquery.SourceFormat.NEWLINE_DELIMITED_JSON
job_config.schema = table_result.schema
table_id_constructor = ".".join([project_id,dataset_id,table_id])
# 変更点: row じゃなくて、 rows を渡す
job: LoadJob = client.load_table_from_json(json_rows=rows, destination=table_id_constructor, job_config=job_config)
print(job.result())