我们如何使用 apache 气流 API 创建 dataproc 集群

Question

我是 Python 和 Airflow 的新手，我在我的 Python 脚本中使用pythonoperator. 第一个和第二个任务从读取数据中检索 zip 文件GCP Bucket，另一个任务是合并两个文件数据。现在我需要再创建一个可以创建的任务Dataproc Cluster。

我见过Airflow API，但我没有得到足够的信息和线索。有什么可以帮助的例子吗？

提前致谢！

Answer 1

有一个名为的运算符DataprocClusterCreateOperator将为您创建 Dataproc 集群。检查https://airflow.apache.org/_api/airflow/contrib/operators/dataproc_operator/index.html#module-airflow.contrib.operators.dataproc_operatorDataprocClusterCreateOperator的文档from airflow.contrib.operators import dataproc_operatorcreate_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator(    task_id='create_dataproc_cluster',    # Give the cluster a unique name by appending the date scheduled.    # See https://airflow.apache.org/code.html#default-variables    cluster_name='hadoop-cluster',    num_workers=2,    zone='europe-west1-b',    master_machine_type='n1-standard-1',    worker_machine_type='n1-standard-1',    dag=dag)

Answer 2

是的，我们需要使用 DataprocClusterCreateOperator。首先我们需要导入dataproc_operator，然后我们还需要使用dag参数传递所有参数，否则会出现错误from airflow.contrib.operators import dataproc_operatorcreate_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator(    task_id='create_dataproc_cluster',    project_id='trim-karma-248213',    cluster_name='airflow-cluster',    num_workers=2,    zone='europe-west1-c',    master_machine_type='n1-standard-1',    worker_machine_type='n1-standard-1',    dag=dag)