从 python 字典自动生成 BigQuery 架构

Posted 2023-03-25

【中文标题】从 python 字典自动生成 BigQuery 架构

【英文标题】：Autogenerating BigQuery schema from python dictionary

【发布时间】：2019-05-10 14:43:54

【问题描述】：

如何从 Python 字典自动生成 BigQuery 表架构？

例如

dict = 'data': 'some_data', 'me': 8
schema = BigQuery.generateSchema(dict)

#schema is now:
# 'fields': [
#    'name': 'data', 'type': 'STRING', 'mode': 'NULLABLE',
#    'name': 'me', 'type': 'INT', 'mode': 'NULLABLE'
# ]

这样的东西存在吗？

【问题讨论】：

作为更新。我很确定这不可能没有一些第三方插件。看来我们得自己写代码了！

【参考方案1】：

目前没有通过 BigQuery Python 库执行此操作的方法。

这里有一个递归函数来实现它。

import datetime
from google.cloud.bigquery.schema import SchemaField

# [START] map_dict_to_bq_schema

# FieldType Map Dictionary
field_type = 
        str: 'STRING',
        bytes: 'BYTES',
        int: 'INTEGER',
        float: 'FLOAT',
        bool: 'BOOLEAN',
        datetime.datetime: 'DATETIME',
        datetime.date: 'DATE',
        datetime.time: 'TIME',
        dict: 'RECORD',



# Function to take a dictionary
# and return a bigquery schema
def map_dict_to_bq_schema(source_dict):

    # SchemaField list
    schema = []

    # Iterate the existing dictionary
    for key, value in source_dict.items():

        try:
            schemaField = SchemaField(key, field_type[type(value)]) # NULLABLE BY DEFAULT
        except KeyError:

            # We are expecting a REPEATED field
            if value and len(value) > 0:
                schemaField = SchemaField(key, field_type[type(value[0])], mode='REPEATED') # REPEATED

        # Add the field to the list of fields
        schema.append(schemaField)

        # If it is a STRUCT / RECORD field we start the recursion
        if schemaField.field_type == 'RECORD':

            schemaField._fields = map_dict_to_bq_schema(value)

    # Return the dictionary values
    return schema

# [END] map_dict_to_bq_schema

例子：

>>> map_dict_to_bq_schema('data': 'some_data', 'me': 8)
# Output
>>> [SchemaField('data', 'STRING', 'NULLABLE', None, ()), SchemaField('me', 'INTEGER', 'NULLABLE', None, ())]


>>> map_dict_to_bq_schema('data': 'data2': 'some_data', 'me2': 8, 'me': 8, 'h':[5,6,7])
# Output
>>> [SchemaField('h', 'INTEGER', 'REPEATED', None, ()), SchemaField('me', 'INTEGER', 'NULLABLE', None, ()), SchemaField('data', 'RECORD', 'NULLABLE', None, [SchemaField('data2', 'STRING', 'NULLABLE', None, ()), SchemaField('me2', 'INTEGER', 'NULLABLE', None, ())])]

我在这个问题中使用了@luckylwk 的代码作为参考：How to map a Python Dict to a Big Query Schema，专门用于nested and repeated 列。

另外，检查 BQ python 库中的 SchemaField 类。从那里你可以得到你想使用模式与 python 客户端、CLI 或匹配你的用例的格式。

【讨论】：

+9000 在 Python 中比 bigquery_schema_g​​enerator 效果更好，谢谢