Update Airflow version and samples

Author: luisbelloch

airflow/Makefile

@@ -1,11 +1,13 @@
+AIRFLOW_VERSION:=2.4.3
+
 .PHONY: all
 all: clean docker-compose.yaml airflow init up
 
 docker-compose.yaml:
-	curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.2.2/docker-compose.yaml'
+	curl -LfO 'https://airflow.apache.org/docs/apache-airflow/${AIRFLOW_VERSION}/docker-compose.yaml'
 
 airflow:
-	curl -Lf 'https://airflow.apache.org/docs/apache-airflow/2.2.2/airflow.sh' > airflow
+	curl -Lf 'https://airflow.apache.org/docs/apache-airflow/${AIRFLOW_VERSION}/airflow.sh' > airflow
 	chmod +x airflow
 
 .PHONY: init

airflow/dags/s3_bucket_operations.py

@@ -4,7 +4,7 @@
 from airflow.decorators import task
 from airflow.models.dag import DAG
 from airflow.providers.amazon.aws.hooks.s3 import S3Hook
-from airflow.providers.amazon.aws.operators.s3_bucket import S3CreateBucketOperator, S3DeleteBucketOperator
+from airflow.providers.amazon.aws.operators.s3 import S3CreateBucketOperator, S3DeleteBucketOperator
 
 # By default, it will use 'aws_default' connection. You can create it here by running `make minio_credentials`
 # If you want to change it, use a variable and pass it as `aws_conn_id` to all AWS operators.

airflow/dags/s3_file_sensor.py

@@ -4,7 +4,7 @@
 from airflow.decorators import task
 from airflow.models.dag import DAG
 from airflow.models.variable import Variable
-from airflow.providers.amazon.aws.sensors.s3_key import S3KeySensor, S3KeySizeSensor
+from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
 
 BUCKET_NAME = os.environ.get('BUCKET_NAME', 'patatas')
 

airflow/dags/spark_ondemand.py

@@ -1,3 +1,8 @@
+# ./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
 
@@ -27,14 +32,15 @@
         task_id='create_dataproc_cluster',
         cluster_name='spark-cluster-{{ ds_nodash }}',
         num_workers=2,
-        zone=models.Variable.get('gce_zone'),
+        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='europe-west1',
+        region=models.Variable.get('gcp_region'),
         main='gs://bigdataupv_code/compras_top_ten_countries.py',
         files=['gs://bigdataupv_code/helpers.py'])
 

airflow/dags/spark_simple.py

@@ -26,7 +26,7 @@
   run_step = dataproc_operator.DataProcPySparkOperator(
       task_id='run_spark',
       cluster_name='cluster-9c11',
-      region='europe-west1',
+      region=models.Variable.get('gcp_region'),
       main='gs://bigdataupv_code/compras_top_ten_countries.py',
       files=['gs://bigdataupv_code/helpers.py'])