BigQuery: Upload pandas DataFrame containing arrays

[AETDDraper]

The support for python Bigquery API indicates that arrays are possible, however, when passing from a pandas dataframe to bigquery there is a pyarrow struct issue.

The only way round it seems its to drop columns then use JSON Normalise for a separate table.

from google.cloud import bigquery

project = 'lake'
client = bigquery.Client(credentials=credentials, project=project)
dataset_ref = client.dataset('XXX')
table_ref = dataset_ref.table('RAW_XXX')
job_config = bigquery.LoadJobConfig()
job_config.autodetect = True
job_config.write_disposition = 'WRITE_TRUNCATE'

client.load_table_from_dataframe(appended_data, table_ref,job_config=job_config).result()

This is the error recieved. NotImplementedError: struct

The reason I wanted to use this API as it indicates Nested Array support, which is perfect for our data lake in BQ but I assume this doesn't work?

[tswast]

Thanks for the report. Struct support is reported at https://github.com/googleapis/google-cloud-python/issues/8191, but I'll keep this issue open as a feature request for arrays of scalar types.

[sumit-ql]

I am not able to repro this issue with dataframe containing arrays only. Its not giving any exception.

i can see an exception when we use dict(STRUCT/RECORD) in dataframe.

@AETDDraper Could you please provide a sample dataframe containing arrays which causes the error.

google-cloud-bigquery 1.20.0
pyarrow 0.14.0
pandas 0.25.1

[JonathanBonnaud]

Loading a dataframe containing an array (a Python type List) does work if you don't specific the schema in the LoadJobConfig() and just let it detect it with autodetect=True. You just get a warning from pyarrow: UserWarning: Pyarrow could not determine the type of columns: COLUMN_NAME. ", ".join(field.name for field in unknown_type_fields)

df = pd.DataFrame([{'uuid':'test1', 'class':['cls1', 'cls1']}])
job_config = bigquery.LoadJobConfig(
	autodetect=True
)

bqclient.load_table_from_dataframe(
	df,
	"DATASET.TABLE",
	job_config=job_config
)

The detected schema looks like this:

field_name	type	mode
class	RECORD	NULLABLE
class.list	RECORD	REPEATED
class.list.item	STRING	NULLABLE

There is no way to provide a schema with the ARRAY type though, which is a bit frustrating. Or even RECORD or STRUCT type.

My config:

google-cloud-bigquery==1.24.0
pandas==1.0.1
pyarrow==0.16.0

[plamut]

There is no way to provide a schema with the ARRAY type though, which is a bit frustrating. Or even RECORD or STRUCT type.

FWIW, uploading STRUCT columns through a dataframe will become possible when a fix in the pyarrow dependency will be released. I tried that with the development pyarrow version and it worked.

At least uploading REPEATED fields (even those consisting of STRUCTs) works with the load_table_from_json() method, if that workaround is feasible. I managed to successfully upload the following data with it:

schema = [
    bigquery.SchemaField(
        "bar",
        "STRUCT",
        fields=[
            bigquery.SchemaField("aaa", "INTEGER", mode="REQUIRED"),
            bigquery.SchemaField("bbb", "INTEGER", mode="REQUIRED"),
        ],
        mode="REPEATED",
    ),
]

json_data = [
    {"bar": [{"aaa": 1, "bbb": 2}, {"aaa": 10, "bbb": 20}]},
    {"bar": [{"aaa": 3, "bbb": 4}, {"aaa": 5, "bbb": 6}]},
]

job_config = bigquery.LoadJobConfig(schema=schema)
client.load_table_from_json(json_data, table_ref, job_config=job_config).result()

[aaaaahaaaaa]

As far as I can tell, the use case of loading a dataframe that contains an array (no struct) doesn't work even with the new version of Pyarrow (0.17.1).

Dataframe looks like this:

    field_A        field_B
0   1000135057     [-0.04591, -0.00742, ...
1   1100299862     [-0.15857, -0.17590, ...
2   1100174775     [0.075828, -0.1888, ...
3   1300002087     [0.097694, 0.06120, ...
4   1100190614     [0.097450, -0.10271, ...

Loading the dataframe:

schema = [
    bigquery.SchemaField("field_A", "INTEGER"),
    bigquery.SchemaField("field_B", "FLOAT", "REPEATED"),
]

table = bigquery.Table(table_id, schema=schema)
bq_client.create_table(table, exists_ok=True)

job_config = bigquery.LoadJobConfig(
    schema=schema,
    destination=table_id,
    write_disposition="WRITE_TRUNCATE",
)
job = bq_client.load_table_from_dataframe(data, table_id, job_config=job_config)
job.result()

Produces:

google.api_core.exceptions.BadRequest: 400 Incompatible type for field 'field_B'. Expected: 'FLOAT64', actual: 'STRUCT'.

[HemangChothani]

@aaaaahaaaaa A use case of loading a dataframe that contains an array is not supported by bigquery and raised an error Field is specified as REPEATED in provided schema which does not match REQUIRED as specified in the file, it is due to dataframe to parquest file conversion and for more information you can see #17 old issue related to that.

Parquet limitation note for REPEATED fields has added and you could found in documentation of load_table_from_dataframe when BigQuery's 1.26.0 version will release.

Instead of using load_table_from_dataframe you can use load_table_from_json

example:

    from google.cloud import bigquery
    import pandas
    client = bigquery.Client()

    schema = [bigquery.SchemaField("nested_repeated", "INTEGER", mode="REPEATED")]
    job_config = bigquery.LoadJobConfig(schema=schema)
    data = [{"nested_repeated": record}]
   
    client.load_table_from_json(data, "table_id", job_config=job_config).result()

[abuckenheimer]

So just to confirm my read here, am I correct in saying:

There is no way to use a REPEATED type with the load_table_from_dataframe function right now. This is due to how parquet serializes dataframes which is ultimately how this library serializes frames for transfer to bigquery. The current workaround is to use load_table_from_json. I put together a test to showcase how different methods breakdown here (run with code at master af2c987):

import sys

import pytest
import typing
import pandas as pd
from google.cloud.bigquery import Client, schema, job

T = typing.TypeVar("T")


def construct_repeated(
    lst: typing.List[T],
) -> typing.Mapping[str, typing.List[typing.Mapping[str, T]]]:
    """ when bigquery reads parquet arrays it ultimately structures the schema to look like this struct<list array<struct<item i>>>  """
    return {"list": [{"item": obj} for obj in lst]}


def destruct_repeated(
    dct: typing.Mapping[str, typing.List[typing.Mapping[str, T]]]
) -> typing.List[T]:
    return [d["item"] for d in dct["list"]]


def apply_col(
    col: str, callable: typing.Callable[[T], typing.Any]
) -> typing.Callable[[pd.DataFrame], pd.DataFrame]:
    def wrapper(df: pd.DataFrame) -> pd.DataFrame:
        df[col] = df[col].apply(callable)
        return df

    return wrapper


repeated_df = pd.DataFrame({"id": ["a", "b"], "targets": [[1], [2, 3]]})

repeated_schema = [
    schema.SchemaField("id", "STRING", "NULLABLE", None, (), None),
    schema.SchemaField("targets", "INTEGER", "REPEATED", None, (), None),
]

nested_df = repeated_df.copy()
nested_df["targets"] = nested_df["targets"].apply(construct_repeated)

nested_schema = [
    schema.SchemaField("id", "STRING", "NULLABLE", None, (), None),
    schema.SchemaField(
        "targets",
        "RECORD",
        "NULLABLE",
        None,
        (
            schema.SchemaField(
                "list",
                "RECORD",
                "REPEATED",
                None,
                (schema.SchemaField("item", "INTEGER", "NULLABLE", None, (), None),),
                None,
            ),
        ),
        None,
    ),
]


@pytest.fixture
def client():
    client = Client()
    yield client
    client.close()


@pytest.fixture
def dataset(client):
    d = client.create_dataset("test_dataset")
    yield d
    client.delete_dataset(d)


@pytest.fixture
def table(client, dataset):
    table = dataset.table("test_table")
    yield table
    client.delete_table(table, not_found_ok=True)


@pytest.mark.parametrize(
    "df, schema, post",
    [
        # FAILS WITH:
        # google.api_core.exceptions.BadRequest: 400 Error while reading data,
        # error message: Provided schema is not compatible with the file
        # 'prod-scotty-81de9f48-1a4f-424e-be99-2258b706fc46'. Field 'targets' is
        # specified as REPEATED in provided schema which does not match REQUIRED as
        # specified in the file.
        (repeated_df, repeated_schema, None),
        # FAILS comparing `df.to_dict() == got_back.to_dict()`:
        # doesn't upload structs properly, got_back ->
        # id targets
        # 0  a    None
        # 1  b    None
        (nested_df, nested_schema, None),
    ],
)
def test_roundtrip_from_dataframe(df, schema, post, client, table):
    client.load_table_from_dataframe(
        df, table, job_config=job.LoadJobConfig(schema=schema)
    ).result()

    got_back = client.list_rows(table).to_dataframe()
    if post is not None:
        got_back = post(got_back)

    assert df.to_dict() == got_back.to_dict()


@pytest.mark.parametrize(
    "df, schema, post",
    [
        # FAILS WITH:
        # google.api_core.exceptions.BadRequest: 400 Error while reading data,
        # error message: Provided schema is not compatible with the file
        # 'prod-scotty-81de9f48-1a4f-424e-be99-2258b706fc46'. Field 'targets' is
        # specified as REPEATED in provided schema which does not match REQUIRED as
        # specified in the file.
        (repeated_df, repeated_schema, None),
        # FAILS comparing `df.to_dict() == got_back.to_dict()`:
        # doesn't upload structs properly, got_back ->
        # id targets
        # 0  a    None
        # 1  b    None
        (nested_df, nested_schema, None),
    ],
)
def test_roundtrip_from_dataframe_existing_table(df, schema, post, client, table):
    table.schema = schema
    client.create_table(table)
    client.load_table_from_dataframe(df, table).result()

    got_back = client.list_rows(table).to_dataframe()
    if post is not None:
        got_back = post(got_back)

    assert df.to_dict() == got_back.to_dict()


@pytest.mark.parametrize(
    "df, schema, post",
    [
        # works but effectively applies construct_repeated on targets elements
        (repeated_df, repeated_schema, apply_col("targets", destruct_repeated)),
    ],
)
def test_roundtrip_from_dataframe_no_schema(df, schema, post, client, table):
    client.load_table_from_dataframe(df, table).result()
    got_back = client.list_rows(table).to_dataframe()

    if post is not None:
        got_back = post(got_back)

    assert df.to_dict() == got_back.to_dict()


@pytest.mark.parametrize(
    "df, schema, post",
    [
        # to_dataframe here returns an `np.array` type for lists which you can't compare
        # with a `list` type so we convert them for comparison
        (repeated_df, repeated_schema, apply_col("targets", list)),
        (nested_df, nested_schema, apply_col("targets", lambda d: {"list": list(d['list'])})),
    ],
)
def test_roundtrip_from_json(df, schema, post, client, table):
    client.load_table_from_json(
        df.to_dict(orient="records"), table, job_config=job.LoadJobConfig(schema=schema)
    ).result()

    got_back = client.list_rows(table).to_dataframe()
    if post is not None:
        got_back = post(got_back)

    assert df.to_dict() == got_back.to_dict()


if __name__ == "__main__":
    pytest.main([__file__, *sys.argv[1:]])

[tswast]

Pyarrow 2.0 released with improvements to the Parquet serialization. We should revisit to see if this issue can be resolved with pyarrow 2.0

[HemangChothani]

@tswast I have tried with these two following examples and it's working fine.

Example 1:

import pandas as pd
import numpy as np
from google.cloud import bigquery

client = bigquery.Client()

job_config = bigquery.LoadJobConfig(autodetect=True)
table_id = bigquery.Table('table') 
table_id = client.create_table(table, exists_ok=True)
df = pd.DataFrame({'A': [np.array([1,2,3]), np.array([4,5,6]), np.array([7,8,9])]})
job = client.load_table_from_dataframe(df,"table_id",job_config=job_config).result()

Example 2:

table_id_1 = bigquery.Table('table'_1) 
table_id_1= client.create_table(table, exists_ok=True)
job_config = bigquery.LoadJobConfig(autodetect=True)
df = pd.DataFrame([{'uuid':'test1', 'class':[['cls1','class2'], ['cls1','class3']]}])
job = client.load_table_from_dataframe(df,"table_id_1",job_config=job_config).result()

[tswast]

Let's add some system test cases and/or code samples for this (which are skipped unless we have pyarrow >=2.0) before we close this issue out.

I imagine it'd be useful to have samples which show more complex dataframes in the following docs:

• https://cloud.google.com/bigquery/docs/bigquery-storage-python-pandas
• https://cloud.google.com/bigquery/docs/pandas-gbq-migration