Json 架构文件不会在 BigQuery Python API 中执行

Question

我遇到了 Bigquery Python API 的问题。这是我执行脚本时的堆栈跟踪：

Traceback (most recent call last):
  File "createTable.py", line 17, in <module>
    open_schema()
  File "createTable.py", line 12, in open_schema
    table = bigquery.Table(table_id, schema=schema)
    ...
    "Schema items must either be fields or compatible "
ValueError: Schema items must either be fields or compatible mapping representations.

脚本很简单，打开一个schema文件并创建表：

   from google.cloud import bigquery
   # Construct a BigQuery client object.
   client = bigquery.Client()
   table_id = "project-py-290522:bq_dts.bq-test"
       
   def open_schema():
       with open("hcl-schema.json","r", encoding = "utf-8") as fName:
           schema = fName.readlines()
       
           table = bigquery.Table(table_id, schema=schema)
           print(repr(table))
           client.create_table(table)  # Make an API request.
       
   if __name__ == "__main__":
       open_schema()
       
   print("Created table {}.{}.{}".format(table.project, table.dataset_id, table.table_id)) 

当我在控制台和 CLI 中执行架构时，表会按原样完美创建。控制台和 CLI 如何执行创建表但在 API 中阻塞。我已经搜索和搜索并没有找到答案。有人可以帮忙吗？

这是存储在 hcl-schema.json 文件中的架构。为简洁起见，我缩短了属性列表，但在其他方面保持不变：

    [  
      {
        "name":"user_id",
        "type":"STRING",
        "mode":"NULLABLE"
      },
      {
        "name":"msg_version",
        "type":"STRING",
        "mode":"REQUIRED"           
      },
      {
        "name":"APIStreamData",
        "type":"RECORD",
        "mode":"REQUIRED",
        "fields":
        [
          {
            "name":"msg_version",
            "type":"STRING",
            "mode":"REQUIRED"
          },
          {
            "name":"streams",
            "type":"RECORD",
            "mode":"REPEATED",
            "fields":
            [
                {
                  "name":"length",
                  "type":"STRING",
                  "mode":"REQUIRED"
                },                      
                {
                  "name":"cached",
                  "type":"STRING",
                  "mode":"NULLABLE"
                },  
              {
                "name":"track",
                "type":"RECORD",
                "mode":"REQUIRED",
                "fields":
                [
                  {
                    "name":"msg_version",
                    "type":"STRING",
                    "mode":"REQUIRED"
                  },
                  {
                    "name":"track_id",
                    "type":"STRING",
                    "mode":"REQUIRED"
                  }
                ]
              }    
    
            ]
          }
        ]
      }
    ]

谢谢

茫然和困惑

Answer 1

您可以使用 json 文件的 dict 表示，而不是原始问题中的字符串：

with open("schema.json") as json_file:
    schema_dict = json.load(json_file)
    table = bigquery.Table(table_id, schema=schema_dict)
    table = client.create_table(table)

Answer 2

我所指的以下文档显示了要在 Python 中指定的架构，如下所示 https://cloud.google.com/bigquery/docs/tables#creating_an_empty_table_with_a_schema_definition

schema = [bigquery.SchemaField("full_name", "STRING", mode="REQUIRED"),bigquery.SchemaField("age", "INTEGER", mode="REQUIRED")]

我尝试使用以下 JSON 和代码，效果很好。我认为您提供的 json 在 RECORD 中有 RECORD 所以我们需要相应地处理

from google.cloud import bigquery
import json
# Construct a BigQuery client object.
client = bigquery.Client()
table_id = "my-project.mock_dataset.bq-test"

def open_schema():
            
    bigquerySchema = []
    bigqueryfieldSchema = []
    with open('test.json') as f:
        bigqueryColumns = json.load(f)
        print(bigqueryColumns)
        for col in bigqueryColumns:
            if col['type'] != 'RECORD':
                print(col['name'])
                bigquerySchema.append(bigquery.SchemaField(col['name'], col['type'],mode=col['mode']))
            else:
                for colfield in col['fields']:
                    bigqueryfieldSchema.append(bigquery.SchemaField(colfield['name'], colfield['type'],colfield['mode']))
                print(bigqueryfieldSchema)
                print(col['fields'])
                bigquerySchema.append(bigquery.SchemaField(col['name'], col['type'],col['mode'],'',bigqueryfieldSchema))
        print(bigquerySchema)

    table = bigquery.Table(table_id, schema=bigquerySchema)
    print(repr(table))
    client.create_table(table)  # Make an API request.
    print("Created table {}.{}.{}".format(table.project, table.dataset_id, table.table_id)) 

if __name__ == "__main__":
    open_schema()

 [  
      {
        "name":"user_id",
        "type":"STRING",
        "mode":"NULLABLE"
      },
      {
        "name":"msg_version",
        "type":"STRING",
        "mode":"REQUIRED"           
      },
      {
        "name":"APIStreamData",
        "type":"RECORD",
        "mode":"REQUIRED",
        "fields":
        [
          {
            "name":"msg_version",
            "type":"STRING",
            "mode":"REQUIRED"
          },
          {
             "name":"track_id",
             "type":"STRING",
             "mode":"REQUIRED"
          }
       ]
      }
 ]