python - 在 python 中使用 BigQuery 客户端库时出现错误 "Clients have non-trivial state that is local and unpickleable."？

Question

我正在开发一个 GCP 云函数，它将在 BigQuery 表上执行查询，将数据加载到某个临时表中，然后该数据可能会被该临时表中的其他一些云函数使用。

在我的查询中，我使用一个表，我想在云函数中执行查询之前检查该表是否存在。为此，我编写了两个 try except block ，第一个 block 将检查表是否存在。如果表尚未创建，那么它将使用提供的 create_table 查询创建该表。之后它应该执行第二个 try except block ，但在这里我收到错误。第一个 try except block 正在创建表(如果尚未创建)，但之后执行第二个 try except block 失败并出现错误:

""客户端具有本地且不可pickle的非平凡状态。", _pickle.PicklingError:明确不支持pickling客户端对象。客户端具有本地且不可pickle的非平凡状态。”

到目前为止，我已经在 python 中尝试了以下代码:

def main(request):
    my_client = bigquery.Client()
    create_table = 'CREATE OR REPLACE TABLE {}.{} (customerName STRING, IDNumber STRING)'.format(dest_dataset, dest_table)
    job_configs = bigquery.QueryJobConfig()
    destination_dataset = my_client.dataset(dest_dataset, dest_project)
    destination_table = destination_dataset.table(dest_table)
    job_configs.destination = destination_table

    # Check the final destination table is already exists, if not then create it based on create_table query.

    try:
        table = my_client.get_table(destination_table)
        if table:
            print('Table {}\'s existence sucessfully proved!'.format(destination_table))

    except NotFound as error:
        temporary_table = my_client.query(create_table, location='US')
        temporary_table.result()
        print('Table {} is created!'.format(destination_table))


    # After checking destination_table existance, run actual query (destination_table being used in this query) and load data into temporary table
    try:

        client = bigquery.Client()
        job_config = bigquery.QueryJobConfig()
        dest_dataset = client.dataset(temporary_dataset, temporary_project)
        dest_table = dest_dataset.table(temporary_table)
        job_config.destination = dest_table
        job_config.create_disposition = 'CREATE_IF_NEEDED'
        job_config.write_disposition = 'WRITE_TRUNCATE'
        query_job = client.query(query, location='US', job_config=job_config)
        query_job.result()
        table = client.get_table(dest_table)
        expiration = (datetime.now() + timedelta(minutes=expiration_time))
        table.expires = expiration
        table = client.update_table(table, ["expires"])
        logger.info("Query result loaded into temporary table: {}".format(temporary_table))

    except RuntimeError:
        logger.error("Exception occurred {}".format(RuntimeError))

有什么方法可以解决此错误或有任何不同的方法来检查表是否存在

Answer 1

我找到了解决我的问题的解决方法，我想做的是，我为每个 try except block 编写 2 个单独的函数，如下所示:

def check_dest_table_existence(request):
    my_client = bigquery.Client()
    create_table = 'CREATE OR REPLACE TABLE {}.{} (customerName STRING, IDNumber STRING)'.format(dest_dataset, dest_table)
    job_configs = bigquery.QueryJobConfig()
    destination_dataset = my_client.dataset(dest_dataset, dest_project)
    destination_table = destination_dataset.table(dest_table)
    job_configs.destination = destination_table

    # Check the final destination table is already exists, if not then create it based on create_table query.

    try:
        table = my_client.get_table(destination_table)
        if table:
            print('Table {}\'s existence sucessfully proved!'.format(destination_table))

    except NotFound as error:
        temporary_table = my_client.query(create_table, location='US')
        temporary_table.result()
        print('Table {} is created!'.format(destination_table))



def main(request):

    # Check destination_table existance, run actual query (destination_table being used in this query) and load data into temporary table
    try:

        # Calling another function to check existence of destination table.
        check_dest_table_existence(request)

        client = bigquery.Client()
        job_config = bigquery.QueryJobConfig()
        dest_dataset = client.dataset(temporary_dataset, temporary_project)
        dest_table = dest_dataset.table(temporary_table)
        job_config.destination = dest_table
        job_config.create_disposition = 'CREATE_IF_NEEDED'
        job_config.write_disposition = 'WRITE_TRUNCATE'
        query_job = client.query(query, location='US', job_config=job_config)
        query_job.result()
        table = client.get_table(dest_table)
        expiration = (datetime.now() + timedelta(minutes=expiration_time))
        table.expires = expiration
        table = client.update_table(table, ["expires"])
        logger.info("Query result loaded into temporary table: {}".format(temporary_table))

    except RuntimeError:
        logger.error("Exception occurred {}".format(RuntimeError))