JSON table schema to bigquery.TableSchema for BigQuerySink

The above snippet posted by Andrea Pierleoni works with older versions of the google-cloud-bigquery python client, for example for version 0.25.0 of google-cloud-bigquery that happens to install via pip install apache-beam[gcp].

However, the BigQuery Python client API has changed drastically in more recent versions of google-cloud-bigquery, for example in version 1.8.0 that I am currently on, bigquery.TableFieldSchema() and bigquery.TableSchema() don't work.

If you're on a more recent version of the google-cloud-bigquery package, here's how you can get the required SchemaField list (required to create the table, for example) from a JSON file. This is an adaptation of the code posted above by Andrea Pierleoni (thanks for that!)

def _get_field_schema(field):
    name = field['name']
    field_type = field.get('type', 'STRING')
    mode = field.get('mode', 'NULLABLE')
    fields = field.get('fields', [])

    if fields:
        subschema = []
        for f in fields:
            fields_res = _get_field_schema(f)
            subschema.append(fields_res)
    else:
        subschema = []

    field_schema = bigquery.SchemaField(name=name, 
        field_type=field_type,
        mode=mode,
        fields=subschema
    )
    return field_schema


def parse_bq_json_schema(schema_filename):
    schema = []
    with open(schema_filename, 'r') as infile:
        jsonschema = json.load(infile)

    for field in jsonschema:
        schema.append(_get_field_schema(field))

    return schema

Now, suppose you had a table's schema already defined in JSON. Say you had this particular "schema.json" file, then using the above helper methods, you could get the required SchemaField representation for the Python client like so:

>>> res_schema = parse_bq_json_schema("schema.json")

>>> print(res_schema)

[SchemaField(u'event_id', u'INTEGER', u'REQUIRED', None, ()), SchemaField(u'event_name', u'STRING', u'REQUIRED', None, ()), SchemaField(u'event_types', u'STRING', u'REPEATED', None, ()), SchemaField(u'product_code', u'STRING', u'REQUIRED', None, ()), SchemaField(u'product_sub_code', u'STRING', u'REPEATED', None, ()), SchemaField(u'source', u'RECORD', u'REQUIRED', None, (SchemaField(u'internal', u'RECORD', u'NULLABLE', None, (SchemaField(u'name', u'STRING', u'REQUIRED', None, ()), SchemaField(u'timestamp', u'TIMESTAMP', u'REQUIRED', None, ()))), SchemaField(u'external', u'RECORD', u'NULLABLE', None, (SchemaField(u'name', u'STRING', u'REQUIRED', None, ()), SchemaField(u'timestamp', u'TIMESTAMP', u'REQUIRED', None, ()))))), SchemaField(u'timestamp', u'TIMESTAMP', u'REQUIRED', None, ()), SchemaField(u'user_key', u'RECORD', u'REQUIRED', None, (SchemaField(u'device_id', u'STRING', u'NULLABLE', None, ()), SchemaField(u'cookie_id', u'STRING', u'NULLABLE', None, ()), SchemaField(u'profile_id', u'STRING', u'NULLABLE', None, ()), SchemaField(u'best_id', u'STRING', u'REQUIRED', None, ()))), SchemaField(u'message_id', u'STRING', u'REQUIRED', None, ()), SchemaField(u'message_type', u'STRING', u'REQUIRED', None, ()), SchemaField(u'tracking_id', u'STRING', u'NULLABLE', None, ()), SchemaField(u'funnel_stage', u'STRING', u'NULLABLE', None, ()), SchemaField(u'location', u'RECORD', u'NULLABLE', None, (SchemaField(u'latitude', u'FLOAT', u'REQUIRED', None, ()), SchemaField(u'longitude', u'FLOAT', u'REQUIRED', None, ()), SchemaField(u'geo_region_id', u'INTEGER', u'NULLABLE', None, ()))), SchemaField(u'campaign_id', u'STRING', u'NULLABLE', None, ()), SchemaField(u'topic', u'STRING', u'REQUIRED', None, ())]

Now to create a table having the above schema using the Python SDK, you would do:

dataset_ref = bqclient.dataset('YOUR_DATASET')
table_ref = dataset_ref.table('YOUR_TABLE')
table = bigquery.Table(table_ref, schema=res_schema)

You could optionally set time-based partitioning (if needed) like this:

table.time_partitioning = bigquery.TimePartitioning(
    type_=bigquery.TimePartitioningType.DAY,
    field='timestamp'  # name of column to use for partitioning
)

And this finally creates the table:

table = bqclient.create_table(table)

print('Created table {}, partitioned on column {}'.format(
    table.table_id, table.time_partitioning.field))

There is a build-in transformer function in the BigQuery library:

from google.cloud import bigquery
...
client = bigquery.Client()
client.schema_from_json('path/to/schema.json`)

Here is a simple program that can help you.

import json
from apache_beam.io.gcp.internal.clients import bigquery


def bq_schema(json_schema):
    table_schema = bigquery.TableSchema()
    with open(json_schema) as json_file:
        data = json.load(json_file)
        for p in data:
            field = bigquery.TableFieldSchema()
            field.name = p['name']
            field.type = p['type']
            field.mode = p['mode']
            table_schema.fields.append(field)
    return table_schema

Nowadays, you can use the built-in parse_table_schema_from_json function:

from apache_beam.io.gcp.bigquery_tools import parse_table_schema_from_json

with open('schema.json') as f:
    schema_string = f.read()
table_schema = parse_table_schema_from_json(schema_string)