Skip to content

Commit 6580e94

Browse files
Sharmila-D3Sharmila-D3
committed
Initial commit: Airflow + EMR + Snowflake pipeline with PySpark jobs and architecture diagram
0 parents  commit 6580e94

File tree

8 files changed

+410
-0
lines changed

8 files changed

+410
-0
lines changed

‎.github/workflows/ci.yml‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
name: CI
2+
3+
on:
4+
push:
5+
branches: [ main ]
6+
pull_request:
7+
branches: [ main ]
8+
9+
jobs:
10+
lint-and-test:
11+
runs-on: ubuntu-latest
12+
steps:
13+
- uses: actions/checkout@v4
14+
- name: Set up Python
15+
uses: actions/setup-python@v4
16+
with:
17+
python-version: '3.10'
18+
- name: Install deps
19+
run: |
20+
python -m pip install --upgrade pip
21+
pip install flake8 pytest
22+
- name: Lint with flake8
23+
run: |
24+
flake8 . --max-line-length=120 || true
25+
- name: Run tests
26+
run: |
27+
pytest -q || true

‎.gitignore‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
# Python
2+
*.pyc
3+
__pycache__/
4+
5+
# Local env files
6+
.env
7+
.env.*
8+
secrets/
9+
10+
# Airflow
11+
airflow.cfg
12+
unittests.cfg
13+
logs/
14+
dags/__pycache__/
15+
16+
# OS
17+
.DS_Store

‎Architecture.png‎

550 KB
Loading

‎README.md‎

Lines changed: 133 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,133 @@
1+
# Automated Data Pipeline using Apache Airflow, AWS EMR, PySpark & Snowflake
2+
A fully automated data engineering pipeline orchestrated using **Apache Airflow**, running **PySpark** jobs on **AWS EMR**, and integrating data from **Snowflake** into aggregated datasets stored in **Amazon S3**.
3+
4+
![Architecture Diagram](Architecture.png)
5+
6+
This project demonstrates a production-style workflow including:
7+
- EMR cluster provisioning from Airflow
8+
- Executing multiple PySpark jobs in sequence
9+
- Extract-transform-load (ETL) from Snowflake
10+
- S3 data ingestion + aggregations
11+
- Joining processed datasets into a final output table
12+
- Sanitized, environment-variable driven configuration
13+
- CI workflow included (GitHub Actions)
14+
15+
---
16+
17+
## 🚀 Architecture Overview
18+
19+
**Airflow DAG → EMR Cluster → PySpark Jobs → S3 → Snowflake → S3 Final Output**
20+
21+
Pipeline Flow:
22+
1. Airflow creates an EMR cluster dynamically using `EmrCreateJobFlowOperator`
23+
2. Airflow submits 3 Spark jobs to EMR:
24+
- `s3.py`: Reads raw data from S3 → aggregates IP counts
25+
- `snow.py`: Reads Snowflake table via Spark Snowflake Connector → aggregates scores
26+
- `master.py`: Joins the outputs and stores final results in S3
27+
3. EMR auto-terminates after job completion
28+
4. All configuration is handled via environment variables (no hardcoded secrets)
29+
30+
---
31+
32+
## 📂 Project Structure
33+
34+
```
35+
dags/
36+
final_dag.py # Airflow DAG (sanitized, uses env vars)
37+
spark_jobs/
38+
s3.py # PySpark job: S3 ingest + aggregation
39+
snow.py # PySpark job: Snowflake extract + S3 write
40+
master.py # PySpark job: join + final output
41+
.env.example # Safe reference for required env variables
42+
.gitignore
43+
README.md
44+
.github/workflows/ci.yml # CI pipeline for linting
45+
```
46+
47+
---
48+
49+
## 🔐 Environment Variables (Required)
50+
All sensitive values must come through environment variables.
51+
52+
See `.env.example` for the full list:
53+
54+
### AWS / EMR
55+
```
56+
AWS_REGION
57+
AWS_CONN_ID
58+
LOG_BUCKET
59+
S3_BUCKET
60+
EC2_KEYNAME
61+
EC2_SUBNET
62+
MASTER_SG
63+
SLAVE_SG
64+
EMR_SERVICE_ROLE
65+
EMR_JOBFLOW_ROLE
66+
```
67+
68+
### Snowflake
69+
```
70+
SF_URL
71+
SF_ACCOUNT
72+
SF_USER
73+
SF_PASSWORD
74+
SF_DATABASE
75+
SF_SCHEMA
76+
SF_WAREHOUSE
77+
SF_ROLE
78+
```
79+
80+
### S3 Paths
81+
```
82+
SRC_S3_PATH
83+
DEST_IPCOUNT_PATH
84+
DEST_SCORES_PATH
85+
MASTER_OUT_PATH
86+
```
87+
88+
---
89+
90+
## 🧪 How to Run (High-Level)
91+
92+
1. Upload `spark_jobs/*.py` to S3 bucket defined in `S3_BUCKET`
93+
2. Place `final_dag.py` inside Airflow's `dags/` folder
94+
3. Set required environment variables in Airflow (or use Connections/Variables)
95+
4. Airflow triggers the DAG (manually or on schedule)
96+
5. EMR spins up → runs Spark jobs → terminates automatically
97+
98+
---
99+
100+
## 📌 CI Pipeline Included
101+
This repo includes GitHub Actions (`ci.yml`) that runs:
102+
- Lint checks (`flake8`)
103+
- Basic test structure
104+
105+
---
106+
107+
## 📸 Screenshots
108+
You can add:
109+
- Airflow DAG view
110+
- EMR cluster step execution
111+
- Snowflake query results
112+
- S3 output folders
113+
114+
*(Ensure no secrets or account IDs are visible.)*
115+
116+
---
117+
118+
## 📘 Why This Project is Valuable
119+
This repository demonstrates real-world production data engineering skills:
120+
- Workflow orchestration
121+
- Distributed Spark processing
122+
- Data integration (Snowflake ↔ AWS)
123+
- CI/CD and clean repo structure
124+
- Secure coding practices (env vars only)
125+
- Cloud-native automation (EMR, S3, IAM)
126+
127+
Excellent for showcasing on a **resume, LinkedIn, or interviews**.
128+
129+
---
130+
131+
## 🏷️ License
132+
Open-source. No proprietary secrets included.
133+

‎dags/final_dag.py‎

Lines changed: 147 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
1+
from airflow import DAG
2+
from airflow.providers.amazon.aws.operators.emr import EmrCreateJobFlowOperator, EmrAddStepsOperator
3+
from airflow.utils.dates import days_ago
4+
from datetime import timedelta
5+
import pendulum
6+
import os
7+
8+
local_tz = pendulum.timezone("Asia/Kolkata")
9+
10+
default_args = {
11+
"owner": "airflow",
12+
"depends_on_past": False,
13+
"email_on_failure": False,
14+
"email_on_retry": False,
15+
"retries": 1,
16+
"retry_delay": timedelta(minutes=5),
17+
"start_date": days_ago(0),
18+
"timezone": local_tz,
19+
}
20+
21+
dag = DAG(
22+
"FINALDAG",
23+
default_args=default_args,
24+
description="DAG to create an EMR cluster and submit Spark steps",
25+
schedule_interval="3 9 * * *",
26+
tags=["emr", "spark"],
27+
catchup=False,
28+
)
29+
30+
# Read important values from environment variables (set these in Airflow environment/Connections/Variables)
31+
LOG_BUCKET = os.getenv("LOG_BUCKET", "<your-log-bucket>")
32+
S3_BUCKET = os.getenv("S3_BUCKET", "<your-bucket>")
33+
EMR_RELEASE_LABEL = os.getenv("EMR_RELEASE_LABEL", "emr-7.1.0")
34+
EMR_SERVICE_ROLE = os.getenv("EMR_SERVICE_ROLE", "<EMR-Service-Role-ARN>")
35+
EMR_JOBFLOW_ROLE = os.getenv("EMR_JOBFLOW_ROLE", "<EMR-JobFlow-Role>")
36+
EC2_KEYNAME = os.getenv("EC2_KEYNAME", "<ec2-keypair>")
37+
EC2_SUBNET = os.getenv("EC2_SUBNET", "<subnet-id>")
38+
MASTER_SG = os.getenv("MASTER_SG", "<master-sg>")
39+
SLAVE_SG = os.getenv("SLAVE_SG", "<slave-sg>")
40+
41+
JOB_FLOW_OVERRIDES = {
42+
"Name": "prod-cluster",
43+
"LogUri": f"s3://{LOG_BUCKET}/elasticmapreduce",
44+
"ReleaseLabel": EMR_RELEASE_LABEL,
45+
# Use env vars for all ARNs/role names
46+
"ServiceRole": EMR_SERVICE_ROLE,
47+
"Instances": {
48+
"InstanceGroups": [
49+
{
50+
"Name": "Primary",
51+
"Market": "ON_DEMAND",
52+
"InstanceRole": "MASTER",
53+
"InstanceType": "m5.xlarge",
54+
"InstanceCount": 1,
55+
"EbsConfiguration": {
56+
"EbsBlockDeviceConfigs": [
57+
{
58+
"VolumeSpecification": {"VolumeType": "gp2", "SizeInGB": 32},
59+
"VolumesPerInstance": 2,
60+
}
61+
]
62+
},
63+
}
64+
],
65+
"Ec2KeyName": EC2_KEYNAME,
66+
"Ec2SubnetId": EC2_SUBNET,
67+
"EmrManagedMasterSecurityGroup": MASTER_SG,
68+
"EmrManagedSlaveSecurityGroup": SLAVE_SG,
69+
"KeepJobFlowAliveWhenNoSteps": True,
70+
"TerminationProtected": False,
71+
},
72+
"Applications": [{"Name": "Hadoop"}, {"Name": "Spark"}],
73+
"VisibleToAllUsers": True,
74+
"JobFlowRole": EMR_JOBFLOW_ROLE,
75+
"Tags": [{"Key": "env", "Value": os.getenv("ENV", "prod")}],
76+
"ScaleDownBehavior": "TERMINATE_AT_TASK_COMPLETION",
77+
"AutoTerminationPolicy": {"IdleTimeout": int(os.getenv("EMR_IDLE_TIMEOUT", "60"))},
78+
}
79+
80+
SPARK_STEPS = [
81+
{
82+
"Name": "s3Job",
83+
"ActionOnFailure": "CONTINUE",
84+
"HadoopJarStep": {
85+
"Jar": "command-runner.jar",
86+
"Args": [
87+
"spark-submit",
88+
"--deploy-mode", "client",
89+
"--master",
90+
"local[*]",
91+
f"s3://{S3_BUCKET}/pyfiles/s3.py",
92+
],
93+
},
94+
},
95+
{
96+
"Name": "SnowJob",
97+
"ActionOnFailure": "CONTINUE",
98+
"HadoopJarStep": {
99+
"Jar": "command-runner.jar",
100+
"Args": [
101+
"spark-submit",
102+
"--deploy-mode",
103+
"client",
104+
"--packages",
105+
"net.snowflake:spark-snowflake_2.12:3.1.1",
106+
"--master",
107+
"local[*]",
108+
f"s3://{S3_BUCKET}/pyfiles/snow.py",
109+
],
110+
},
111+
},
112+
{
113+
"Name": "MasterJob",
114+
"ActionOnFailure": "CONTINUE",
115+
"HadoopJarStep": {
116+
"Jar": "command-runner.jar",
117+
"Args": [
118+
"spark-submit",
119+
"--deploy-mode",
120+
"client",
121+
"--packages",
122+
"net.snowflake:spark-snowflake_2.12:3.1.1",
123+
"--master",
124+
"local[*]",
125+
f"s3://{S3_BUCKET}/pyfiles/master.py",
126+
],
127+
},
128+
},
129+
]
130+
131+
create_emr_cluster = EmrCreateJobFlowOperator(
132+
task_id="create_emr_cluster",
133+
job_flow_overrides=JOB_FLOW_OVERRIDES,
134+
aws_conn_id=os.getenv("AWS_CONN_ID", "aws_default"),
135+
region_name=os.getenv("AWS_REGION", "ap-south-1"),
136+
dag=dag,
137+
)
138+
139+
add_spark_steps = EmrAddStepsOperator(
140+
task_id="add_spark_steps",
141+
job_flow_id="{{ task_instance.xcom_pull(task_ids='create_emr_cluster', key='return_value') }}",
142+
aws_conn_id=os.getenv("AWS_CONN_ID", "aws_default"),
143+
steps=SPARK_STEPS,
144+
dag=dag,
145+
)
146+
147+
create_emr_cluster >> add_spark_steps

‎spark_jobs/master.py‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
from pyspark import SparkConf, SparkContext
2+
from pyspark.sql import SparkSession
3+
import os
4+
5+
conf = SparkConf().setAppName("master_job").setMaster("local[*]").set("spark.default.parallelism", "1")
6+
sc = SparkContext(conf=conf)
7+
sc.setLogLevel("ERROR")
8+
spark = SparkSession.builder.getOrCreate()
9+
10+
IPCOUNT_PATH = os.getenv("DEST_IPCOUNT_PATH", "s3://<your-bucket>/dest/ipcount")
11+
SCORES_PATH = os.getenv("DEST_SCORES_PATH", "s3://<your-bucket>/dest/scores")
12+
MASTER_OUT = os.getenv("MASTER_OUT_PATH", "s3://<your-bucket>/dest/master")
13+
14+
ipdf = spark.read.load(IPCOUNT_PATH)
15+
scoresdf = spark.read.load(SCORES_PATH)
16+
17+
joindf = ipdf.join(scoresdf, ["username"], "inner")
18+
joindf.write.mode("overwrite").save(MASTER_OUT)
19+
20+
sc.stop()

‎spark_jobs/s3.py‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
from pyspark import SparkConf, SparkContext
2+
from pyspark.sql import SparkSession
3+
from pyspark.sql.functions import count
4+
import os
5+
6+
conf = SparkConf().setAppName("s3_job").setMaster("local[*]").set("spark.default.parallelism", "1")
7+
sc = SparkContext(conf=conf)
8+
sc.setLogLevel("ERROR")
9+
spark = SparkSession.builder.getOrCreate()
10+
11+
SRC_PATH = os.getenv("SRC_S3_PATH", "s3://<your-bucket>/src/")
12+
DEST_IPCOUNT_PATH = os.getenv("DEST_IPCOUNT_PATH", "s3://<your-bucket>/dest/ipcount")
13+
14+
# Read raw data from S3 (parquet/json/csv supported depending on how data was written)
15+
df = spark.read.load(SRC_PATH)
16+
17+
# basic aggregation
18+
aggdf = df.groupBy("username").agg(count("ip").alias("ipcount"))
19+
20+
# write output
21+
aggdf.write.mode("overwrite").save(DEST_IPCOUNT_PATH)
22+
23+
sc.stop()

0 commit comments

Comments
 (0)