Схема таблицы JSON для bigquery.TableSchema для BigQuerySink

Question

У меня есть нетривиальная схема таблицы (включая вложенные и повторяющиеся поля), определенные в формате JSON (с именем, типом, атрибутами режима) и хранящиеся в файле. Он был успешно использован для заполнения таблицы bigquery командой bq load.

Но когда я пытаюсь сделать то же самое с Dataflow Python SDK и BigQuerySink, то schema аргумент должен быть либо разделенный запятыми список 'name':'type' элементов, или bigquery.TableSchema объект.

Есть ли удобный способ получить мою схему JSON до bigquery.TableSchema, или мне нужно преобразовать ее в список name:value?

Answer 1

В настоящее время вы не можете напрямую указать схему JSON. Вы должны указать схему как строку, содержащую список полей, разделенных запятыми, или объект bigquery.TableSchema.

Если схема сложна и содержит вложенные и/или повторяющиеся поля, мы рекомендуем создать объект bigquery.TableSchema.

Приведен пример bigquery.TableSchema объекта с вложенными и повторяющимися полями.

from apitools.clients import bigquery 

table_schema = bigquery.TableSchema() 

# ‘string’ field 
field_schema = bigquery.TableFieldSchema() 
field_schema.name = 'fullName' 
field_schema.type = 'string' 
field_schema.mode = 'required' 
table_schema.fields.append(field_schema) 

# ‘integer’ field 
field_schema = bigquery.TableFieldSchema() 
field_schema.name = 'age' 
field_schema.type = 'integer' 
field_schema.mode = 'nullable' 
table_schema.fields.append(field_schema) 

# nested field 
field_schema = bigquery.TableFieldSchema() 
field_schema.name = 'phoneNumber' 
field_schema.type = 'record' 
field_schema.mode = 'nullable' 

area_code = bigquery.TableFieldSchema() 
area_code.name = 'areaCode' 
area_code.type = 'integer' 
area_code.mode = 'nullable' 
field_schema.fields.append(area_code) 

number = bigquery.TableFieldSchema() 
number.name = 'number' 
number.type = 'integer' 
number.mode = 'nullable' 
field_schema.fields.append(number) 
table_schema.fields.append(field_schema) 

# repeated field 
field_schema = bigquery.TableFieldSchema() 
field_schema.name = 'children' 
field_schema.type = 'string' 
field_schema.mode = 'repeated' 
table_schema.fields.append(field_schema) 

Answer 2

У меня была та же проблема. В моем случае у меня уже был некоторый json, загруженный в bigquery с автоматически созданной схемой.

Так что я был в состоянии получить автогенерируемые schemawith команду:

bq show --format prettyjson my-gcp-project:my-bq-table |jq .schema > my-bq-table.json 

схема может быть преобразована в bigquery.TableSchema с этим фрагментом

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


def _get_field_schema(**kwargs): 
    field_schema = bigquery.TableFieldSchema() 
    field_schema.name = kwargs['name'] 
    field_schema.type = kwargs.get('type', 'STRING') 
    field_schema.mode = kwargs.get('mode', 'NULLABLE') 
    fields = kwargs.get('fields') 
    if fields: 
     for field in fields: 
      field_schema.fields.append(_get_field_schema(**field)) 
    return field_schema 


def _inject_fields(fields, table_schema): 
    for field in fields: 
     table_schema.fields.append(_get_field_schema(**field)) 


def parse_bq_json_schema(schema): 
    table_schema = bigquery.TableSchema() 
    _inject_fields(schema['fields'], table_schema) 
    return table_schema 

Он будет работать со схемой BigQuery JSon спецификации, и если вы ленивы, как и я, вы можете не указывать type и mode, если вы довольны полем, которое по умолчанию равно nullable.