JSON表架构到bigquery.TableSchema对于BigQuerySin

Question

我有一个非常重要的表模式（包括嵌套和重复的字段），以JSON格式定义（带有name、type、mode属性）并存储在一个文件中。它已成功地用于用bqload命令填充bigquery表。在

但是当我尝试使用Dataflow Python SDK和BigQuerySink执行相同的操作时，schema参数需要是一个以逗号分隔的'name':'type'元素列表，或者是一个bigquery.TableSchema对象。在

有没有什么方便的方法可以将我的JSON模式转换为bigquery.TableSchema，还是必须将其转换为name:value列表？在

Answer 1

andreapierleoni发布的上述代码片段适用于google-cloud-bigquerypython客户机的旧版本，例如，google-cloud-bigquery的0.25.0版本，它碰巧通过pip install apache-beam[gcp]安装。在

但是，bigquerypython客户机API在google-cloud-bigquery的较新版本中发生了巨大的变化，例如在我当前使用的1.8.0版本中，bigquery.TableFieldSchema()和{}都不起作用。在

如果您使用的是google-cloud-bigquery包的最新版本，下面介绍如何从JSON文件获取所需的SchemaField列表（例如，创建表所必需的）。这是AndreaPierleoni发布的代码的改编版（谢谢！）在

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

现在，假设您有一个表的schema already defined in JSON。假设您有this particular "schema.json" file，那么使用上面的helper方法，您可以获得Python客户机所需的SchemaField表示，如下所示：

^{pr2}$

现在，对于create a table having the above schema using the Python SDK，您将执行以下操作：

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

您可以选择按如下方式设置基于时间的分区（如果需要）：

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

最后创建了表格：

table = bqclient.create_table(table)

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