create dag to execute docker image on ECS Fargate

Author: andresionek91

Dockerfile

@@ -10,7 +10,7 @@ ENV DEBIAN_FRONTEND noninteractive
 ENV TERM linux
 
 # Airflow
-ARG AIRFLOW_VERSION=1.10.10
+ARG AIRFLOW_VERSION=1.10.11
 ENV AIRFLOW_HOME=/usr/local/airflow
 ENV AIRFLOW_GPL_UNIDECODE=yes
 
@@ -66,7 +66,7 @@ RUN set -ex \
     && pip install pyOpenSSL \
     && pip install ndg-httpsclient \
     && pip install pyasn1 \
-    && pip install apache-airflow[async,crypto,celery,kubernetes,jdbc,password,postgres,s3,slack]==${AIRFLOW_VERSION} \
+    && pip install apache-airflow[async,crypto,celery,kubernetes,jdbc,password,postgres,s3,slack,amazon]==${AIRFLOW_VERSION} \
     && pip install werkzeug==${WERKZEUG_VERSION} \
     && pip install redis==${PYTHON_REDIS_VERSION} \
     && pip install celery[redis]==${CELERY_REDIS_VERSION} \

dags/crypto_extract_dag.py

@@ -0,0 +1,55 @@
+from datetime import datetime
+import logging
+
+from airflow import DAG
+from airflow.contrib.operators.ecs_operator import ECSOperator
+from airflow.models import Variable
+
+logger = logging.getLogger(__name__)
+
+default_args = {
+    'owner': 'andresionek91',
+    'start_date': datetime(2020, 5, 20),
+    'depends_on_past': False,
+    'provide_context': True
+}
+
+execution_date = '{{ ds }}'  # Access execution date
+
+with DAG('crypto_extract',
+         description='DAG to extract Daily Summaries from MercadoBitcoin',
+         schedule_interval='* 1 * * *',
+         catchup=True,
+         default_args=default_args) as dag:
+
+    t1 = ECSOperator(
+        task_id="crypto_extract",
+        dag=dag,
+        aws_conn_id="aws_default",
+        cluster="airflow-dev-ecs-cluster",
+        task_definition="dev-crypto-extract-image",
+        launch_type="FARGATE",
+        overrides={
+            "containerOverrides": [
+                {
+                    "name": "dev-crypto-extract-image",
+                    "command": ["python", "main.py", execution_date],
+                    "environment": [
+                        {
+                            'name': 'AWS_ACCESS_KEY_ID',
+                            'value': 'string'
+                        }
+                    ]
+                }
+            ],
+        },
+        network_configuration = {
+            "awsvpcConfiguration": {
+                "securityGroups": [Variable.get("security-group")],
+                "subnets": [Variable.get("subnet")],
+                "assignPublicIp": "ENABLED"
+            }
+        }
+    )
+
+    t1

requirements.txt

@@ -3,4 +3,6 @@ cryptography==2.9.2
 pandas==0.24.0
 toolz==0.9.0
 jinja2==2.11.2
-click==7.1.2
+click==7.1.2
+docker==4.2.0
+apache-airflow-backport-providers-amazon==2020.6.24