我们如何使用 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_operator

create_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_operator

create_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)