Pulling JSONs from Google Cloud Storage, converting into pandas DF, and writing to Google BigQuery

Question

Summary: different types when appending pandas dataframe to BigQuery causing issues with daily ETL process.

I am working on a straight-forward ETL with Airflow: pull data from an API daily, back that raw data up in JSON files in Google Cloud Storage (GCS), and then append the data from GCS into a BigQuery database. I am doing okay with the extract part of the ETL, calling the API and saving the results of each API call (which will be a row in the database table) as its own JSON object in GCS. For a table in BigQuery with 1K rows then, I will first create / save 1K separate objects saved into a bucket in GCS, each the result of an API call.

I am now struggling with the load part of ETL. So far, I have written the following script to do the transfer from GCS to BQ:

# load libraries, connect to google
from google.cloud import storage
import os
import gcsfs
import json
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = '/path/to/my/credentials'

# transfer data
def load_directory_to_bq():

    # get list of filenames from GCS directory
    client = storage.Client()
    files = []
    blobs = client.list_blobs('my-gcs-bucket', prefix='gcs-path-to-files')
    for blob in blobs:
        files.append(f'my-gcs-bucket/{blob.name}')
    

    # approach A: This loop pulls json, converts into df, writes to BigQuery, each 1 file at a time
    fs = gcsfs.GCSFileSystem() # GCP's Google Cloud Storage (GCS) File System (FS)
    for file in files:
        with fs.open(file, 'r') as f:
            gcs_data = json.loads(f.read())
            data = [gcs_data] if isinstance(gcs_data, dict) else gcs_data
            this_df = pd.DataFrame(data)
            pd.DataFrame.to_gbq(this_df, 'my-bq-tablename', project_id='my-gcp-project-id', if_exists='append')


    # approach B: This loop loops all the files, creates 1 large dataframe, and does 1 large insert into BigQuery
    output_df = pd.DataFrame()
    fs = gcsfs.GCSFileSystem() # GCP's Google Cloud Storage (GCS) File System (FS)
    for file in files:
        with fs.open(file, 'r') as f:
            gcs_data = json.loads(f.read())
            data = [gcs_data] if isinstance(gcs_data, dict) else gcs_data
            this_df = pd.DataFrame(data)
            output_df = output_df.append(this_df)

    pd.DataFrame.to_gbq(output_df, 'my-bq-tablename', project_id='my-gcp-project-id', if_exists='append')

The 1K objects in GCS are all similar, but do not always have exactly the same structure:

• nearly all the same keys
• almost always the same "type" for each key

However, for some of the JSON objects, the "types" can be different, for the same key, across different objects. When loaded into python as a 1-row pandas dataframe, the same key key1 may be a float or an integer depending on the value. Also, sometimes a key is missing in an object, or its value/property is null, which can mess up the "type" and cause issues when using the to_gbq function.

With approach A above, the first time an object / pandas DF has a different type, the following error is thrown: Please verify that the structure and data types in the DataFrame match the schema of the destination table. Approach A seems inefficient as well because it calls to_gbq for each of the 1K rows, and each call takes 2-3 seconds.

With approach B, the different "types" issue is seemingly resolved, as pandas handles different "types" in its append function for appending 2 dataframes together. As a result, I get 1 dataframe, and can append it to BigQuery. However, I remain concerned that in the future, there may be new data that I need to append that will not match the type already in the existing-table. After all, I am not querying BigQuery for the old table, appending to new data, and then re-creating the table. I am simply appending new rows, and I am worried that a table with a different "type" for one of the keys will cause an error and break my pipeline then.

In theory, approach A is nice because an approach that can handle any individual row being appended to the table with to_gbq without errors is good. But it requires ensuring the same keys / types for every single row. With approach B, i don't think it's good that python auto-coalesces different types into 1 type for the table, as this can seemingly cause issues down the line with new data coming in.

I am considering what the best approach here would be. As both are Google products, going from GCS to BQ should be straightforward, yet imperfect data makes it slightly tougher. In particular, should I define an explicit table schema somewhere, for each different BQ table, and write a python function that ensures the right types / converts wrong types to right types? Should I recreate the table in BQ each time? Should I avoid Python all-together and transfer from GCS to BQ in another way?

Answer 1

About your approaches A and B, I have the following considerations:

1. If the request is slow and you have a big amount of rows, the approach B will definitely work faster.
2. I don't know your amount of data, but keep in mind that if you have a huge amount you must pay attention in your machine capacity to avoid bad performance and errors.
3. If your process is executed only once per day the time taken to insert all the data into the table might not be a problem at all.
4. As you said, the approach B can avoid the schema problem but theres not a guarantee.

Given that, I'd like to propose the actions below.

1. For the keys that can miss information (or can be NULL) in your files, set the respective field in your BigQuery table as NULLABLE.
2. Using either approach A or B, ensure that the Dataframe have the correct types by using some function that casts your Dataframe columns. You can change the type of a Dataframe column doing for example df.astype({"key1": float, "key2": int, [...]}) as you can find in this reference.

Answer 2

Well, actually you ask about transform stage in your ETL, because the load is evidently done just by pandas.DataFrame.to_gbq() method you already use.

Let's look at you ETL flow in a whole as you describe it:

Source: API -> GCS -> Pandas DataFrame -> Destination: GBQ

Notice:

• what transformations of the data do you perform between API and GCS?

How ever, actually, you have 2 ETL flows here:

1. Source: API -> ?? -> Destination: GCS (JSON objects)
2. Source: GCS (JSON objects) -> Pandas DataFrame -> Destination: GBQ (table)

Practically, the root cause of the data formats variation come from your API as it returns JSON as response. As JSON is schema-less object. Naturally, then this formats variation is propogated into you GCS objects. On the other side as destination you have GBQ table that has strict schema from the creation moment and cannot be altered after.

So, to efficiently load data coming from REST API to GBQ you could follow the such ideas:

1. JSON is a nested data structure and a table is a flat one. So the task is to transform the first one into the second one.

2. Solve this by examining you API Response object and define

   • the widest set of possible fields that can be normalized into flat table schema. Like, all optional fields will come at once.
   • an arrays in your JSON which are it self complex objects and you need very it lo extract and load. Do with them the step 1.

3. Having such flat schema understanding plan to create GBQ tables (separate ones per each object you will actually extract) with all NULLABLE fields.

4. If you use Pandas DataFrame to transformation purpose, then:

   • define the dtypes for your columns explicitly. This allow to avoid problems when pandas dtypes are infered depends on coming data. Note here the the pandas-gbq documentation
   • arrays naturally will transformed into DataFrame and after you will load all records in one GBQ API call.

Also, you can rethink you ETL Flows.

Currently, you said, GCS serves as:

• (a) to have a backup of the raw data
• (b) serves as source of truth for the raw data if there's ever an issue in BQ or elsewhere
• (c) to prevent having to make the same API call twice if there's an issue before uploading to BQ

All of these may be achieved when you load data in parallel both into GCS and GBQ. But you can do this with one common transformation stage.

Source: API -> Pandas DataFrame
    1. |-> Destination: GBQ (table)
    2. |-> Destination: GCS (objects)

The transformation stage you can perform with Pandas DataFrame in follow manner:

1. Nested JSON object into flat table (DataFrame):

   df = pd.json_normalize(api_response_json_object, 'api_response_nested_json_object', sep='_')
   

2. Force field data types:

   def force_df_schema(df, columns_list, columns_dtypes):
       df = df.reindex(columns_list, axis="columns")
       df = df.astype(columns_dtypes)
       return df
   
   API_TRANSACTION_OBJECT_COLUMNS = ['c1', 'c2', 'c3', 'c4']
   API_TRANSACTION_OBJECT_COLUMNS_DTYPES = {
       'c1': 'object',
       'c2': 'datetime64[ns]',
       'c3': 'float64',
       'c4': 'int'
   }
   
   # Let's this call will returns JSON with, for example,
   # {transaction} nested structure, which we need to extract, transform and load 
   api_response_json_object = api.call()
   
   df = pd.json_normalize(api_response_json_object, 
                          'api_response_nested_json_object', sep='_')
   
   df = force_df_schema(df, API_TRANSACTION_OBJECT_COLUMNS,
                            API_TRANSACTION_OBJECT_COLUMNS_DTYPES)
   

3. Load to destination storagees:

to GBQ actually as you already do

 ```
 pd.DataFrame.to_gbq(df, 'bq-tablename', project_id='gcp-project-id', if_exists='append') 
 #also this can create the initial GBQ table,
 #types will be inffered as mentioned in the pandas-bgq docs above.
 ```

to GCS also as you already do.