使用Airflow BigQuery Hook无法获取表计数，打印无结果求助

你的问题在于hook.insert_job()返回的是BigQuery的Job对象，而非查询的实际结果，所以直接打印只会输出对象信息，看不到计数数值。下面提供两种可行的解决方法：

方法一：使用get_pandas_df直接获取结果（推荐） #

BigQueryHook提供了更便捷的get_pandas_df方法，可以直接执行查询并返回Pandas DataFrame，提取计数非常方便：

def count():
    from airflow.providers.google.cloud.hooks.bigquery import BigQueryHook

    hook = BigQueryHook(gcp_conn_id='connect_id', location='location_name')
    # 执行查询并返回DataFrame
    df = hook.get_pandas_df(sql="select count(*) as total_count from project_id.dataset.table", use_legacy_sql=False)
    # 提取计数数值
    target_count = df['total_count'].iloc[0]
    print(f"总记录数：{target_count}")
    # 若需跨任务共享结果，可存入Airflow变量
    from airflow.models import Variable
    Variable.set("bq_table_total_count", target_count)

# DAG定义部分保持不变
with models.Dag(
        'count_test',
        start_date = datetime(20, 9, 1),
        schedule_interval = None, 
        catchup = False,
        ) as dag:
    
    from airflow.operators.python_operator import PythonOperator
    count_bqHook = PythonOperator(
                                   task_id='task1',
                                   python_callable=count
                                 )
    
chain( count_bqHook )

方法二：通过insert_job获取Job结果（适合复杂场景） #

如果必须使用insert_job执行查询，需要从Job对象中获取查询结果的存储位置，再读取数据：

def count():
    from airflow.providers.google.cloud.hooks.bigquery import BigQueryHook

    config={
        "job_type":"Query",
        "query":{
            "query":"select count(*) from project_id.dataset.table",
            "useLegacySql":False,
            "allow_large_results":True,
            "location":"location_name"
        }}
    
    hook = BigQueryHook(gcp_conn_id='connect_id')
    # 执行查询并等待完成
    job = hook.insert_job(configuration=config, nowait=False)
    # 获取BigQuery客户端并读取结果
    client = hook.get_client()
    query_result = client.get_job(job.job_id).result()
    # 提取计数
    for row in query_result:
        target_count = row[0]
        print(f"总记录数：{target_count}")

# DAG定义部分不变
with models.Dag(
        'count_test',
        start_date = datetime(20, 9, 1),
        schedule_interval = None, 
        catchup = False,
        ) as dag:
    
    from airflow.operators.python_operator import PythonOperator
    count_bqHook = PythonOperator(
                                   task_id='task1',
                                   python_callable=count
                                 )
    
chain( count_bqHook )

关键说明 #

• get_pandas_df是最简洁的方式，适合简单查询场景，直接返回结构化数据。
• 若需处理大结果集或复杂查询配置，用insert_job配合客户端读取结果更灵活，但步骤更多。
• 跨任务共享计数时，可将结果存入Airflow的Variable中。

内容的提问来源于stack exchange，提问作者D M