使用 Google Composer 运行 Bigquery 查询

Posted 2023-03-25

【中文标题】使用 Google Composer 运行 Bigquery 查询

【英文标题】：Using Google Composer to run Bigquery query

【发布时间】：2020-02-26 06:50:10

【问题描述】：

我是 Google 作曲家和 Apache 气流的新手。

我正在尝试通过创建 DAG 来查询 Bigquery。

import datetime

import airflow
from airflow.operators import bash_operator


from airflow.contrib.operators import bigquery_operator


YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

default_args = 
    'owner': 'me',
    'depends_on_past': False,
    'email': [''],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'start_date': YESTERDAY,


with airflow.DAG(
        'composer_test_dag',
        'catchup=False',
        default_args=default_args,
        schedule_interval=datetime.timedelta(days=1)) as dag:

    bq_recent_questions_query = bigquery_operator.BigQueryOperator(
        task_id='bq_weather_query',
        bql="""
        SELECT owner_display_name, title, view_count
        FROM `bigquery-public-data.***.posts_questions`
        ORDER BY view_count DESC
        LIMIT 100
        """,
        use_legacy_sql=False)     

这是正确的方法吗？ 如何从中获取查询结果？

【问题讨论】：

你绝对需要使用 Airflow/Composer 吗？这很难、复杂且昂贵。另一种选择是使用 Cloud Scheduler + Cloud Build。更容易......而且更便宜。

【参考方案1】：

BigQueryOperator 通常用于在 BigQuery 中执行查询，然后将结果加载到另一个 BigQuery 表（转换操作）。我假设您正在尝试从 BigQuery 公共表中选择 3 列并加载到另一个表。所以在 BigQueryOperator 中提供destination_dataset_table。

请注意以下几点：

***.posts_questions 表非常大，即使使用 LIMIT 仍然会扫描整个表。所以要注意成本。 使用sql 参数代替bql，因为它不再使用。

【讨论】：

谢谢！ BigQueryOperator 中的destination_dataset_table >> 通过提供此字段，可以使用查询结果创建一个新视图/表？

是的，它创建了一个表。此处还提供了使用 BigQueryOperator 创建 view 而不是表的实现：github.com/apache/airflow/blob/… 因此，如果您需要视图而不是表，您可以参考该代码并将实现复制到您的代码库中。

何时创建新表的行为可以通过create_disposition参数进行配置。默认情况下，此参数设置为CREATE_IF_NEEDED，它将按照您的描述进行。另请查看相关参数write_disposition。默认设置为WRITE_EMPTY，只有在表为空或刚刚创建时才会写入。