ssh -i /path/to/your-key.pem ubuntu@<your-ec2-public-ip>

# Update package lists and upgrade existing packages
sudo apt update && sudo apt upgrade -y

# Install Python tools and PostgreSQL client libraries
sudo apt install -y python3-pip python3.12-venv postgresql-client libpq-dev

sudo apt install -y postgresql postgresql-contrib

sudo -i -u postgres psql

-- Create a dedicated user for Airflow
CREATE USER airflow WITH PASSWORD 'strong_password';

-- Create the database
CREATE DATABASE airflow;

-- Set the new user as owner
ALTER DATABASE airflow OWNER TO airflow;

-- Exit psql
\q

exit

echo "export AIRFLOW_HOME=~/airflow" >> ~/.bashrc
source ~/.bashrc

python3 -m venv ~/airflow_venv
source ~/airflow_venv/bin/activate

pip install "apache-airflow[postgres]==2.9.2" --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.9.2/constraints-3.12.txt"

airflow db upgrade

nano ~/airflow/airflow.cfg

# Change executor to allow parallel task execution
executor = LocalExecutor

# Update database connection string
sql_alchemy_conn = postgresql+psycopg2://airflow:strong_password@localhost/airflow

# (Optional) Disable example DAGs
load_examples = False

airflow db upgrade

airflow users create \
  --username admin \
  --role Admin \
  --firstname Your \
  --lastname Name \
  --email you@example.com

# Start the webserver on port 8080
airflow webserver --port 8080 -D

# Start the scheduler
airflow scheduler -D

http://<your-ec2-public-ip>:8080

sudo nano /etc/systemd/system/airflow-webserver.service

[Unit]
Description=Airflow Webserver
After=network.target postgresql.service
Requires=postgresql.service

[Service]
Environment="AIRFLOW_HOME=/home/ubuntu/airflow"
User=root
Group=root
Type=simple
ExecStart=/home/ubuntu/airflow_venv/bin/airflow webserver --port 8080
Restart=on-failure
RestartSec=10s

[Install]
WantedBy=multi-user.target

sudo nano /etc/systemd/system/airflow-scheduler.service

[Unit]
Description=Airflow Scheduler
After=network.target postgresql.service
Requires=postgresql.service

[Service]
Environment="AIRFLOW_HOME=/home/ubuntu/airflow"
User=root
Group=root
Type=simple
ExecStart=/home/ubuntu/airflow_venv/bin/airflow scheduler
Restart=on-failure
RestartSec=10s

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl start airflow-webserver
sudo systemctl start airflow-scheduler
sudo systemctl enable airflow-webserver
sudo systemctl enable airflow-scheduler
sudo systemctl status airflow-webserver
sudo systemctl status airflow-scheduler

sudo systemctl restart airflow-webserver airflow-scheduler

from __future__ import annotations
import pendulum
from airflow.decorators import dag, task

@dag(
    dag_id="my_first_cron_dag",
    start_date=pendulum.datetime(2025, 9, 30, tz="UTC"),
    schedule="*/5 * * * *",
    catchup=False,
    tags=["example", "cron"],
)
def my_simple_cron_workflow():
    @task
    def print_a_message():
        current_time = pendulum.now("UTC")
        print(f"Hello from Airflow! The time is now: {current_time}")

    print_a_message()

my_simple_cron_workflow()

from __future__ import annotations
import pendulum
from datetime import timedelta
from airflow.decorators import dag, task
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator

@dag(
    dag_id="chaotic_testing_dag",
    start_date=pendulum.datetime(2025, 9, 30, tz="UTC"),
    schedule="*/10 * * * *",
    catchup=False,
    tags=["example", "testing", "failure"],
)
def chaotic_workflow():
    start = EmptyOperator(task_id="start")
    
    task_that_fails_on_purpose = BashOperator(
        task_id="task_that_fails_on_purpose",
        bash_command='echo "This task will fail with exit code 1."; exit 1',
    )
    
    task_that_will_not_run = BashOperator(
        task_id="task_that_will_not_run",
        bash_command='echo "This task will not run."',
    )
    
    @task
    def task_with_a_python_exception():
        raise ValueError("This is a deliberate Python error!")
    
    task_that_will_timeout = BashOperator(
        task_id="task_that_will_timeout",
        bash_command='echo "This task will time out."; sleep 60',
        execution_timeout=timedelta(seconds=15),
    )
    
    end = EmptyOperator(task_id="end")
    
    start >> [task_that_fails_on_purpose, task_with_a_python_exception(), task_that_will_timeout]
    task_that_fails_on_purpose >> task_that_will_not_run >> end
    task_with_a_python_exception() >> end
    task_that_will_timeout >> end

chaotic_workflow()