Merge pull request #32 from aaditshah18/feature/intermediate_labs2

Added Intermediate lab for cloud composer

Author: raminmohammadi

Labs/GCP_Labs/Cloud_Composer_Labs/Lab1/README.md

@@ -1 +1,61 @@
 # Watch the tutorial video for this lab at [Cloud Composer Lab1](https://youtu.be/JB_I416LQ7A)
+
+
+# Google Cloud Composer Airflow Tutorial
+
+Welcome to the Google Cloud Composer Airflow tutorial! This guide will help you understand the basics of using Apache Airflow with Google Cloud Composer, set up a simple Directed Acyclic Graph (DAG), and deploy it on Google Cloud Composer.
+
+## What is Google Cloud Composer?
+
+Google Cloud Composer is a fully managed workflow orchestration service built on Apache Airflow. It allows you to author, schedule, and monitor workflows as directed acyclic graphs (DAGs) of tasks. Cloud Composer helps you automate and orchestrate tasks across cloud and on-premises environments.
+
+## What is Apache Airflow?
+
+Apache Airflow is an open-source platform to programmatically author, schedule, and monitor workflows. It allows you to define workflows as code, ensuring they are versionable, testable, and maintainable. Airflow uses DAGs to represent workflows, where each node is a task and edges define the execution order.
+
+## Tutorial Overview
+
+In this tutorial, we will create a simple DAG consisting of two tasks:
+1. A `BashOperator` task that prints "Hello, Cloud Composer!".
+2. A `PythonOperator` task that runs a simple Python function.
+
+Refer to [this page](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/operators.html) for more operators.
+
+
+### Prerequisites
+
+- A Google Cloud Platform account.
+- A Google Cloud Composer environment set up.
+- Basic knowledge of Python and Airflow concepts.
+
+## Explanation of the DAG
+- default_args: Contains default arguments for the DAG. Here, it specifies the owner and start date.
+- dag: Defines the DAG with an ID, default arguments, description, and schedule interval.
+- hello_world_task: A BashOperator that prints "Hello, Cloud Composer!".
+- say_hello_task: A PythonOperator that runs a simple Python function.
+
+## Setting Up the Environment
+
+1. **Create a Google Cloud Composer environment:**
+   - Navigate to the Cloud Console.
+   - Go to the Cloud Composer section.
+   - Create a new environment.
+
+2. **Upload the DAG to Cloud Composer:**
+   - Once the environment is ready, upload the DAG file to the `dags` folder in your Cloud Storage bucket associated with the Composer environment.
+
+3. **Trigger the DAG:**
+   - Go to the Airflow web interface through the Cloud Composer environment.
+   - Manually trigger the DAG and observe the task execution.
+
+
+## Conclusion
+This tutorial provided a basic introduction to creating and deploying a DAG in Google Cloud Composer using Apache Airflow. With these foundations, you can start building more complex workflows to automate various tasks in your cloud and on-premises environments.
+
+
+## Important Links
+
+- Quickstart Guide to Google Cloud Composer - [Getting Started With Composer](https://cloud.google.com/composer/docs/run-apache-airflow-dag)
+- Installing Google Cloud CLI -  [Installing gcloud](https://cloud.google.com/sdk/docs/install-sdk)
+- Identity Management of Cloud Composer - [Control Access](https://cloud.google.com/composer/docs/how-to/access-control#composer-sa)
+- Airflow Operators - [Operators](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/operators.html)

Labs/GCP_Labs/Cloud_Composer_Labs/Lab2/README.md

@@ -0,0 +1,72 @@
+# Cloud Composer Intermediate Tutorial
+
+This lab contains a Cloud Composer Intermediate tutorial that demonstrates various features of Apache Airflow including task parameterization, file operations, HTTP requests, task dependencies, and the use of sensors in GCP's Cloud Composer. The tutorial is structured into three Directed Acyclic Graphs (DAGs), each showcasing different functionalities.
+
+## Prerequisites
+
+- Apache Airflow installed
+- Google Cloud Storage (GCS) access with appropriate permissions
+- Python 3.x
+- Required Python packages: pandas, requests, apache-airflow, apache-airflow-providers-google
+
+
+## DAGs Overview
+
+### DAG 1: Parameterize File Path and Use FileSensor
+- This DAG reads, processes a file, and uses a GCS File Sensor to detect the existence of a processed file.
+
+Tasks
+- read_and_serialize_task: Reads a CSV file from GCS, serializes it to JSON, and stores the result in XCom.
+- process_task: Processes the serialized data from the previous task, fills missing values, and writes the output back to GCS.
+- file_sensor_task: Waits for the processed file to appear in GCS, then logs the task completion status.
+
+### DAG 2: File Operations and HTTP Request
+This DAG performs file operations on GCS and makes an HTTP request.
+
+Tasks
+- file_op_task: Reads a file from GCS and logs its content.
+- http_request_task: Makes an HTTP GET request to a given URL and logs the response.
+
+### DAG 3: Task Dependencies
+This DAG demonstrates task dependencies using dummy tasks, a bash command, and Python callable tasks.
+
+Tasks
+- start_task: Dummy task used to start the DAG.
+- bash_task: Runs a simple bash command.
+- middle_task: Logs a message indicating the task execution.
+- branch_task: Dummy task used for branching logic.
+- end_task: Logs a message indicating the task completion.
+
+## Functions
+`read_and_serialize`
+Reads a CSV file from GCS, serializes it to JSON, and logs the content.
+
+`read_and_serialize_return`
+Wrapper function that returns the result of read_and_serialize
+
+`process_file`
+Processes the serialized data, fills missing values, and writes the output back to GCS.
+
+`file_operation`
+Reads a file from GCS and logs its content.
+
+`make_http_request`
+Makes an HTTP GET request to a specified URL and logs the response.
+
+`log_file_sensor_output`
+Logs the output of the file sensor task.
+
+## How to Run
+- Ensure Airflow is properly installed and configured.
+- Place the Python script and required files in the appropriate directories.
+- Trigger the DAGs using the Airflow UI or CLI.
+
+```
+airflow dags trigger dag_1_parameterize
+airflow dags trigger dag_file_and_http
+airflow dags trigger dag_3_dependencies
+```
+
+
+## Conclusion
+This tutorial covers the basics of using Airflow with google cloud composer for various tasks such as file operations, HTTP requests, and demonstrating task dependencies. By following this tutorial, you should gain a good understanding of how to create and manage DAGs in Airflow.

Labs/GCP_Labs/Cloud_Composer_Labs/Lab2/__init__.py

@@ -0,0 +1,175 @@
+from airflow import DAG
+from airflow.operators.dummy import DummyOperator
+from airflow.operators.python import PythonOperator
+from airflow.operators.bash import BashOperator
+from airflow.providers.google.cloud.sensors.gcs import GCSObjectExistenceSensor
+from datetime import datetime, timedelta
+from dag_functions import (
+    file_operation,
+    make_http_request,
+    process_file,
+    read_and_serialize_return,
+    log_file_sensor_output,
+    final_task,
+)
+import logging
+
+AIRFLOW_TASK = "airflow.task"
+OUTPUT_PATH = "us-east1-composer-airflow-1c67778d-bucket/data/dag_processed_file.csv"
+logger = logging.getLogger(AIRFLOW_TASK)
+
+default_args = {
+    'owner': 'Aadit',
+    'start_date': datetime(2023, 9, 17),
+    'retries': 0,  # Number of retries in case of task failure
+    'retry_delay': timedelta(minutes=5),  # Delay before retries
+}
+
+dag_1 = DAG(
+    'dag_1_parameterize',
+    default_args=default_args,
+    description='DAG to parameterize file path, process file, and use FileSensor',
+    schedule_interval=None,
+    catchup=False,
+)
+
+read_serialize_task = PythonOperator(
+    task_id='read_and_serialize',
+    python_callable=read_and_serialize_return,
+    op_kwargs={
+        'file_path': 'us-east1-composer-airflow-1c67778d-bucket/data/dag_processing_file.csv'
+    },
+    dag=dag_1,
+)
+
+process_task = PythonOperator(
+    task_id='process_file',
+    python_callable=process_file,
+    op_kwargs={
+        'output_path': OUTPUT_PATH,
+    },
+    provide_context=True,
+    dag=dag_1,
+)
+
+# File sensor task to check for the processed file's existence
+file_sensor_task = GCSObjectExistenceSensor(
+    task_id='file_sensor_task',
+    bucket='us-east1-composer-airflow-1c67778d-bucket',
+    object='data/dag_processed_file.csv',
+    poke_interval=10,
+    timeout=300,
+    dag=dag_1,
+    on_success_callback=log_file_sensor_output,
+    on_failure_callback=log_file_sensor_output,
+)
+
+# Final task to execute after the file sensor task
+final_processing_task = PythonOperator(
+    task_id='final_processing_task',
+    python_callable=final_task,
+    op_kwargs={
+        'output_path': OUTPUT_PATH,
+    },
+    dag=dag_1,
+)
+
+
+read_serialize_task >> process_task >> file_sensor_task >> final_processing_task
+
+dag_2 = DAG(
+    'dag_file_and_http',
+    default_args=default_args,
+    description='DAG for file operations and HTTP request',
+    schedule_interval=None,
+    catchup=False,
+)
+
+file_op_task = PythonOperator(
+    task_id='file_operation',
+    python_callable=file_operation,
+    op_kwargs={'file_path': OUTPUT_PATH},
+    dag=dag_2,
+)
+
+http_request_task = PythonOperator(
+    task_id='http_request',
+    python_callable=make_http_request,
+    op_kwargs={'url': 'https://jsonplaceholder.typicode.com/todos/1'},
+    dag=dag_2,
+)
+
+file_op_task >> http_request_task
+
+### DAG 3: Task Dependencies
+
+dag_3 = DAG(
+    'dag_3_dependencies',
+    default_args=default_args,
+    description='DAG to demonstrate task dependencies',
+    schedule_interval=None,
+    catchup=False,
+)
+
+# DummyOperator: Used for grouping and branching logic
+start_task = DummyOperator(
+    task_id='start_task',
+    dag=dag_3,
+)
+
+# BashOperator: Runs a simple bash command
+bash_task = BashOperator(
+    task_id='bash_task',
+    bash_command='echo "This is a bash command"',
+    dag=dag_3,
+)
+
+# PythonOperator: Runs a Python callable
+middle_task = PythonOperator(
+    task_id='middle_task',
+    python_callable=lambda: logger.info("Middle Task"),
+    dag=dag_3,
+    trigger_rule='all_done',  # Execute regardless of the upstream task's status
+)
+
+# DummyOperator: Used for grouping and branching logic
+branch_task = DummyOperator(
+    task_id='branch_task',
+    dag=dag_3,
+)
+
+# PythonOperator: Runs a Python callable
+end_task = PythonOperator(
+    task_id='end_task',
+    python_callable=lambda: logger.info("End Task"),
+    dag=dag_3,
+)
+
+# Set task dependencies
+
+"""
+The task executes with the following dependencies:
+
+- `start_task`: Initiates the workflow and triggers two parallel tasks.
+- `bash_task`: Executes a bash script (or a set of operations) and passes the result to `middle_task`.
+- `branch_task`: Executes an independent task (or a set of operations).
+- `middle_task`: Processes the output of `bash_task`.
+- `end_task`:  Finalizes the workflow by consuming the results of `middle_task` and `branch_task`.
+
+The dependency structure ensures that:
+    - `bash_task` and `branch_task` run concurrently.
+    - `middle_task` depends on the completion of `bash_task`.
+    - `end_task` waits for both `middle_task` and `branch_task` to finish.
+
+This parallel execution can optimize the overall runtime, especially when tasks are I/O bound or computationally independent.
+"""
+
+start_task >> [bash_task, branch_task]
+bash_task >> middle_task >> end_task
+branch_task >> end_task
+
+# If this script is run directly, allow command-line interaction with the DAG
+if __name__ == "__main__":
+    dag_1.cli()
+    dag_2.cli()
+    dag_3.cli()