Описание
Здравствуйте!
При попытке перенести часть кода, работающего с Pandas, мне не удалось устранить ошибку ниже. код дает мне при использовании кадра данных Polars.
В случае Pandas я могу видеть результаты в целевой таблице.
Да у кого-нибудь есть идеи, что может быть не так?
Спасибо
Ошибка
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
Код
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.
:param data: Data to be stored.
:param 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'}
]
:param destination_table: Full path of the destination table.
:param 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. |
:param 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. |
:param 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,
)
Подробнее здесь: https://stackoverflow.com/questions/781 ... uery-table
