fix: b/219021576 (GoogleCloudPlatform#7473)

## Description

Fixes b/219021576, corresponding cl/429403697

Also adds jinja templated variables as best practice as a part of addressing GoogleCloudPlatform#6555 

[Airflow 1 successful DAG Run](https://screenshot.googleplex.com/AwmdtnCmh94dKXE)
[Airflow 2 successful DAG Run](https://screenshot.googleplex.com/7sNuCXtQ6NvXbyp)

Note: It's a good idea to open an issue first for discussion.

## Checklist
- [x] I have followed [Sample Guidelines from AUTHORING_GUIDE.MD](https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/AUTHORING_GUIDE.md)
- [ ] README is updated to include [all relevant information](https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/AUTHORING_GUIDE.md#readme-file)
- [ ] **Tests** pass:   `nox -s py-3.6` (see [Test Environment Setup](https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/AUTHORING_GUIDE.md#test-environment-setup))
- [ ] **Lint** pass:   `nox -s lint` (see [Test Environment Setup](https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/AUTHORING_GUIDE.md#test-environment-setup))
- [ ] These samples need a new **API enabled** in testing projects to pass (let us know which ones)
- [ ] These samples need a new/updated **env vars** in testing projects set to pass (let us know which ones)
- [x] Please **merge** this PR for me once it is approved.
- [ ] This sample adds a new sample directory, and I updated the [CODEOWNERS file](https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/.github/CODEOWNERS) with the codeowners for this sample

Author: leahecole

composer/airflow_1_samples/constraints.txt

@@ -1,3 +1,4 @@
 pandas-gbq==0.14.1 # must be under 0.15.0 until https://github.com/apache/airflow/issues/15113 is addressed
 SQLAlchemy==1.3.23 # must be under 1.4 until at least Airflow 2.0 (check airflow setup.py for restrictions)
-WTForms==2.3.3 # Must stay under 3.0 because of https://github.com/dpgaspar/Flask-AppBuilder/issues/1732
+WTForms==2.3.3 # Must stay under 3.0 because of https://github.com/dpgaspar/Flask-AppBuilder/issues/1732
+MarkupSafe==1.1.1 # as found in Airflow 1.10.15 constraints

composer/airflow_1_samples/hadoop_tutorial.py

@@ -19,7 +19,7 @@
 This DAG relies on three Airflow variables
 https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
 * gcp_project - Google Cloud Project to use for the Cloud Dataproc cluster.
-* gce_zone - Google Compute Engine zone where Cloud Dataproc cluster should be
+* gce_region - Google Compute Engine region where Cloud Dataproc cluster should be
   created.
 * gcs_bucket - Google Cloud Storage bucket to use for result of Hadoop job.
   See https://cloud.google.com/storage/docs/creating-buckets for creating a
@@ -38,7 +38,7 @@
 # see https://airflow.apache.org/docs/apache-airflow/stable/timezone.html
 # for best practices
 output_file = os.path.join(
-    models.Variable.get('gcs_bucket'), 'wordcount',
+    '{{ var.value.gcs_bucket }}', 'wordcount',
     datetime.datetime.now().strftime('%Y%m%d-%H%M%S')) + os.sep
 # Path to Hadoop wordcount example available on every Dataproc cluster.
 WORDCOUNT_JAR = (
@@ -63,7 +63,7 @@
     # If a task fails, retry it once after waiting at least 5 minutes
     'retries': 1,
     'retry_delay': datetime.timedelta(minutes=5),
-    'project_id': models.Variable.get('gcp_project')
+    'project_id': '{{ var.value.gcp_project }}'
 }
 
 # [START composer_hadoop_schedule_airflow_1]
@@ -81,7 +81,7 @@
         # See https://airflow.apache.org/docs/apache-airflow/stable/macros-ref.html
         cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
         num_workers=2,
-        zone=models.Variable.get('gce_zone'),
+        region='{{ var.value.gce_region }}',
         master_machine_type='n1-standard-2',
         worker_machine_type='n1-standard-2')
 
@@ -90,13 +90,16 @@
     run_dataproc_hadoop = dataproc_operator.DataProcHadoopOperator(
         task_id='run_dataproc_hadoop',
         main_jar=WORDCOUNT_JAR,
+        region='{{ var.value.gce_region }}',
         cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
         arguments=wordcount_args)
 
     # Delete Cloud Dataproc cluster.
     delete_dataproc_cluster = dataproc_operator.DataprocClusterDeleteOperator(
         task_id='delete_dataproc_cluster',
         cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
+        region='{{ var.value.gce_region }}',
+
         # Setting trigger_rule to ALL_DONE causes the cluster to be deleted
         # even if the Dataproc job fails.
         trigger_rule=trigger_rule.TriggerRule.ALL_DONE)

composer/airflow_1_samples/requirements.txt

@@ -8,4 +8,3 @@ scipy==1.4.1; python_version > '3.0'
 scipy==1.2.3; python_version < '3.0'
 numpy==1.19.5; python_version > '3.0'
 numpy==1.16.6; python_version < '3.0'
-markupsafe==2.0.1

composer/workflows/hadoop_tutorial.py

@@ -19,7 +19,7 @@
 This DAG relies on three Airflow variables
 https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
 * gcp_project - Google Cloud Project to use for the Cloud Dataproc cluster.
-* gce_zone - Google Compute Engine zone where Cloud Dataproc cluster should be
+* gce_region - Google Compute Engine region where Cloud Dataproc cluster should be
   created.
 * gcs_bucket - Google Cloud Storage bucket to use for result of Hadoop job.
   See https://cloud.google.com/storage/docs/creating-buckets for creating a
@@ -38,7 +38,7 @@
 # see https://airflow.apache.org/docs/apache-airflow/stable/timezone.html
 # for best practices
 output_file = os.path.join(
-    models.Variable.get('gcs_bucket'), 'wordcount',
+    '{{ var.value.gcs_bucket }}', 'wordcount',
     datetime.datetime.now().strftime('%Y%m%d-%H%M%S')) + os.sep
 # Path to Hadoop wordcount example available on every Dataproc cluster.
 WORDCOUNT_JAR = (
@@ -49,7 +49,7 @@
 wordcount_args = ['wordcount', input_file, output_file]
 
 HADOOP_JOB = {
-    "reference": {"project_id": models.Variable.get('gcp_project')},
+    "reference": {"project_id": '{{ var.value.gcp_project }}'},
     "placement": {"cluster_name": 'composer-hadoop-tutorial-cluster-{{ ds_nodash }}'},
     "hadoop_job": {
         "main_jar_file_uri": WORDCOUNT_JAR,
@@ -83,8 +83,8 @@
     # If a task fails, retry it once after waiting at least 5 minutes
     'retries': 1,
     'retry_delay': datetime.timedelta(minutes=5),
-    'project_id': models.Variable.get('gcp_project'),
-    'location': models.Variable.get('gce_region'),
+    'project_id': '{{ var.value.gcp_project }}',
+    'region': '{{ var.value.gce_region }}',
 
 }
 
@@ -104,7 +104,8 @@
         # See https://airflow.apache.org/docs/apache-airflow/stable/macros-ref.html
         cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
         cluster_config=CLUSTER_CONFIG,
-        region=models.Variable.get('gce_region'))
+        region='{{ var.value.gce_region }}'
+    )
 
     # Run the Hadoop wordcount example installed on the Cloud Dataproc cluster
     # master node.
@@ -116,7 +117,7 @@
     delete_dataproc_cluster = dataproc.DataprocDeleteClusterOperator(
         task_id='delete_dataproc_cluster',
         cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
-        region=models.Variable.get('gce_region'),
+        region='{{ var.value.gce_region }}',
         # Setting trigger_rule to ALL_DONE causes the cluster to be deleted
         # even if the Dataproc job fails.
         trigger_rule=trigger_rule.TriggerRule.ALL_DONE)