spark_ondemand.py

# ./airflow variables set gcp_project bigdataupv2022
# ./airflow variables set gcp_region europe-west1
# ./airflow variables set gcp_zone europe-west1-b
# ./airflow variables set gcp_bucket bigdataupv_data

import datetime
import os

from airflow import models
from airflow.contrib.operators import dataproc_operator
from airflow.utils import trigger_rule

yesterday = datetime.datetime.combine(
    datetime.datetime.today() - datetime.timedelta(1),
    datetime.datetime.min.time())

default_dag_args = {
    'start_date': yesterday,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'project_id': models.Variable.get('gcp_project')
}

with models.DAG(
        'spark_ondemand',
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:

    create_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator(
        task_id='create_dataproc_cluster',
        cluster_name='spark-cluster-{{ ds_nodash }}',
        num_workers=2,
        zone=models.Variable.get('gcp_zone'),
        region=models.Variable.get('gcp_region'),
        master_machine_type='n1-standard-1',
        worker_machine_type='n1-standard-1')

    run_dataproc_pyspark = dataproc_operator.DataProcPySparkOperator(
        task_id='run_spark',
        cluster_name='spark-cluster-{{ ds_nodash }}',
        region=models.Variable.get('gcp_region'),
        main='gs://bigdataupv_code/compras_top_ten_countries.py',
        files=['gs://bigdataupv_code/helpers.py'])

    delete_dataproc_cluster = dataproc_operator.DataprocClusterDeleteOperator(
        task_id='delete_dataproc_cluster',
        cluster_name='spark-cluster-{{ ds_nodash }}',
        trigger_rule=trigger_rule.TriggerRule.ALL_DONE)

    create_dataproc_cluster >> run_dataproc_pyspark >> delete_dataproc_cluster