收到错误“客户端具有非平凡的本地和不可腌制的状态。”在 python 中使用 BigQuery 客户端库时？

Question

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

在我的查询中，我正在使用一个表，并且我想在云函数中执行我的查询之前检查该表的存在。为此，我编写了两个 try except 块，第一个将检查表是否存在。如果尚未创建表，则它将使用提供的 create_table 查询创建该表。在它应该执行第二次尝试除了块之后，但在这里我得到了错误。如果尚未创建表，则第一次尝试除块正在创建表，但之后它无法执行第二次尝试除块并出现错误：

""客户端具有非平凡状态，即本地且不可腌制。"，_pickle.PicklingError：明显不支持腌制客户端对象。客户端具有非平凡状态，即本地且不可腌制。”

到目前为止，我已经在 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

我有一个解决我的问题的解决方法，我想要做的是，我为每个尝试编写 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))