a.py

from airflow import models

from airflow.contrib.operators import dataproc_operator

from airflow.operators import bash_operator

from airflow.operators import dummy_operator

from airflow.utils import dates

from airflow.utils.trigger_rule import TriggerRule

from o2a.o2a_libs.el_basic_functions import *

from o2a.o2a_libs.el_wf_functions import *

import datetime


PARAMS = {
    "dataproc_cluster": "oozie-51",
    "examplesRoot": "examples",
    "gcp_conn_id": "google_cloud_default",
    "gcp_region": "europe-west3",
    "gcp_uri_prefix": "gs:///dags",
    "nameNode": "hdfs://localhost:8020",
    "oozie.use.system.libpath": "true",
    "oozie.wf.application.path": "/user/kamilbregula/examples/apps/git",
    "queueName": "default",
    "resourceManager": "localhost:8032",
    "user.name": "kamilbregula",
    "userName": "kamilbregula",
}

with models.DAG(
    "test_git_dag",
    schedule_interval=None,  # Change to suit your needs
    start_date=dates.days_ago(0),  # Change to suit your needs
) as dag:

    git_node_prepare = bash_operator.BashOperator(
        task_id="git_node_prepare",
        trigger_rule="dummy",
        bash_command='$DAGS_FOLDER/../data/prepare.sh -c oozie-51 -r europe-west3 -d "/user/kamilbregula/examples/apps/git/repo"',
    )

    git_node = bash_operator.BashOperator(
        task_id="git_node",
        trigger_rule="dummy",
        bash_command="$DAGS_FOLDER/../data/git.sh --cluster {dataproc_cluster} --region {gcp_region} --git-uri https://github.com/GoogleCloudPlatform/cloud-composer.git --destination-path /user/kamilbregula/examples/apps/git/repo --branch master".format(
            dataproc_cluster=PARAMS["dataproc_cluster"], gcp_region=PARAMS["gcp_region"]
        ),
    )

    git_node_prepare.set_downstream(git_node)