如何自动化 BigQuery SQL 管道

Question

我使用 BigQuery SQL 创建了一个数据管道。 首先从 Cloud Storage 导入 CSV 文件，然后进行不同的分析，包括使用 BigQueryML 进行预测建模 使用地理函数进行地理计算，以及 使用分析函数计算 KPI。

我能够成功地手动运行不同的查询，现在我想自动化数据管道。

我的第一选择是 DataFlow SQL，但事实证明 Dataflow SQL 查询语法不支持地理函数。

DataFlow python 不是一个选项，因为完整的分析是在 SQL 中完成的，我想保持这种方式。

我的问题是可用于自动化数据管道的其他 GCP 选项是什么。

Answer 1

正如我在评论中提到的，如果您需要编排查询，可以使用 Cloud Composer，一个完全托管的 Airflow 集群。

我创建了下面的代码，或多或少地向您展示如何使用此工具编排查询。请注意，这是一个基本代码，可以在编码标准方面进行改进。 该代码基本上编排了 3 个查询：

1. 第一个从公共表读取并写入项目中的另一个表
2. 第二个读取在第一个查询中创建的表，并根据日期列选择 10000 条最新的行。之后，它会将结果保存到项目中的表中。

3. 第三个读取步骤 2 中创建的表并计算一些聚合。之后，它将结果保存到项目中的另一个表中。

   import datetime
   from airflow import models
   from airflow.contrib.operators import bigquery_operator
   
   """The condiguration presented below will run your DAG every five minutes as specified in the 
   schedule_interval property starting from the datetime specified in the start_date property"""
   
   default_dag_args = {
       'start_date': datetime.datetime(2020, 4, 22, 15, 40), 
       'email_on_failure': False,
       'email_on_retry': False,
       'retries': 1,
       'retry_delay': datetime.timedelta(minutes=1),
       'project_id': "<your_project_id>",
   }
   
   with models.DAG(
           'composer_airflow_bigquery_orchestration',
           schedule_interval = "*/5 * * * *",
           default_args=default_dag_args) as dag:
   
       run_first_query = bigquery_operator.BigQueryOperator(
           sql = "SELECT * FROM `bigquery-public-data.catalonian_mobile_coverage.mobile_data_2015_2017`",
           destination_dataset_table = "<your_project>.<your_dataset>.orchestration_1",
           task_id = 'xxxxxxxx',
           write_disposition = "WRITE_TRUNCATE",
           #create_disposition = "",
           allow_large_results = True,
           use_legacy_sql = False
       )
   
       run_second_query = bigquery_operator.BigQueryOperator(
           sql = "SELECT * FROM `<your_project>.orchestration_1` ORDER BY date LIMIT 10000 ",
           destination_dataset_table = "<your_project>.<your_dataset>.orchestration_2",
           task_id = 'yyyyyyyy',
           write_disposition = "WRITE_TRUNCATE",
           #create_disposition = "",
           allow_large_results = True,
           use_legacy_sql = False
       )
   
       run_third_query = bigquery_operator.BigQueryOperator(
           sql = "SELECT round(lat) r_lat, round(long) r_long, count(1) total FROM`<your_project>.orchestration_2` GROUP BY r_lat,r_long",
           destination_dataset_table = "<your_project>.<your_dataset>.orchestration_3",
           task_id = 'zzzzzzzz',
           write_disposition = "WRITE_TRUNCATE",
           #create_disposition = "",
           allow_large_results = True,
           use_legacy_sql = False
       )
   
   
      # Define DAG dependencies.
       run_first_query >> run_second_query >> run_third_query
   

一步一步来：

• 首先，它导入了一些 Airflow 库，如模型和 bigquery_operator

  from airflow import models
  from airflow.contrib.operators import bigquery_operator
  

• 然后它定义了一个名为 default_dag_args 的字典，将在您创建 DAG 时进一步使用。

  default_dag_args = {
      'start_date': datetime.datetime(2020, 4, 22, 15, 40), 
      'email_on_failure': False,
      'email_on_retry': False,
      'retries': 1,
      'retry_delay': datetime.timedelta(minutes=1),
      'project_id': "<your_project_id>",
  }
  

• 创建 DAG 时，将 default_dag_args dict 作为默认参数传递，并添加将定义 DAG 何时运行的 schedule interval 参数。您可以将此参数与一些预设表达式或使用 CRON 表达式一起使用，如您所见here

  with models.DAG(
          'composer_airflow_bigquery_orchestration',
          schedule_interval = "*/5 * * * *",
          default_args=default_dag_args) as dag:
  

• 之后，您可以创建运营商的实例。在这种情况下，我们只使用BigQueryOperator

      run_first_query = bigquery_operator.BigQueryOperator(
          sql = "SELECT * FROM `bigquery-public-data.catalonian_mobile_coverage.mobile_data_2015_2017`",
          destination_dataset_table = "<your_project>.<your_dataset>.orchestration_1",
          task_id = 'xxxxxxxx',
          write_disposition = "WRITE_TRUNCATE",
          #create_disposition = "",
          allow_large_results = True,
          use_legacy_sql = False
      )
  
      run_second_query = bigquery_operator.BigQueryOperator(
          sql = "SELECT * FROM `<your_project>.orchestration_1` ORDER BY date LIMIT 10000 ",
          destination_dataset_table = "<your_project>.<your_dataset>.orchestration_2",
          task_id = 'yyyyyyyy',
          write_disposition = "WRITE_TRUNCATE",
          #create_disposition = "",
          allow_large_results = True,
          use_legacy_sql = False
      )
  
      run_third_query = bigquery_operator.BigQueryOperator(
          sql = "SELECT round(lat) r_lat, round(long) r_long, count(1) total FROM`<your_project>.orchestration_2` GROUP BY r_lat,r_long",
          destination_dataset_table = "<your_project>.<your_dataset>.orchestration_3",
          task_id = 'zzzzzzzz',
          write_disposition = "WRITE_TRUNCATE",
          #create_disposition = "",
          allow_large_results = True,
          use_legacy_sql = False
      )
  

• 作为最后一步，我们可以定义 DAG 的依赖关系。这段代码的意思是run_second_query的操作依赖于run_first_query的结论，这样就可以了。

      run_first_query >> run_second_query >> run_third_query
  

最后，我想添加这个article，讨论如何在使用 CRON 表达式时正确设置 start_date 和 schedule_interval。

Answer 2

BigQuery 有一个内置的调度机制，目前处于测试阶段。

要自动化 BQ 本机 SQL 管道，您可以使用此实用程序。 使用 CLI：

$ bq query \
--use_legacy_sql=false \
--destination_table=mydataset.mytable \
--display_name='My Scheduled Query' \
--replace=true \
'SELECT
1
FROM
mydataset.test'