I have recently undertaken the task of rebuilding our analytics stack at Sidelines. I’ll share more details as to what motivated me to do so and why in a later post. One of the solutions I’m exploring, is streaming data to be queried by Google BigQuery. The basic premise is that each event will have its own table and each event carries a lot of properties to give it context. For example, a click on a brand creative will contain properties describing the user that clicked, such as browser, screen resolution, etc. It will also have properties around the type of creative; was an image used? What was the resolution? How long was the title? What was the template. Overall, each event averages to about 50 different properties. The problem I faced, is that Google BigQuery requires the table schema to be defined prior to inserting data. Some other services I’m exploring like Keen.io are kind enough to do the schema inference for you. Because I really didn’t want to define schemas for all these events, I wrote a quick script that will create the JSON required to create the table prior to inserting the first event.
def infer_schema(self, event_data):
"""
Return a Schema compatible with BigQuery
:param event_data: A dictionary with the data we want to send. Be sure to leave dates as objects so we can
infer they are timestamps
:returns: a dictionary that can be passed as the `fields` parameter
"""
def _bg_type(key, type):
"""
Just a convenience method to avoid boilerplate code
"""
return {'type': type,
'description': key,
'mode': 'NULLABLE',
'name': key.lower().replace(' ', '_')}
def _any(k, data):
"""
For a given key,value pair, return the Field Schema. If the value is a dictionary, recursively parse
that our as well. If it is a list, rely on the first element, BigQuery does not make sense with mixed
data types anyway
"""
if isinstance(data, list):
field_schema = _any(k, data[0]) if len(data) > 0 else _any(k, '').update({'mode': 'REPEATED'}) # force a string
field_schema.update({'mode': 'REPEATED'})
return field_schema
elif isinstance(data, dict):
field_schema = _bg_type(k, 'RECORD')
field_schema.update({'fields': [_any(k,v) for k,v in data.iteritems()]})
return field_schema
elif isinstance(data, Decimal) or isinstance(data, float):
return _bg_type(k, 'FLOAT')
elif isinstance(data, bool): # this needs to be higher than int because bool is also int
return _bg_type(k, 'BOOLEAN')
elif isinstance(data, int):
return _bg_type(k, 'INTEGER')
elif isinstance(data, basestring):
return _bg_type(k, 'STRING')
elif isinstance(data, datetime) or isinstance(data, date):
return _bg_type(k, 'TIMESTAMP')
elif data is None:
return _bg_type(k, 'STRING') # if we don't know yet, make it a string
else:
return data
return [_any(k,v) for k,v in event_data.iteritems()]
The code loops through all fields in your data and recursively also creates a schema for any nested objects. The output is compatible with the what BigQuery expects as a TableFieldSchema, see here for a description of the resource. This is not meant to be production code, it just helped me quickly bootstrap the tables needed for our events.