Revamp airflow samples

Author: luisbelloch

airflow/.env

@@ -0,0 +1 @@
+AIRFLOW_UID=50000

airflow/.gitignore

@@ -0,0 +1,5 @@
+docker-compose.yaml
+logs/
+plugins/
+.minio/
+airflow

airflow/Dockerfile

@@ -1,30 +1,34 @@
-IMAGE_NAME:=luisbelloch/airflow
-VERSION_TAG:=2019.1
-DAG_VOLUME:=${PWD}/dags:/var/airflow/dags
-WEB_PORT:=8080:8080
+.PHONY: all
+all: clean docker-compose.yaml airflow init up
 
-.PHONY: build
-build:
-	docker build -t $(IMAGE_NAME) .
+docker-compose.yaml:
+	curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.2.2/docker-compose.yaml'
 
-.PHONY: tag
-tag:
-	docker tag ${IMAGE_NAME} ${IMAGE_NAME}:${VERSION_TAG}
+airflow:
+	curl -Lf 'https://airflow.apache.org/docs/apache-airflow/2.2.2/airflow.sh' > airflow
+	chmod +x airflow
 
-.PHONY: push
-push:
-	docker push ${IMAGE_NAME}
-	docker push ${IMAGE_NAME}:${VERSION_TAG}
+.PHONY: init
+init:
+	docker-compose up airflow-init
 
-.PHONY: web
-web:
-	docker run -p ${WEB_PORT} -v ${DAG_VOLUME} -ti $(IMAGE_NAME) airflow webserver
+.PHONY: up
+up:
+	docker-compose up
 
-.PHONY: scheduler
-scheduler:
-	docker run -p ${WEB_PORT} -v ${DAG_VOLUME} -ti $(IMAGE_NAME) airflow scheduler
+.PHONY: down
+down:
+	docker-compose down --remove-orphans
 
-.PHONY: shell
-shell:
-	docker run -p ${WEB_PORT} -v ${DAG_VOLUME} -ti $(IMAGE_NAME) /bin/bash
+.PHONY: minio
+minio:
+	docker-compose -f docker-compose.yaml -f docker-compose.minio.yaml up minio
 
+.PHONY: minio_connection
+minio_connection: airflow
+	./airflow connections import minio_connection.json
+
+.PHONY: clean
+clean:
+	-docker-compose down --volumes --remove-orphans
+	-rm -rf logs/ plugins/ .minio/ docker-compose.yaml airflow dags/__pycache__

airflow/Makefile

@@ -0,0 +1,39 @@
+import json
+from datetime import datetime
+
+from airflow import AirflowException
+from airflow.decorators import dag, task
+
+import requests
+
+codigo_postal = "50197"
+endpoint = "https://sedeaplicaciones.minetur.gob.es/ServiciosRESTCarburantes/PreciosCarburantes/EstacionesTerrestres/"
+
+@dag(schedule_interval=None, start_date=datetime(2021, 1, 1), catchup=False, tags=['upv'])
+def extraer_precio_gasolina_naive():
+
+  @task
+  def recogida():
+    print("Recogiendo datos...")
+    response = requests.get(endpoint)
+    if response.status_code != 200:
+      AirflowException(f"Fallo de conexión {response.status_code}")
+
+    datos = response.json()
+    return datos['ListaEESSPrecio']
+
+  @task
+  def filtrado(datos, codigo_postal):
+    return list(filter(lambda x: x['C.P.'] == codigo_postal, datos))
+
+  @task
+  def almacenamiento(datos):
+    print("Almacenando datos...")
+    print(json.dumps(datos, indent=2))
+
+  todos_los_datos = recogida()
+  datos_del_codigo_postal_x = filtrado(todos_los_datos, codigo_postal)
+  almacenamiento(datos_del_codigo_postal_x)
+
+dag_gasolina = extraer_precio_gasolina_naive()
+

airflow/dags/gasolina_naive.py

@@ -0,0 +1,74 @@
+import json
+import socket
+
+from datetime import datetime
+
+from airflow import AirflowException
+from airflow.decorators import dag, task
+from airflow.operators.bash_operator import BashOperator
+from airflow.contrib.sensors.file_sensor import FileSensor
+
+import boto3
+import botocore.client
+import requests
+
+codigo_postal = "50197"
+bucket_name = "gasolina"
+endpoint = "https://sedeaplicaciones.minetur.gob.es/ServiciosRESTCarburantes/PreciosCarburantes/EstacionesTerrestres/"
+
+def s3_resource():
+  return boto3.resource('s3',
+        endpoint_url='http://minio:9000',
+        aws_access_key_id='bigdataupv',
+        aws_secret_access_key='bigdataupv',
+        config=botocore.client.Config(signature_version='s3v4'), region_name='us-east-1')
+
+def read_json_from_s3(key):
+  obj = s3_resource().Object(bucket_name, key)
+  return json.loads(obj.get()['Body'].read().decode('utf-8'))
+
+def save_to_s3(key, data):
+  obj = s3_resource().Object(bucket_name, key)
+  obj.put(Body=data)
+
+@dag(schedule_interval=None, start_date=datetime(2021, 1, 1), catchup=False, tags=['upv'])
+def extraer_precio_gasolina_s3():
+
+  @task
+  def recogida_s3():
+    print("Recogiendo datos...")
+    response = requests.get(endpoint)
+    if response.status_code != 200:
+      AirflowException(f"Fallo de conexión {response.status_code}")
+
+    filename = f'recogida-{datetime.now().strftime("%Y%m%d%H%M%S")}.json'
+    save_to_s3(filename, response.text)
+    return { "recogida": filename }
+
+  @task
+  def filtrado_s3(contexto, codigo_postal):
+    print("Filtrando datos...")
+
+    datos = read_json_from_s3(contexto['recogida'])
+    filtrados = list(filter(lambda x: x['C.P.'] == codigo_postal, datos['ListaEESSPrecio']))
+
+    filename = f'filtrado-{datetime.now().strftime("%Y%m%d%H%M%S")}.json'
+    save_to_s3(filename, json.dumps(filtrados))
+
+    return { **contexto, "filtrado": filename }
+
+  @task
+  def almacenamiento_s3(contexto):
+    print("Almacenando datos... Nothing to do!")
+    import socket
+    print("hostname:", socket.gethostname())
+    return 42
+
+  todos_los_datos = recogida_s3()
+  datos_del_codigo_postal_x = filtrado_s3(todos_los_datos, codigo_postal)
+  almacenamiento_s3(datos_del_codigo_postal_x)
+
+dag_gasolina = extraer_precio_gasolina_s3()
+
+# Additionally, use Amazon operator, particularly S3KeySensor
+# https://airflow.apache.org/docs/apache-airflow-providers-amazon/stable/operators/s3.html

airflow/dags/gasolina_s3.py

@@ -7,7 +7,7 @@
 from airflow.operators.dummy_operator import DummyOperator
 from airflow.operators.python_operator import PythonOperator
 
-dag = DAG('hello_dags', schedule_interval=None, start_date=airflow.utils.dates.days_ago(2))
+dag = DAG('hello_dags', schedule_interval=None, start_date=airflow.utils.dates.days_ago(2), tags=['upv'])
 
 def print_hello():
     return 'Hello world!'
@@ -18,8 +18,7 @@ def print_hello():
 paso3 = DummyOperator(task_id='paso3', dag=dag)
 ultima_tarea = DummyOperator(task_id='ultima_tarea', dag=dag)
 
-inicio >> paso1
-inicio >> paso3
+inicio >> [paso1, paso3]
 paso1 >> paso2 >> ultima_tarea
 paso3 >> ultima_tarea
 

airflow/dags/hello_dags.py

@@ -12,6 +12,7 @@ def print_hello():
     description='Simple tutorial DAG',
     schedule_interval='20 * * * *',
     start_date=datetime(2017, 3, 20),
+    tags=['upv'],
     catchup=False)
 
 dummy_operator = DummyOperator(task_id='dummy_task', retries=3, dag=dag)

airflow/dags/hello_python_operator.py

@@ -4,7 +4,7 @@
 from airflow.operators.dummy_operator import DummyOperator
 
 default_args = {"start_date": datetime(2019, 2, 5)}
-dag = DAG('hello', default_args=default_args, schedule_interval=None)
+dag = DAG('hello', default_args=default_args, schedule_interval=None, tags=['upv'],)
 
 dummy_operator = DummyOperator(task_id='dummy_task', dag=dag)
 hello_operator = DummyOperator(task_id='hello_task', dag=dag)

airflow/dags/hello_simple.py

@@ -0,0 +1,34 @@
+import os
+from datetime import datetime
+
+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
+
+# 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.
+AWS_CONN_ID = 'aws_default'
+
+BUCKET_NAME = os.environ.get('BUCKET_NAME', 'patatas')
+
+@task(task_id="s3_bucket_dag_add_keys_to_bucket")
+def upload_keys():
+  s3_hook = S3Hook()
+  for i in range(0, 3):
+    s3_hook.load_string(string_data="input", key=f"path/data{i}", bucket_name=BUCKET_NAME)
+
+with DAG(
+  dag_id='s3_bucket_dag',
+  schedule_interval=None,
+  start_date=datetime(2021, 1, 1),
+  catchup=False,
+  default_args={"bucket_name": BUCKET_NAME},
+  max_active_runs=1,
+  tags=['upv'],
+) as dag:
+
+  create_bucket = S3CreateBucketOperator(task_id='s3_bucket_dag_create', region_name='us-east-1')
+  add_keys_to_bucket = upload_keys()
+  delete_bucket = S3DeleteBucketOperator(task_id='s3_bucket_dag_delete', force_delete=True)
+  create_bucket >> add_keys_to_bucket >> delete_bucket