[Python] Struct column fails to be store in a Big Query table

Description


Hello,

While trying to migrate a part of a code that works with Pandas, I fail to resolve the error below that the code gives me when using a Polars dataframe.

In case of Pandas, I am able to see the results in the destination table.

Does anyone have an idea what could be wrong?

Thank you

Error


google.api_core.exceptions.BadRequest: 400 Error while reading data,
error message:
Schema mismatch: referenced variable 'items.list.item.id_item' has array levels of 1,
while the corresponding field path to Parquet column has 0 repeated fields;
reason: invalid,
message: Error while reading data,
error message:
Schema mismatch: referenced variable 'items.list.item.id_item' has array levels of 1,
while the corresponding field path to Parquet column has 0 repeated fields

Code


import io
from typing import Dict, List, Union
import pandas as pd
import polars as pl
from google.cloud import bigquery


def append_dataframe_to_table(
data: Union[pd.DataFrame, pl.DataFrame],
bq_client: bigquery.Client,
table_schema: List[Dict[str, str]],
destination_table: str,
create_disposition: str,
write_disposition: str,
wait_until_finished: bool,
) -> bigquery.LoadJob:
"""
Append a DataFrame to a BigQuery table.

aram data: Data to be stored.
aram table_schema: Intended schema for the target table. Example:
table_schema = [
{'name': 'id', 'type': 'INT64', 'mode': 'REQUIRED'},
{'name': 'full_name', 'type': 'STRING', 'mode': 'REQUIRED'},
{'name': 'date_of_birth', 'type': 'DATETIME', 'mode': 'NULLABLE'}
]
aram destination_table: Full path of the destination table.
aram create_disposition: Describes conditions when a job should create a table.
Possible values:
|--------------------------------|-------------------------------------------------------------|
| CREATE_DISPOSITION_UNSPECIFIED | Unknown. |
| CREATE_NEVER | This job should never create tables. |
| CREATE_IF_NEEDED | This job should create a table if it doesn't already exist. |

aram write_disposition: Describes whether a mutation to a table should overwrite or append.
Possible values:
|-------------------------------|-----------------------------------------------------------------|
| WRITE_DISPOSITION_UNSPECIFIED | Unknown. |
| WRITE_EMPTY | This job should only be writing to empty tables. |
| WRITE_TRUNCATE | This job will truncate table data and write from the beginning. |
| WRITE_APPEND | This job will append to a table. |

aram wait_until_finished: If true, wait for the job to finish.
"""
job_config = bigquery.LoadJobConfig()
job_config.create_disposition = create_disposition
job_config.write_disposition = write_disposition
job_config.schema = table_schema

if isinstance(data, pl.DataFrame):
job_config.source_format = bigquery.SourceFormat.PARQUET
# Write DataFrame to stream as parquet file; does not hit disk
with io.BytesIO() as stream:
data.write_parquet(stream)
stream.seek(0)
load_job = bq_client.load_table_from_file(
file_obj=stream, destination=destination_table, job_config=job_config
)
else:
load_job = bq_client.load_table_from_dataframe(
dataframe=data, destination=destination_table, job_config=job_config
)

if wait_until_finished:
print("Waiting for load job to finish.")
load_job.result()

return load_job


if __name__ == "__main__":
table_schema = [
{"name": "hash", "type": "STRING", "mode": "REQUIRED"},
{
"name": "items",
"type": "RECORD",
"mode": "REPEATED",
"fields": [
{"name": "id_item", "type": "INT64", "mode": "NULLABLE"},
{"name": "id_category", "type": "INT64", "mode": "NULLABLE"},
{"name": "manufacturer", "type": "STRING", "mode": "NULLABLE"},
],
},
{"name": "prediction", "type": "BOOLEAN", "mode": "NULLABLE"},
{"name": "confidence", "type": "FLOAT64", "mode": "NULLABLE"},
]

df = pl.DataFrame(
{
"hash": ["abcd", "efg"],
"items": [
[
{"id_item": 1234, "id_category": 12, "manufacturer": "A"},
{"id_item": 1235, "id_category": 12, "manufacturer": "B"},
],
[{"id_item": 2345, "id_category": 13, "manufacturer": "C"}],
],
"prediction": [True, False],
"confidence": [0.4, None],
}
)

bq_client = bigquery.Client(project="my-project")

append_dataframe_to_table(
data=df,
bq_client=bq_client,
table_schema=table_schema,
destination_table="my-project.my_dataset.my_test_table",
create_disposition="CREATE_IF_NEEDED",
write_disposition="WRITE_APPEND",
wait_until_finished=True,
)

Continue reading...