Merge pull request #39 from DataTalksClub/feature/week2_airflow

Feature/week2 airflow

Author: sejalv

week_2_data_ingestion/airflow/1_setup.md

@@ -11,6 +11,9 @@
 2. You may need to upgrade your docker-compose version to v2.x+, and set the memory for your Docker Engine to minimum 5GB
 (ideally 8GB). If enough memory is not allocated, it might lead to airflow-webserver continuously restarting.
 
+3. Python version: 3.7+
+
+
 ### Airflow Setup
 
 1. Create a new sub-directory called `airflow` in your `project` dir (such as the one we're currently in)

week_2_data_ingestion/airflow/2_concepts.md

@@ -27,7 +27,22 @@ Backend to the Airflow environment. Used by the scheduler, executor and webserve
 All these services allow you to run Airflow with CeleryExecutor. 
 For more information, see [Architecture Overview](https://airflow.apache.org/docs/apache-airflow/stable/concepts/overview.html).
 
-Directories:
+#### Project Structure:
 * `./dags` - `DAG_FOLDER` for DAG files
 * `./logs` - contains logs from task execution and scheduler.
 * `./plugins` - for custom plugins
+
+
+#### Workflow components
+* `DAG`: Directed acyclic graph, specifies the dependencies between a set of tasks with explicit execution order, and has a beginning as well as an end. (Hence, “acyclic”)
+    * `DAG Structure`: DAG Definition, Tasks (eg. Operators), Task Dependencies (control flow: `>>` or `<<` )
+    
+* `Task`: a defined unit of work (aka, operators in Airflow). The Tasks themselves describe what to do, be it fetching data, running analysis, triggering other systems, or more.
+    * Common Types: Operators (used in this workshop), Sensors, TaskFlow decorators
+    * Sub-classes of Airflow's BaseOperator
+
+* `DAG Run`: individual execution/run of a DAG
+    * scheduled or triggered
+
+* `Task Instance`: an individual run of a single task. Task instances also have an indicative state, which could be “running”, “success”, “failed”, “skipped”, “up for retry”, etc.
+    * Ideally, a task should flow from `none`, to `scheduled`, to `queued`, to `running`, and finally to `success`.

week_2_data_ingestion/airflow/README.md

@@ -27,6 +27,10 @@ docker-compose up
 
 5. Run your DAG on the Web Console.
 
+6. On finishing your run or to shut down the container/s:
+```shell
+docker-compose down
+```
 
 For more info, check out these official docs:
    * https://airflow.apache.org/docs/apache-airflow/stable/start/docker.html

week_2_data_ingestion/airflow/dags/data_ingestion_gcs_dag.py

@@ -1,5 +1,4 @@
 import os
-from datetime import datetime
 import logging
 
 from airflow import DAG
@@ -12,15 +11,13 @@
 import pyarrow.csv as pv
 import pyarrow.parquet as pq
 
-PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "pivotal-surfer-336713")
-BUCKET = os.environ.get("GCP_GCS_BUCKET", "dtc_data_lake_pivotal-surfer-336713")
+PROJECT_ID = os.environ.get("GCP_PROJECT_ID")
+BUCKET = os.environ.get("GCP_GCS_BUCKET")
 
 dataset_file = "yellow_tripdata_2021-01.csv"
 dataset_url = f"https://s3.amazonaws.com/nyc-tlc/trip+data/{dataset_file}"
 path_to_local_home = os.environ.get("AIRFLOW_HOME", "/opt/airflow/")
 parquet_file = dataset_file.replace('.csv', '.parquet')
-
-path_to_creds = f"{path_to_local_home}/google_credentials.json"
 BIGQUERY_DATASET = os.environ.get("BIGQUERY_DATASET", 'trips_data_all')
 
 
@@ -61,6 +58,7 @@ def upload_to_gcs(bucket, object_name, local_file):
     "retries": 1,
 }
 
+# NOTE: DAG declaration - using a Context Manager (an implicit way)
 with DAG(
     dag_id="data_ingestion_gcs_dag",
     schedule_interval="@daily",
@@ -83,7 +81,7 @@ def upload_to_gcs(bucket, object_name, local_file):
         },
     )
 
-    # TODO: Homework: research and try XCOM to communicate output values between 2 tasks/operators
+    # TODO: Homework - research and try XCOM to communicate output values between 2 tasks/operators
     local_to_gcs_task = PythonOperator(
         task_id="local_to_gcs_task",
         python_callable=upload_to_gcs,
@@ -109,4 +107,4 @@ def upload_to_gcs(bucket, object_name, local_file):
         },
     )
 
-    download_dataset_task >> format_to_parquet >> local_to_gcs_task >> bigquery_external_table_task
+    download_dataset_task >> format_to_parquet >> local_to_gcs_task >> bigquery_external_table_task

week_2_data_ingestion/airflow/docker-compose.yaml

@@ -60,12 +60,15 @@ x-airflow-common:
     _PIP_ADDITIONAL_REQUIREMENTS: ${_PIP_ADDITIONAL_REQUIREMENTS:-}
     GOOGLE_APPLICATION_CREDENTIALS: /.google/credentials/google_credentials.json
     AIRFLOW_CONN_GOOGLE_CLOUD_DEFAULT: 'google-cloud-platform://?extra__google_cloud_platform__key_path=/.google/credentials/google_credentials.json'
+    GCP_PROJECT_ID: 'pivotal-surfer-336713'
+    GCP_GCS_BUCKET: "dtc_data_lake_pivotal-surfer-336713"
 
   volumes:
     - ./dags:/opt/airflow/dags
     - ./logs:/opt/airflow/logs
     - ./plugins:/opt/airflow/plugins
     - ~/.google/credentials/:/.google/credentials:ro
+
   user: "${AIRFLOW_UID:-50000}:0"
   depends_on:
     &airflow-common-depends-on