python - 使用 python 将 BigQuery 查询结果写入表中

Question

我想将存储在 txt 文件中的 BigQuery 查询结果写入 BigQuery 表中。我将查询文本作为变量传递到下面的函数中，但出现以下错误:

error_info=method + ' ' + url) google.cloud.exceptions.BadRequest: 400 缺少必需参数 (POST https://www.googleapis.com/bigquery/v2/projects/myproject/jobs )

我错过了什么？

功能:

from google.cloud import bigquery
import uuid

def async_query(query, dataset_id, dest_table, project_Id):


    # configuration json
    query_data = {
        "configuration": {
            "query": {
                "query": query,
                "defaultDataset": dataset_id,
                "allowLargeResults": True,
                "destinationTable": {
                    "projectId": project_Id,
                    "datasetId": dataset_id,
                    "tableId": dest_table
                    },
                "createDisposition": 'CREATE_IF_NEEDED',
                "writeDisposition": 'WRITE_TRUNCATE'
          }
        }
    }

    client = bigquery.Client()
    query_job = client.run_async_query(str(uuid.uuid4()), query_data)
    query_job.use_legacy_sql = False
    query_job.begin()
    wait_for_job(query_job)

    # Drain the query results by requesting a page at a time.
    query_results = query_job.results()
    page_token = None

    while True:
        rows, total_rows, page_token = query_results.fetch_data(
            max_results=10,
            page_token=page_token)

        for row in rows:
            print(row)

        if not page_token:
            break
def wait_for_job(job):
    while True:
        job.reload()  # Refreshes the state via a GET request.
        if job.state == 'DONE':
            if job.error_result:
                raise RuntimeError(job.errors)
            return
        time.sleep(1)

Answer 1

1. 您可以按如下方式修复配置中的 defaultDataset

   # configuration json
   query_data = {
       "configuration": {
           "query": {
               "query": query,
               "defaultDataset": {
                   "projectId": project_Id,
                   "datasetId": dataset_id
                   },
               "allowLargeResults": True,
               "destinationTable": {
                   "projectId": project_Id,
                   "datasetId": dataset_id,
                   "tableId": dest_table
                   },
               "createDisposition": 'CREATE_IF_NEEDED',
               "writeDisposition": 'WRITE_TRUNCATE'
         }
       }
   }
   

注意:“projectId”:project_Id 在defaultDataset 中是可选的

整个 defaultDataset 也是可选的，根据您的情况，您可以省略它，如

# configuration json
query_data = {
    "configuration": {
        "query": {
            "query": query,
            "allowLargeResults": True,
            "destinationTable": {
                "projectId": project_Id,
                "datasetId": dataset_id,
                "tableId": dest_table
                },
            "createDisposition": 'CREATE_IF_NEEDED',
            "writeDisposition": 'WRITE_TRUNCATE'
      }
    }
}

查看更多configuration.query.defaultDataset