DEV Community: Eric Katumo

Building a YouTube Channel Analytics Dashboard with Airflow, Spark, and Grafana

Eric Katumo — Fri, 25 Apr 2025 13:18:54 +0000

Introduction

In today's creator economy, YouTube content creators rely heavily on performance metrics to guide their content strategies. While YouTube's native analytics dashboard provides basic insights, it often lacks the flexibility, depth, and customization options that serious creators need. To address this limitation, I built an automated data engineering pipeline that extracts data from the YouTube API, processes it with Apache Spark, stores it in PostgreSQL, and visualizes it with Grafana.

This article details my implementation of this YouTube Channel Analytics Dashboard, developed as part of the Data Engineering Easter Holiday Challenge. The solution provides content creators with deeper insights into their channel performance, audience engagement, and optimal content strategies.

Problem Statement

YouTube content creators face several challenges when trying to analyze their channel performance:

YouTube's native dashboard offers limited flexibility and customization
Manual data collection is time-consuming, error-prone, and not scalable
Creators lack a centralized platform for tracking performance metrics over time
Identifying optimal publishing schedules and content types requires complex analysis

To solve these problems, creators need an automated analytics data pipeline that:

Periodically pulls video-level performance data from the YouTube API
Cleans, aggregates, and stores that data in a structured format
Delivers actionable insights through intuitive visualizations

Solution Architecture

The solution is built as a complete ETL (Extract, Transform, Load) pipeline with visualization capabilities:

The pipeline consists of the following components:

Extract: A Python script that pulls channel and video data from the YouTube API v3
Transform: Apache Spark processes the extracted data and adds derived metrics
Load: Processed data is stored in a PostgreSQL database
Orchestrate: Apache Airflow schedules and manages the pipeline execution
Visualize: Grafana dashboards provide interactive visualizations of the data

All components were deployed on an Azure VM, making this a cost-effective solution for individual creators or small agencies.

Implementation Details

Data Extraction

The extraction component uses the YouTube Data API v3 to collect channel and video data. The script handles API authentication, retrieves the channel's uploaded videos playlist, and extracts detailed metadata for each video.

Here's the implementation of the extraction script:

import os
from googleapiclient.discovery import build
from datetime import datetime
import json
import sys
from dotenv import load_dotenv

# Load variables from .env
load_dotenv()

# --- Configuration ---
YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY")
CHANNEL_ID = os.getenv("YOUTUBE_CHANNEL_ID")
OUTPUT_FILENAME = os.getenv("OUTPUT_FILENAME")

def validate_config():
    if not YOUTUBE_API_KEY:
        print("Error: YOUTUBE_API_KEY environment variable not set.", file=sys.stderr)
        sys.exit(1)
    if not CHANNEL_ID:
        print("Error: YOUTUBE_CHANNEL_ID environment variable not set.", file=sys.stderr)
        sys.exit(1)

def build_youtube_client():
    return build("youtube", "v3", developerKey=YOUTUBE_API_KEY)

def get_playlist_id(youtube, channel_id):
    try:
        request = youtube.channels().list(part="contentDetails", id=channel_id)
        response = request.execute()
        return response['items'][0]['contentDetails']['relatedPlaylists']['uploads'] if response['items'] else None
    except Exception as e:
        print(f"Error fetching channel details: {e}", file=sys.stderr)
        return None

def get_video_ids(youtube, playlist_id):
    video_ids, next_page_token = [], None
    while True:
        try:
            request = youtube.playlistItems().list(
                part="contentDetails", 
                playlistId=playlist_id, 
                maxResults=50, 
                pageToken=next_page_token
            )
            response = request.execute()
            video_ids.extend([item['contentDetails']['videoId'] for item in response['items']])
            next_page_token = response.get('nextPageToken')
            if not next_page_token:
                break
        except Exception as e:
            print(f"Error fetching playlist items: {e}", file=sys.stderr)
            break
    return video_ids

def get_video_details(youtube, video_ids):
    all_video_data = []
    for i in range(0, len(video_ids), 50):
        video_chunk = ",".join(video_ids[i:i+50])
        try:
            request = youtube.videos().list(part="snippet,statistics", id=video_chunk)
            response = request.execute()
            timestamp = datetime.utcnow().isoformat() + 'Z'
            for item in response['items']:
                video_data = {
                    'videoId': item['id'],
                    'title': item['snippet']['title'],
                    'publishedAt': item['snippet']['publishedAt'],
                    'viewCount': item['statistics'].get('viewCount'),
                    'likeCount': item['statistics'].get('likeCount'),
                    'commentCount': item['statistics'].get('commentCount'),
                    'extractionTimestamp': timestamp
                }
                all_video_data.append(video_data)
        except Exception as e:
            print(f"Error fetching video details: {e}", file=sys.stderr)
    return all_video_data

if __name__ == "__main__":
    print("Starting YouTube data extraction...")

    # Step 1: Validate configuration
    validate_config()

    # Step 2: Build YouTube client
    youtube = build_youtube_client()

    # Step 3: Get the uploads playlist ID
    playlist_id = get_playlist_id(youtube, CHANNEL_ID)
    if not playlist_id:
        print(f"Could not retrieve playlist ID for channel {CHANNEL_ID}.", file=sys.stderr)
        sys.exit(1)

    # Step 4: Fetch video IDs from the playlist
    video_ids = get_video_ids(youtube, playlist_id)
    if not video_ids:
        print(f"No videos found for channel {CHANNEL_ID}.", file=sys.stderr)
        sys.exit(1)

    # Step 5: Get video details
    video_data = get_video_details(youtube, video_ids)
    if not video_data:
        print("No video data fetched.", file=sys.stderr)
        sys.exit(1)

    # Step 6: Save video data to file
    try:
        with open(OUTPUT_FILENAME, 'w') as f:
            json.dump(video_data, f, indent=4)
        print(f"Data saved to {OUTPUT_FILENAME}")
    except IOError as e:
        print(f"Error writing data to {OUTPUT_FILENAME}: {e}", file=sys.stderr)
        sys.exit(1)

    print("Extraction process finished.")
    sys.exit(0)

The script follows these key steps:

Validates environment variables and API key
Retrieves the channel's uploads playlist ID
Extracts all video IDs from the playlist, handling pagination
Fetches detailed metadata for each video (in batches of 50 to respect API limits)
Saves the raw data to a JSON file for further processing

Data Transformation and Enrichment

The transformation component uses Apache Spark to process the raw JSON data and enrich it with additional metrics. This step converts date strings to timestamps, adds temporal dimensions (year, month, day, hour), and classifies videos based on performance.

import os
import json
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from dotenv import load_dotenv

load_dotenv()

POSTGRES_USER = os.getenv("POSTGRES_USER")
POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD")
POSTGRES_HOST = os.getenv("POSTGRES_HOST")
POSTGRES_PORT = os.getenv("POSTGRES_PORT")
POSTGRES_DB = os.getenv("POSTGRES_DB")

# Initialize Spark Session with PostgreSQL JDBC driver
spark = SparkSession.builder \
    .appName("YouTube Data Transformation") \
    .config("spark.jars", "/path/to/postgresql-driver.jar") \
    .getOrCreate()

# Read raw JSON data
df_raw = spark.read.option("multiline", "true").json("/path/to/youtube_videos_raw.json")
df_raw.createOrReplaceTempView("youtube_videos_raw")

# Basic transformation - type conversion and filtering
sql_query = """
    SELECT
        videoId,
        title,
        publishedAt,
        CAST(viewCount AS INT) AS viewCount,
        CAST(likeCount AS INT) AS likeCount,
        CAST(commentCount AS INT) AS commentCount,
        extractionTimestamp
    FROM youtube_videos_raw
    WHERE viewCount IS NOT NULL
    ORDER BY viewCount DESC
"""
df_transformed = spark.sql(sql_query)

# Register transformed DataFrame for enrichment
df_transformed.createOrReplaceTempView("youtube_transformed")

# Enhanced enrichment - adding temporal dimensions and performance classification
enriched_query = """
    SELECT
        *,
        YEAR(TO_TIMESTAMP(publishedAt)) AS year,
        MONTH(TO_TIMESTAMP(publishedAt)) AS month,
        DAY(TO_TIMESTAMP(publishedAt)) AS day_of_month,
        DATE_FORMAT(TO_TIMESTAMP(publishedAt), 'EEEE') AS day_of_week,
        HOUR(TO_TIMESTAMP(publishedAt)) AS hour,
        WEEKOFYEAR(TO_TIMESTAMP(publishedAt)) AS week,
        CASE
            WHEN viewCount > 100000 THEN 'viral'
            WHEN viewCount > 5000 THEN 'high'
            ELSE 'normal'
        END AS performance_class
    FROM youtube_transformed
"""
df_enriched = spark.sql(enriched_query)
df_enriched.show()

# Configure PostgreSQL connection
jdbc_url = f"jdbc:postgresql://{POSTGRES_HOST}:{POSTGRES_PORT}/{POSTGRES_DB}"
connection_properties = {
    "user": POSTGRES_USER,
    "password": POSTGRES_PASSWORD,
    "driver": "org.postgresql.Driver"
}

# Save enriched data to PostgreSQL
df_enriched.write \
    .jdbc(
        url=jdbc_url,
        table="dataengineering.youtube_videos_enriched",
        mode="append",
        properties=connection_properties
    )

print("✅ Enriched data successfully appended to PostgreSQL.")
spark.stop()

The transformation script:

Reads the raw JSON data extracted from the YouTube API
Performs basic transformations (type casting, filtering null values)
Enriches the data with additional dimensions:
- Temporal breakdowns (year, month, day, hour, week)
- Day of week for publishing analysis
- Performance classification based on view count thresholds
Loads the enriched data into a PostgreSQL database

Workflow Orchestration with Airflow

To ensure the pipeline runs consistently and reliably, I used Apache Airflow to orchestrate the workflow. The DAG (Directed Acyclic Graph) defines the sequence of tasks and their dependencies:

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator

# Default arguments
default_args = {
    "owner": "[USER_NAME]",
    "depends_on_past": False,
    "start_date": datetime(2025, 4, 23),
    "email": ["[EMAIL_ADDRESS]"],
    "email_on_failure": False,
    "email_on_retry": True,
    "retries": 2,
    "retry_delay": timedelta(minutes=1),
}

# DAG definition
with DAG(
    "youtube_analytics_pipeline",
    default_args=default_args,
    schedule_interval="@daily",
    catchup=False,
    tags=["YouTube", "Analytics", "Spark"],
) as dag:
    run_scraper = BashOperator(
        task_id="run_youtube_scraper",
        bash_command="""
        source /path/to/virtualenv/bin/activate && \
        python3 /path/to/youtube_scraper.py
        """,
    )

    run_transform = BashOperator(
        task_id="run_spark_transform",
        bash_command="""
        source /path/to/virtualenv/bin/activate && \
        python3 /path/to/spark_transform.py
        """,
    )

    run_scraper >> run_transform

The Airflow DAG:

Runs daily, with configurable retry logic
First executes the YouTube data extraction script
Once extraction is complete, triggers the Spark transformation and loading process
Provides error handling and notification capabilities
Can be extended to include additional tasks (like data validation or report generation)

Data Visualization with Grafana

After setting up the data pipeline, I created a comprehensive Grafana dashboard to visualize the YouTube channel analytics. The dashboard addresses the key questions content creators need answered:

Key Visualizations

Channel Growth Overview
- Time series graph of views, likes, and comments over time
- Cumulative view count by month
- Weekly growth rate comparison
Top Performing Videos
- Table of highest viewed videos with title, view count, likes, and comments
- Engagement rate calculation (likes + comments / views)
- Performance classification distribution
Publishing Time Analysis
- Heatmap of views by day of week and hour
- Bar chart of average view count by day of week
- Pie chart of views by time of day (morning, afternoon, evening, night)
Audience Engagement Patterns
- Line graph tracking engagement metrics over time
- Scatter plot of video length vs. engagement rate
- Correlation between publishing frequency and view count

Here's a sample query used for the day of week performance visualization:

SELECT 
  day_of_week, 
  AVG(viewCount) as avg_views,
  AVG(likeCount) as avg_likes,
  AVG(commentCount) as avg_comments
FROM 
  dataengineering.youtube_videos_enriched
GROUP BY 
  day_of_week
ORDER BY 
  CASE
    WHEN day_of_week = 'Monday' THEN 1
    WHEN day_of_week = 'Tuesday' THEN 2
    WHEN day_of_week = 'Wednesday' THEN 3
    WHEN day_of_week = 'Thursday' THEN 4
    WHEN day_of_week = 'Friday' THEN 5
    WHEN day_of_week = 'Saturday' THEN 6
    WHEN day_of_week = 'Sunday' THEN 7
  END

Data Analysis and Insights

Views by Time of Day

The pie chart reveals clear viewing patterns throughout the day:

Time Period	Audience Behavior
Morning (5am-11am)	Highest engagement - nearly half of all views
Night (9pm-4am)	Strong secondary viewing window
Afternoon (12pm-4pm)	Moderate engagement
Evening (5pm-8pm)	Lowest engagement period

Day of Week Performance

The bar chart highlights significant day-to-day variations:

Day	Avg. Views	Performance
Sunday	569	Peak performance (100%)
Thursday	507	Strong performer (89% of peak)
Wednesday	428	Good performer (75% of peak)
Friday	322	Moderate performer (57% of peak)
Monday	313	Moderate performer (55% of peak)
Saturday	310	Moderate performer (54% of peak)
Tuesday	183	Lowest performer (32% of peak)

Content Strategy Recommendations

Based on the analytics, here are data-driven recommendations:

Optimal Publishing Schedule:
- Primary: Sunday mornings (5am-11am)
- Secondary: Thursday mornings (5am-11am)
- Avoid: Tuesday releases entirely
Content Distribution Strategy:
- Release major content on Sundays to maximize initial viewership
- Schedule promotional content on Thursdays to build weekend momentum
- Use Wednesday slots for mid-week engagement maintenance
- Consider Monday content to start the week strong

Challenges and Solutions

During the implementation of this project, I encountered several challenges:

API Rate Limiting
- Challenge: YouTube API has strict quotas (10,000 units per day)
- Solution: Implemented batch processing and clever quota usage by minimizing API calls
Data Quality Issues
- Challenge: Missing statistics fields for some videos
- Solution: Added robust error handling and null value management in the Spark transformation

Future Enhancements

While the current implementation provides valuable insights, several enhancements could further improve the analytics platform:

Sentiment Analysis
- Integrate comment analysis using NLP to measure audience sentiment
- Track sentiment changes over time and correlation with content topics
Competitor Analysis
- Add capability to benchmark performance against similar channels
- Identify content gaps and opportunities
Recommendation Engine
- Develop a recommendation system for optimal publishing times
- Provide content suggestions based on historical performance
Automated Reporting
- Implement automatic weekly/monthly reports via email
- Add anomaly detection to alert on unusual performance changes
Cross-Platform Integration
- Connect with other social media APIs (Twitter, Instagram, TikTok)
- Provide holistic view of content performance across platforms

References

Building an Automated Weather Data Pipeline with Apache Kafka and Cassandra

Eric Katumo — Mon, 07 Apr 2025 10:02:16 +0000

This article creates an end-to-end weather data pipeline that collects weather data from multiple African cities, processes it through Apache Kafka, and stores it in a Cassandra database for analysis. The infrastructure is hosted on Microsoft Azure.

Introduction

Weather data pipelines are essential components of modern environmental monitoring systems. They enable organizations to collect, process, and analyze meteorological information in real-time, facilitating better decision-making. This tutorial demonstrates how to build a simple yet robust weather data pipeline using:

Python for data fetching and processing
OpenWeatherMap API as our data source
Apache Kafka for handling real-time data streams
Apache Cassandra for scalable data storage
Microsoft Azure for cloud infrastructure

By the end of this article, you will understand how these technologies are combined to construct a data pipeline that can be scaled to more complex use cases.

Architecture Overview

Our weather data pipeline consists of the following workflow:

Data Extraction: Weather data is extracted for five African cities from the OpenWeatherMap API using the producer script
Data Streaming: Weather data is streamed into a Kafka topic
Data Consumption: A consumer script reads from the Kafka topic
Data Storage: The data is processed and stored in a Cassandra database
Data Query: We can query the stored data using CQL (Cassandra Query Language)

Azure Infrastructure Setup

This project leverages Microsoft Azure for hosting our components:

Azure Virtual Machine: Ubuntu 20.04 LTS server to run our Python scripts
Network Security Group: Configured to allow necessary traffic for Kafka and Cassandra
Azure Managed Disks: For persistent storage of Cassandra data

To set up an Azure VM for this project:

Log in to the Azure Portal
Create a new Virtual Machine with Ubuntu 20.04 LTS
Select at least 2 vCPUs and 8GB RAM for optimal performance
Configure networking to allow inbound SSH (port 22), Kafka (port 9092), and Cassandra (port 9042)
Create and download SSH keys for secure access

After provisioning, connect to your VM using SSH:

ssh -i /path/to/your/key.pem azureuser@your-vm-ip-address

Setting Up Confluent Cloud for Your Weather Data Pipeline

Before diving into the implementation, we'll set up a managed Kafka environment on Confluent Cloud.

Creating a Confluent Cloud Account

Visit Confluent.io and click on "Get Started Free"
Complete the registration process by providing your email and setting up a password
Verify your email address to activate your account

Setting Up a Kafka Cluster

Once logged in to Confluent Cloud:

Click on "Create cluster" in the dashboard
Choose Azure as your cloud provider and select the region closest to your Azure VM
Select the "Basic" cluster type for development purposes (you can upgrade later)
Name your cluster (e.g., "weather-pipeline-cluster")
Click "Launch cluster"

Creating a Kafka Topic

After your cluster is provisioned:

Select your cluster from the dashboard
Navigate to the "Topics" section in the left sidebar
Click "Create topic"
Enter "weather_data" as the topic name
Set the number of partitions (start with 6 for this example)
Leave the default retention settings (7 days)
Click "Create with defaults" (or customize advanced settings if needed)

Creating API Keys for Authentication

To connect your application to the Confluent Cloud Kafka cluster:

In your cluster dashboard, click on "API keys" in the left sidebar
Click "Create key"
Select "Global access" (or "Granular access" if you prefer more control)
Provide a description (e.g., "Weather Pipeline API Key")
Click "Create key"
Important: Save both the API key and secret in a secure location as they will only be shown once

Testing the Connection

To verify your connection settings, create a simple test script on your Azure VM:

from confluent_kafka.admin import AdminClient
import os
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Kafka configuration
kafka_config = {
    "bootstrap.servers": os.getenv('BOOTSTRAP_SERVERS'),
    "security.protocol": "SASL_SSL",
    "sasl.mechanisms": "PLAIN",
    "sasl.username": os.getenv('KAFKA_API_KEY'),
    "sasl.password": os.getenv('KAFKA_API_SECRET'),
}

# Create Admin client
admin_client = AdminClient(kafka_config)

# List topics
topics = admin_client.list_topics(timeout=10)
print("Available topics:", list(topics.topics.keys()))

Setting Up Cassandra on Azure VM

We'll install Apache Cassandra directly on our Azure VM:

# Install Java
sudo apt update
sudo apt install -y openjdk-8-jdk

# Add the Apache Cassandra repository
echo "deb https://downloads.apache.org/cassandra/debian 40x main" | sudo tee -a /etc/apt/sources.list.d/cassandra.list
curl https://downloads.apache.org/cassandra/KEYS | sudo apt-key add -
sudo apt update

# Install Cassandra
sudo apt install -y cassandra

# Start the service
sudo systemctl start cassandra
sudo systemctl enable cassandra

# Verify Cassandra is running
nodetool status

Configure Cassandra to accept remote connections by editing /etc/cassandra/cassandra.yaml:

sudo nano /etc/cassandra/cassandra.yaml

Change the following values:

listen_address: <your-vm-ip-address>
rpc_address: 0.0.0.0
seeds: "<your-vm-ip-address>"

Restart Cassandra to apply changes:

sudo systemctl restart cassandra

Setting Up the Environment

Now on your Azure VM, set up the development environment:

# Create a project directory
mkdir weather_data_pipeline
cd weather_data_pipeline

# Create and activate a virtual environment
python3 -m venv myvenv
source myvenv/bin/activate

Then install the required packages:

pip install confluent-kafka requests python-dotenv cassandra-driver

Create a .env file to store your environment variables:

WEATHER_API_KEY=your_openweathermap_api_key
BOOTSTRAP_SERVERS=your_confluent_bootstrap_servers
KAFKA_API_KEY=your_confluent_api_key
KAFKA_API_SECRET=your_confluent_api_secret
CASSANDRA_HOST=localhost

Creating the Producer Script

The producer script is responsible for fetching weather data from the OpenWeatherMap API and sending it to a Kafka topic. Let's break down the key components of our weather_producer.py script:

import json
import time
import os
import requests
from confluent_kafka import Producer
from dotenv import load_dotenv
import logging

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Load environment variables from .env file
load_dotenv()

# List of cities
cities = [
    "Nairobi",
    "Johannesburg",
    "Casablanca",
    "Lagos",
    "Kinshasa"
]

# OpenWeatherMap API setup
owm_api_key = os.getenv('WEATHER_API_KEY')
owm_base_url = "https://api.openweathermap.org/data/2.5/weather"

def fetch_weather_data(city):
    """Fetch weather data from OpenWeatherMap API using city name."""
    url = f"{owm_base_url}?q={city}&appid={owm_api_key}&units=metric"
    try:
        response = requests.get(url)
        response.raise_for_status()
        data = response.json()
        data["extracted_city"] = city
        return data
    except requests.exceptions.RequestException as e:
        logger.error(f"Error fetching data for {city}: {e}")
        return None

def delivery_report(err, msg):
    """Callback for Kafka message delivery status."""
    if err is not None:
        logger.error(f"Message delivery failed: {err}")
    else:
        logger.info(f"Message delivered to {msg.topic()} [{msg.partition()}] at offset {msg.offset()}")

# Kafka configuration
kafka_config = {
    "bootstrap.servers": os.getenv('BOOTSTRAP_SERVERS'),
    "security.protocol": "SASL_SSL",
    "sasl.mechanisms": "PLAIN",
    "sasl.username": os.getenv('KAFKA_API_KEY'),
    "sasl.password": os.getenv('KAFKA_API_SECRET'),
    "broker.address.family": "v4",
    "message.send.max.retries": 5,
    "retry.backoff.ms": 500,
}

producer = Producer(kafka_config)
topic = "weather_data"

def produce_weather_data():
    """Fetch weather data for each city and produce to Kafka."""
    for city in cities:
        data = fetch_weather_data(city)
        if data:
            producer.produce(topic, key=city, value=json.dumps(data), callback=delivery_report)
            producer.poll(0)
        else:
            logger.error(f"Failed to fetch data for {city}")
    producer.flush()

if __name__ == "__main__":
    produce_weather_data()
    logger.info("Data extraction and production complete")

Notable Components of the Producer Script:

Environment Setup: We import dotenv to load environment variables and initialize logging to track script execution.
City List: We define a list of African cities for which we want to collect weather data.
API Integration: The function fetch_weather_data() makes HTTP requests to the OpenWeatherMap API, asking for metric units of measurement for temperature.
Kafka Configuration: We set up a Kafka producer with security credentials and reliability properties.
Data Production: The produce_weather_data() method reads data for each city and produces it to the "weather_data" Kafka topic.
Delivery Reporting: The delivery_report() callback method prints if messages were delivered to Kafka successfully.

Creating the Consumer Script

Now, let's create the consumer script that will read data from the Kafka topic and store it in Cassandra. Here is our weather_consumer.py script:

import json
import uuid
import os
from confluent_kafka import Consumer, KafkaError
from cassandra.cluster import Cluster
from dotenv import load_dotenv
import logging

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()

# Kafka configuration
kafka_config = {
    "bootstrap.servers": os.getenv('BOOTSTRAP_SERVERS'),
    "group.id": "weather_consumer_group",
    "auto.offset.reset": "earliest",
    "security.protocol": "SASL_SSL",
    "sasl.mechanisms": "PLAIN",
    "sasl.username": os.getenv('KAFKA_API_KEY'),
    "sasl.password": os.getenv('KAFKA_API_SECRET'),
    "broker.address.family": "v4",
}

# Cassandra configuration
cassandra_host = os.getenv('CASSANDRA_HOST')

# Connect to Cassandra
cluster = Cluster([cassandra_host])
session = None

def initialize_cassandra():
    """Initialize Cassandra connection and create keyspace/table if needed."""
    global session
    try:
        session = cluster.connect()

        # Create keyspace if it doesn't exist
        session.execute("""
            CREATE KEYSPACE IF NOT EXISTS weather_data
            WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}
            AND durable_writes = true;
        """)

        # Use the keyspace
        session.execute("USE weather_data")

        # Create table if it doesn't exist
        session.execute("""
            CREATE TABLE IF NOT EXISTS simple_weather (
                id uuid PRIMARY KEY,
                city_name text,
                temperature float,
                timestamp timestamp,
                weather_description text,
                weather_main text
            );
        """)
        logger.info("Cassandra table ready")
        return True
    except Exception as e:
        logger.error(f"Cassandra initialization error: {e}")
        return False

def insert_weather_data(weather_data):
    """Insert weather data into Cassandra."""
    try:
        # Extract interesting fields
        city = weather_data["extracted_city"]
        temp = weather_data["main"]["temp"]
        timestamp = weather_data["dt"]
        weather_desc = weather_data["weather"][0]["description"]
        weather_main = weather_data["weather"][0]["main"]

        # Insert data into Cassandra
        query = """
            INSERT INTO simple_weather (id, city_name, temperature, timestamp, weather_description, weather_main)
            VALUES (%s, %s, %s, toTimestamp(now()), %s, %s)
        """
        session.execute(query, (uuid.uuid4(), city, temp, weather_desc, weather_main))
        logger.info(f"Inserted weather for {city} at {timestamp}")
        return True
    except Exception as e:
        logger.error(f"Error inserting data: {e}")
        return False

def consume_weather_data():
    """Consume weather data from Kafka and store in Cassandra."""
    # Initialize Cassandra
    if not initialize_cassandra():
        return

    # Create Kafka consumer
    consumer = Consumer(kafka_config)
    consumer.subscribe(["weather_data"])

    logger.info("Subscribed to topic: weather_data")

    try:
        while True:
            msg = consumer.poll(1.0)

            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                else:
                    logger.error(f"Consumer error: {msg.error()}")
                    break
            try:
                weather_data = json.loads(msg.value())
                insert_weather_data(weather_data)
            except Exception as e:
                logger.error(f"Error processing message: {e}")

    except KeyboardInterrupt:
        logger.info("Stopping consumer")
    finally:
        consumer.close()
        cluster.shutdown()

if __name__ == "__main__":
    consume_weather_data()

Key Components of the Consumer Script:

Kafka Consumer Configuration: We set up a consumer to consume from the "weather_data" topic, with the appropriate security settings.
Cassandra Connection: We connect to the Cassandra cluster directly without authentication credentials.
Database Initialization: The initialize_cassandra() function creates the necessary keyspace and table if they haven't already been created.
Data Processing: The insert_weather_data() function extracts pertinent fields from the weather data and inserts them into Cassandra.
Continuous Consumption: The consume_weather_data() function continuously polls the Kafka topic and processes any messages it finds.

Running the Pipeline on Azure

Now that we've written both the producer and consumer scripts, let's run them on our Azure VM to see our data pipeline in action.

First, make sure your virtual environment is activated and all dependencies are installed. Then, run the producer script:

python3 weather_producer.py

As shown in the first screenshot, the producer is successfully retrieving weather data and producing it to Kafka:

2025-04-07 05:56:47,013 - __main__ - INFO - Message delivered to weather_data [4] at offset 8
2025-04-07 05:56:47,013 - __main__ - INFO - Message delivered to weather_data [4] at offset 9
2025-04-07 05:56:47,035 - __main__ - INFO - Message delivered to weather_data [3] at offset 8
2025-04-07 05:56:47,035 - __main__ - INFO - Message delivered to weather_data [3] at offset 9
2025-04-07 05:56:47,072 - __main__ - INFO - Message delivered to weather_data [2] at offset 4
2025-04-07 05:56:47,072 - __main__ - INFO - Data extraction and production complete

Next, run the consumer script in a separate terminal:

python3 weather_consumer.py

The consumer subscribes to the Kafka topic and begins processing messages, as shown the second screenshot:

Subscribed to topic: weather_data
Cassandra table ready
Inserted weather for Johannesburg at 2025-04-07 05:56:45
Inserted weather for Lagos at 2025-04-07 05:55:29
Inserted weather for Nairobi at 2025-04-07 05:56:45
Inserted weather for Casablanca at 2025-04-07 05:53:05
Inserted weather for Kinshasa at 2025-04-07 05:56:46

Querying the Data

Now that the data is in Cassandra, we can query it using CQL. On your Azure VM, use the cqlsh tool:

cqlsh localhost

Then query to retrieve the weather data:

USE weather_data;
SELECT * FROM simple_weather;

As in your third screenshot, this pulls back all of the weather data for our cities:

id                                   | city_name     | temperature | timestamp                       | weather_description | weather_main
--------------------------------------+--------------+-------------+--------------------------------+---------------------+-------------
9781d8f2-e7d1-484f-a780-4df61ed7c7da | Johannesburg |       15.02 | 2025-04-07 06:22:41.000000+0000 |    overcast clouds  |      Clouds
f64ab11f-a732-4d47-84fb-f450d1e4a9bc |     Kinshasa |       21.21 | 2025-04-07 06:22:41.000000+0000 |         few clouds  |      Clouds
01840d74-0b10-461b-98b4-8eaf5fce2e0c |      Nairobi |       18.62 | 2025-04-07 06:22:41.000000+0000 |     broken clouds   |      Clouds
546d4df6-45b4-4956-94b1-b92b92bc1e84 |        Lagos |       26.57 | 2025-04-07 06:22:41.000000+0000 |    overcast clouds  |      Clouds
95d0dc97-a940-45a9-ab5d-28b9deba7b02 |   Casablanca |       14.07 | 2025-04-07 06:20:30.000000+0000 |   scattered clouds  |      Clouds
(5 rows)

Extending the Pipeline

This basic pipeline can be extended as follows:

Add More Cities: Expand the list of cities to get a broader geographic coverage.
Collect More Data Points: Modify the scripts to collect additional weather parameters like humidity, wind speed, and barometric pressure.
Data Aggregation: Add functionality to calculate averages, minimums, and maximums over time intervals of varying sizes.
Visualizations: Connect a visualization software like Grafana or Azure Data Explorer to your Cassandra database to create dashboards.
Alerts: Configure alerts for extreme weather conditions by adding thresholds for temperature, rainfall, etc.
Scale with Azure Kubernetes Service: For larger deployments, consider containerizing your applications and deploying them on AKS.

Common Issues and Troubleshooting

API Rate Limiting: The OpenWeatherMap API rate limits free tier accounts. If you're dealing with a large amount of cities, you might hit these limits. Consider using a paid tier or request throttling.
Kafka Connection Issues: Ensure that your Kafka credentials and bootstrap servers are correctly configured in the .env file.
Cassandra Connectivity: Ensure that your Cassandra instance is accessible on your Azure VM and that the firewall rules allow connections.
Azure VM Connectivity: Check that your Network Security Group allows the necessary inbound and outbound traffic.
Disk Space: Monitor your Azure VM's disk space, especially if storing large amounts of historical weather data.

Resources

Building an Automated Bitcoin Price ETL Pipeline with Airflow and PostgreSQL

Eric Katumo — Mon, 31 Mar 2025 19:07:56 +0000

Introduction

This article details creating an automated ETL (Extract, Transform, Load) pipeline that retrieves daily Bitcoin price data from the Polygon.io API, performs necessary transformations, and loads the data into a PostgreSQL database. The workflow is orchestrated using Apache Airflow, ensuring reliable daily execution.

This project demonstrates several key data engineering concepts:

API data extraction
Data transformation using pandas
Database integration with PostgreSQL
Workflow orchestration with Apache Airflow
Deployment to a cloud environment

System Architecture

The pipeline consists of the following components:

Data Source: Polygon.io API providing cryptocurrency price data
ETL Script: Python script that handles extraction, transformation, and loading
Database: PostgreSQL for data storage
Orchestration: Apache Airflow for scheduling and monitoring
Infrastructure: Cloud VM for hosting the pipeline

The system flows in a linear fashion: Airflow triggers the ETL script daily, which extracts the latest BTC prices, transforms the data into a suitable format, and loads it into the PostgreSQL database.

Detailed Implementation

Step 1: Creating the ETL Script

The first component is btc_prices.py, which handles the core ETL functionality:

import requests
import os
from sqlalchemy import create_engine
import pandas as pd
from datetime import datetime
from dotenv import load_dotenv

# Define API endpoint
url = 'https://api.polygon.io/v1/open-close/crypto/BTC/USD/2025-03-31?adjusted=true&apiKey=YOUR_API_KEY'
response = requests.get(url)

if response.status_code == 200:
    data = response.json()
    open_price = data.get('open')
    close_price = data.get('close')
    date = data.get('day')
    symbol = data.get('symbol')
else:
    print(f"Failed to retrieve data: {response.status}")
    exit()

# Prepare data for insertion
data_df = {
    'symbol': symbol,
    'open_price': open_price,
    'close_price': close_price,
    'date': date
}
df = pd.DataFrame(data_df, index=[0])
df['date'] = pd.to_datetime(df['date']).dt.strftime('%Y-%m-%d')

# Load environment variables
load_dotenv()
dbname = os.getenv('dbname')
user = os.getenv('user')
password = os.getenv('password')
host = os.getenv('host')
port = os.getenv('port')

# Create database connection
engine = create_engine(f'postgresql://{user}:{password}@{host}:{port}/{dbname}')

df.to_sql("crypto_prices", con=engine, if_exists="append", index=False, schema="dataengineering")
print(f"Successfully loaded crypto data for {df['date'][0]}")

This script:

Extracts Bitcoin price data from the Polygon.io API
Transforms and structures the data using pandas
Loads the data into PostgreSQL
Uses environment variables for secure database connection management

Step 2: Creating the Airflow DAG

Next, the btc_dag.py defines the Airflow DAG (Directed Acyclic Graph) that orchestrates the workflow:

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash_operator import BashOperator

# DAG default arguments
default_args = {
    "owner": "data_engineer",
    "depends_on_past": False,
    "start_date": datetime(2025, 3, 31),
    "email_on_failure": False,
    "email_on_retry": True,
    "retries": 2,
    "retry_delay": timedelta(minutes=2)
}

with DAG(
    'polygon_btc_data',
    default_args=default_args,
    schedule_interval='@daily',
) as dag:

    activate_venv = BashOperator(
        task_id='activate_virtual_env',
        bash_command='source /home/user/project/venv/bin/activate',
    )

    execute_file = BashOperator(
        task_id='execute_python_file',
        bash_command='python /home/user/project/btc_prices.py',
    )

    activate_venv >> execute_file

This DAG:

Defines the execution schedule
Activates the virtual environment
Executes the ETL script

Step 3: Setting Up the Environment

Creating a Virtual Environment:

   python -m venv venv
   source venv/bin/activate

Installing Dependencies:

   pip install requests pandas sqlalchemy python-dotenv psycopg2-binary apache-airflow

Setting Up Environment Variables:

   echo "dbname=your_database_name" >> .env
   echo "user=your_database_user" >> .env
   echo "password=your_database_password" >> .env
   echo "host=your_database_host" >> .env
   echo "port=your_database_port" >> .env

Step 4: Server Deployment

SSH into the cloud VM:

   ssh user@your_server_ip

Create necessary directories:

   mkdir -p ~/crypto_price
   mkdir -p ~/airflow/dags

Transfer scripts to the server:

   scp btc_prices.py user@your_server_ip:~/crypto_price/
   scp btc_dag.py user@your_server_ip:~/airflow/dags/

Step 5: PostgreSQL Configuration

Creating Database Schema:

   CREATE SCHEMA IF NOT EXISTS dataengineering;

   CREATE TABLE IF NOT EXISTS dataengineering.crypto_prices (
       id SERIAL PRIMARY KEY,
       symbol VARCHAR(10) NOT NULL,
       open_price NUMERIC(20, 8) NOT NULL,
       close_price NUMERIC(20, 8) NOT NULL,
       date DATE NOT NULL,
       created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
   );

Conclusion

The architecture follows best practices for data engineering:

Separation of extraction, transformation, and loading concerns
Secure credential management
Robust error handling
Automated scheduling
Cloud-based deployment

The combination of Python, Airflow, and PostgreSQL provides a powerful foundation for financial data analysis, enabling timely insights into cryptocurrency market trends.

Github

Apache Airflow for Data Engineering: Best Practices and Real-World Examples

Eric Katumo — Sun, 30 Mar 2025 17:44:57 +0000

Introduction

Apache Airflow, or simply Airflow, is an open-source tool and framework for running data pipelines in production environments. As an industry-leading workflow management platform, Airflow empowers data practitioners to define their data pipelines as code using Python, providing significant advantages over traditional scheduling tools by enabling version control, testing, and maintenance of complex workflows.

Airflow enhances data pipeline management by offering robust scheduling capabilities, comprehensive monitoring, and detailed observability of pipeline performance. It serves as a centralized hub for all data workflows, whether you're preparing training data for machine learning models, transforming data for analytics, or persisting information in a data lake, with features designed for reliability and scalability.

The core strength of Airflow lies in its ability to represent workflows as directed acyclic graphs (DAGs), explicitly defining dependencies between tasks and providing a clear visual representation of complex processes. This paradigm, combined with Airflow's extensible architecture, makes it uniquely suited to orchestrate the increasingly complex data needs of modern organizations.

Key Airflow Concepts and Terminology

Core Concepts

DAG (Directed Acyclic Graph): A DAG is the backbone of Airflow, structuring a workflow as a set of tasks with defined dependencies, ensuring no cycles exist. Written as Python code, it’s stored in the DAGs folder and dictates the sequence and relationships of operations.
Operator: Operators are the building blocks of tasks, providing predefined templates for specific actions like running a script or sending an email. Popular examples include:
- BashOperator: Runs shell commands or scripts, perfect for quick command-line operations, though it depends on exit codes to determine success.
- PythonOperator: Executes Python functions, offering flexibility for custom logic like processing data or querying APIs, with access to runtime context.
- EmailOperator: Sends email notifications through an SMTP server, useful for alerting teams, though limited to basic messaging features.
Task: A task is a single unit of work in a DAG, created by applying an operator with specific settings, such as running a bash command or a Python function. It represents an actionable step within the broader workflow.
Task Instance: A task instance is a specific execution of a task, tied to a particular DAG run and timestamp, reflecting its state (e.g., running, succeeded, failed) at that moment.

Scheduling and Execution Concepts

Schedule Interval: This determines how frequently a DAG is triggered, set using cron-like expressions (e.g., 0 0 * * * for daily at midnight) or predefined intervals. It governs the timing of automated runs.
Execution Date: The execution date marks the logical start time of a DAG run, often aligned with the schedule interval, though the actual run may occur later due to processing delays.
DAG Run: A DAG run is a single instance of a DAG’s execution, initiated manually or by the scheduler, tied to a specific execution date and encompassing all task instances for that run.
Backfill: Backfilling involves running a DAG for past periods, using historical execution dates to catch up on missed or retroactive data processing.

Airflow Architecture Components

Scheduler: The scheduler is the engine that monitors DAGs, triggers runs based on their schedule intervals, and assigns tasks to executors for processing. It ensures the timely and orderly execution of workflows.
Executor: Executors manage how tasks are run, ranging from simple single-threaded execution to distributed setups. Common types include:
- SequentialExecutor: Runs tasks one-by-one, suitable for testing but not production.
- LocalExecutor: Handles multiple tasks concurrently on one machine, balancing simplicity and efficiency.
- CeleryExecutor: Distributes tasks across worker nodes for scalability, leveraging a message queue.
- KubernetesExecutor: Launches tasks as individual Kubernetes pods, ideal for dynamic, cloud-native environments.
Web Server: The web server powers Airflow’s user interface, offering a visual dashboard to monitor DAG runs, task statuses, and logs. It’s the primary tool for interacting with and overseeing workflows.
Metadata Database: This database stores all operational data, including DAG definitions, task states, and run histories, serving as the central repository for Airflow’s state management.
Worker: Workers are processes that execute tasks, typically used with distributed executors like CeleryExecutor, pulling tasks from a queue to process them.

Advanced Concepts

XComs (Cross-Communications): XComs enable tasks to share small pieces of data, like a file path or a result, by pushing and pulling values during execution. They’re handy for lightweight data exchange but not suited for large datasets.
Sensor: Sensors are specialized operators that pause execution until a condition is met, such as a file appearing or an external process completing. They’re useful for waiting on dependencies outside the DAG’s control.

Writing DAG Definitions

Let’s explore how to define a DAG in Airflow with a practical example suitable for beginners transitioning to more advanced concepts. This ETL pipeline fetches user data from an API, processes it, and loads it into a database, all written as Python code in a file placed in Airflow’s DAGs folder:

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
import pandas as pd
import requests
from sqlalchemy import create_engine

# Set basic settings for the DAG
default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "start_date": datetime(2024, 1, 1),
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

# Define the DAG (workflow)
with DAG(
    "advanced_etl_pipeline",
    default_args=default_args,
    schedule_interval="@daily",  # Runs once a day
) as dag:

    # Step 1: Extract data from an API
    def extract():
        url = "https://jsonplaceholder.typicode.com/users"  # Sample API
        response = requests.get(url)
        data = response.json()
        df = pd.DataFrame(data)
        df.to_csv("/tmp/extracted_data.csv", index=False)

    # Step 2: Transform the data
    def transform():
        df = pd.read_csv("/tmp/extracted_data.csv")
        df = df[['id', 'name', 'email']]  # Keep only these columns
        df.columns = ['user_id', 'full_name', 'email_address']  # Rename columns
        df.to_csv("/tmp/transformed_data.csv", index=False)

    # Step 3: Load data into a database
    def load():
        engine = create_engine("postgresql://username:password@localhost:5432/etl_db")
        df = pd.read_csv("/tmp/transformed_data.csv")
        df.to_sql("customers", engine, if_exists="append", index=False)

    # Define tasks
    extract_task = PythonOperator(task_id="extract", python_callable=extract)
    transform_task = PythonOperator(task_id="transform", python_callable=transform)
    load_task = PythonOperator(task_id="load", python_callable=load)

    # Set the order: extract -> transform -> load
    extract_task >> transform_task >> load_task

This example demonstrates a realistic ETL process:

Extract: Fetches user data from a public API and saves it as a CSV.
Transform: Filters and renames columns, then saves the result.
Load: Writes the data to a PostgreSQL database (update the connection string with your credentials).

Here’s how it’s built:

Imports: Includes necessary libraries (pandas, requests, sqlalchemy) alongside Airflow components.
Default Arguments: Sets basic retry logic and a start date, keeping it minimal but functional.
DAG Definition: Uses a with statement for cleaner syntax, scheduling it daily.
Tasks: Three PythonOperator tasks call the defined functions, linked sequentially with >>.

For beginners, start by running this in Airflow, replacing the database connection with your own, and observing the flow in the UI.

Best Practices for Apache Airflow DAGs

1. Creating DAGs

Make Tasks Reusable and Modular: Design tasks as standalone units with clear inputs and outputs, avoiding tight coupling to enable reuse across DAGs. Encapsulate logic using Python functions or external scripts, keeping DAG files focused on workflow definition.
Use Meaningful IDs and Descriptions: Assign descriptive dag_id and task_id values (e.g., extract_data instead of task1) and add a description to clarify purpose, improving readability and maintenance.
Avoid Top-Level Code Execution: Place logic inside tasks or functions, not at the DAG file’s top level, to prevent unnecessary execution during parsing by the scheduler. For example, avoid API calls outside task definitions to reduce overhead.

2. Testing and Validation

Test DAG Loading: Regularly validate DAG syntax and structure by running airflow dags list or python your_dag_file.py to catch errors early. This ensures the DAG can be parsed without runtime issues.
Unit Test Task Logic: Write unit tests for Python functions used in PythonOperator tasks using frameworks like pytest, isolating business logic from Airflow execution. Mock external dependencies to focus on code behavior.
Simulate DAG Runs: Use airflow tasks test <dag_id> <task_id> <execution_date> to dry-run tasks and verify behavior without affecting the metadata database. This helps debug task execution locally.

3. Workflow Design

Ensure Idempotency: Design tasks to produce consistent results regardless of how often they’re run with the same inputs, simplifying retries and backfills. For example, use INSERT ... ON CONFLICT in SQL tasks to handle duplicates.
Minimize Task Dependencies: Define only necessary dependencies to maximize parallelism and reduce complexity, using >> or set_upstream explicitly. Avoid over-sequencing tasks that can run independently.
Leverage SubDAGs or TaskGroups: Organize complex workflows into SubDAGs (for reusable segments) or TaskGroups (for visual clarity), but prefer TaskGroups in newer versions as SubDAGs can complicate scheduling.
Pass Data Efficiently: Use XComs for small data exchanges between tasks, but store larger datasets in external systems (e.g., S3, databases) and pass references (e.g., file paths) instead. This prevents memory overload in the metadata database.

4. Operational Considerations

Set Appropriate Retries: Configure retries and retry_delay in default_args to handle transient failures (e.g., network issues), balancing resilience with resource use. For example, retries=3 with retry_delay=timedelta(minutes=5) is a common starting point.
Use Pools for Resource Limits: Define pools to cap concurrent task execution for resource-intensive operations (e.g., database queries), preventing system overload. Set this via the Airflow UI or programmatically with pool='my_pool'.
Secure Sensitive Data: Store credentials and secrets in Airflow Connections or Variables (with encryption enabled) rather than hardcoding them in DAG files. Access them using conn_id or Variable.get() for security and flexibility.
Control Backfills with Catchup: Set catchup=False unless backfilling is intentional, and choose a logical start_date to avoid unexpected historical runs when deploying new DAGs. This prevents scheduler overload on first deployment.
Monitor and Log: Enable task logging and set up alerts (e.g., via EmailOperator or custom callbacks) to track failures and performance, using the Airflow UI to inspect logs. Add custom logging in tasks to aid debugging.

Real-World Examples

Apache Airflow’s versatility makes it a valuable tool across industries, orchestrating complex data workflows to enhance efficiency and decision-making. Below are examples of its application in financial services, retail and e-commerce, and healthcare.

Financial Services

In financial services, Apache Airflow streamlines critical operations by automating data workflows with precision and scalability. It supports regulatory reporting to comply with laws like Dodd-Frank and GDPR, enhances fraud detection by curating data for machine learning models to identify anomalies, and aggregates data from multiple sources for a comprehensive customer view, while also aiding risk management and portfolio optimization.

Retail and E-Commerce

In retail and e-commerce, Airflow boosts operational efficiency and customer satisfaction by managing data pipelines that integrate diverse sources. It automates inventory optimization, powers personalized marketing through customer behavior analysis, and improves supply chain coordination by syncing data across vendors, logistics, and sales for accurate demand forecasting and timely deliveries.

Healthcare

In healthcare, Apache Airflow enhances patient care and operational efficiency by orchestrating complex medical data workflows. It automates the processing of electronic health record (EHR) data with compliance tracking, supports diagnostic processes by managing machine learning model training for medical imagery, and gathers actionable insights from patient data to improve outcomes and reduce manual effort.

Amazon Redshift Data Warehousing (Sample Project)

Eric Katumo — Wed, 26 Mar 2025 15:32:33 +0000

Overview

This project focuses on analyzing two key business processes using Amazon Redshift:

Identifying the most common frequency of purchases per season.
Finding products with high review ratings.

Step 1: Setting Up Amazon Redshift

Create a Redshift Cluster
- Navigate to AWS Redshift Console.
- Click Create Cluster.
- Choose a single-node cluster.
- Select an appropriate node type (e.g., dc2.large).
- Set the database name, username, and password.
- Click Create cluster.
Create an IAM Role
- Go to the IAM Console.
- Create a new role with AmazonS3ReadOnlyAccess.
- Attach the role to your Redshift cluster.
Create an S3 Bucket and Upload Data
- Navigate to Amazon S3.
- Create a new bucket (e.g., zzetu).
- Upload shopping_data.csv to this bucket.

Step 2: Loading Data into Redshift

Use the Amazon Redshift Query Editor to run SQL commands.
Create a table to store raw data:

CREATE TABLE shopping (
    CustomerID INTEGER,
    Age INTEGER,
    Gender VARCHAR(50),
    Category VARCHAR(50),
    Location VARCHAR(50),
    Season VARCHAR(50),
    ReviewRating REAL,
    SubscriptionStatus VARCHAR(50),
    PaymentMethod VARCHAR(50),
    ShippingType VARCHAR(50),
    DiscountApplied VARCHAR(50),
    PromoCodeUsed VARCHAR(50),
    PreviousPurchases INTEGER,
    PreferredPaymentMethod VARCHAR(50),
    FrequencyOfPurchases VARCHAR(50)
);

Copy data from S3 into Redshift:

COPY shopping
FROM 's3://zzetu/shopping_data.csv'
IAM_ROLE 'arn:aws:iam::your-account-id:role/your-redshift-role'
FORMAT AS CSV
IGNOREHEADER 1;

Step 3: Schema Design

We are using a star schema with 1 fact table and 3 dimension tables:

1. Dimension Tables

CREATE TABLE dimCustomer (
    CustomerID INTEGER PRIMARY KEY,
    Age INTEGER,
    Gender VARCHAR(50),
    Location VARCHAR(50),
    SubscriptionStatus VARCHAR(50),
    PreferredPaymentMethod VARCHAR(50)
);

CREATE TABLE dimProduct (
    Category VARCHAR(50) PRIMARY KEY,
    DiscountApplied VARCHAR(50),
    PromoCodeUsed VARCHAR(50)
);

CREATE TABLE dimTransaction (
    TransactionID INTEGER IDENTITY(1,1) PRIMARY KEY,
    CustomerID INTEGER,
    PaymentMethod VARCHAR(50),
    ShippingType VARCHAR(50),
    Season VARCHAR(50),
    FrequencyOfPurchases VARCHAR(50),
    FOREIGN KEY (CustomerID) REFERENCES dimCustomer(CustomerID)
);

2. Fact Table

CREATE TABLE factPurchases (
    CustomerID INTEGER,
    Age INTEGER,
    ReviewRating REAL,
    PreviousPurchases INTEGER,
    FOREIGN KEY (CustomerID) REFERENCES dimCustomer(CustomerID)
);

Step 4: Data Insertion

-- Insert data into dimCustomer
INSERT INTO dimCustomer (CustomerID, Age, Gender, Location, SubscriptionStatus, PreferredPaymentMethod)
SELECT DISTINCT CustomerID, Age, Gender, Location, SubscriptionStatus, PreferredPaymentMethod FROM shopping;

-- Insert data into dimProduct
INSERT INTO dimProduct (Category, DiscountApplied, PromoCodeUsed)
SELECT DISTINCT Category, DiscountApplied, PromoCodeUsed FROM shopping;

-- Insert data into dimTransaction
INSERT INTO dimTransaction (CustomerID, PaymentMethod, ShippingType, Season, FrequencyOfPurchases)
SELECT CustomerID, PaymentMethod, ShippingType, Season, FrequencyOfPurchases FROM shopping;

-- Insert data into factPurchases
INSERT INTO factPurchases (CustomerID, Age, ReviewRating, PreviousPurchases)
SELECT CustomerID, Age, ReviewRating, PreviousPurchases FROM shopping;

Step 5: Business Queries

1. Most Used Frequency of Purchase per Season

SELECT t.Season, t.FrequencyOfPurchases, COUNT(*) AS PurchaseCount
FROM factPurchases f
JOIN dimTransaction t ON f.CustomerID = t.CustomerID
GROUP BY t.Season, t.FrequencyOfPurchases
ORDER BY t.Season, PurchaseCount DESC;

2. Products with High Review Ratings

SELECT Category, AVG(f.ReviewRating) AS AvgRating
FROM fact_Purchases f
JOIN shopping s ON f.CustomerID = s.CustomerID
GROUP BY Category
ORDER BY AvgRating DESC;

Setup Guide

Create Redshift Cluster and configure IAM roles.
Load Data from S3 into Redshift tables.
Run the SQL scripts for creating tables and inserting data.
Execute business queries to generate insights.

Repository Link

GitHub Repository

The Ultimate Guide to Apache Kafka: Basics, Architecture, and Core Concepts

Eric Katumo — Mon, 10 Mar 2025 12:16:48 +0000

Introduction

Apache Kafka is a widely-used open-source platform for distributed event streaming, supporting high-performance data pipelines, streaming analytics, data integration, and mission-critical applications across thousands of companies https://kafka.apache.org/

Originally developed by LinkedIn, Kafka is renowned for its high throughput, scalability and durability. It enables real-time data processing and is a key component in modern event-driven architectures.

Kafka Architecture

Brokers

A single Kafka server is called a Kafka Broker. Each broker operates as an independent process on a distinct machine, communicating with other brokers via a reliable and high-speed network.

There can be any number of brokers but 3 are typically shown as a minimum. This allows one out for maintenance and one to fail at the same time.

Producers

A Producer is a client application that publishes (writes) events to a Kafka cluster. The Kafka producer is responsible for creating messages of the appropriate structure and sending them using the Kafka protocol. It has several configuration options to control message creation and delivery.

Consumers

The Kafka consumer works by issuing “fetch” requests to the brokers leading the partitions it wants to consume. The consumer receives back a chunk of log that contains all of the messages in that topic beginning from the offset position.

A consumer group is a set of consumers that cooperate to consume data from some topics

Topic

Kafka topics are the categories used to organize messages. Topics are handled by Kafka as independent queues. This means that a consumer can subscribe to a specific topic and only receive messages marked with that topic.

In Kafka, topics are partitioned and replicated across brokers throughout the implementation. Brokers refer to each of the nodes in a Kafka cluster

Cluster

Kafka clusters are a group of interconnected Kafka brokers that work together to manage the data streams entering and leaving a Kafka system. As user activity increases, so does the need for additional Kafka brokers to cope with the volume and velocity of the incoming data streams.

Kafka clusters enable the replication of data partitions across multiple brokers, ensuring high availability even in the case of node failures. Your data pipeline remains robust and responsive to fluctuating demand.

Partitions

Partitions are essential components within Kafka's distributed architecture that enable Kafka to scale horizontally, allowing for efficient parallel data processing. They are the building blocks for organizing and distributing data across the Kafka cluster.

Each partition can have multiple replicas spread across different brokers, guaranteeing fault tolerance and data redundancy.

KRaft

The consensus protocol that was introduced in KIP-500 to remove Apache Kafka’s dependency on ZooKeeper for metadata management. It leverages the Raft consensus algorithm to manage metadata and handle leader election natively within Kafka.

This eliminates the dependency on an external coordination system, allowing Kafka to function as a self-contained system.

Core Concepts

Events

A digital logbook that keeps track of everything important happening in your system. This logbook is filled with "events" – like a record of each significant action or change. In Kafka, these events are the core of how data is stored and shared.

Think of an event as a single entry in this logbook, with a few key pieces of information:

Key: A unique identifier for the event. This helps you categorize or group related events.
Value: The actual details of what happened.
Timestamp: When the event occurred.
Headers (optional): Extra bits of information about the event.

Replication

Replication is the process of having multiple copies of the data for the sole purpose of availability in case one of the brokers goes down and is unavailable to serve the requests. Copies of the partition are maintained at multiple broker instances using the partition’s write-ahead log.

The write-ahead log is where all the messages for that partition are stored in order. The messages are identified by the unique offset.

Offsets

The consumer offset is a way of tracking the sequential order in which messages are received by Kafka topics. Keeping track of the offset, or position, is important for nearly all Kafka use cases and can be an absolute necessity in certain instances, such as financial services.

The Kafka consumer offset allows processing to continue from where it last left off if the stream application is turned off or if there is an unexpected failure. In other words, by having the offsets persist in a data store, data continuity is retained even when the stream application shuts down or fails.

Consumer Groups

Consumer groups allow Kafka consumers to work together and process events from a topic in parallel. Each topic consists of one or more partitions. When a new consumer is started it will join a consumer group (this happens under the hood) and Kafka will then ensure that each partition is consumed by only one consumer from that group.

So, if you have a topic with two partitions and only one consumer in a group, that consumer would consume records from both partitions. After another consumer joins the same group, each consumer would continue consuming only one partition.

Retention

Kafka retention provides the ability to control the size of the Topic logs and avoid outgrowing the existing disk size. The retention can be configured or controlled based on the size of the logs (log retention) or based on the configured duration (Time Based Retention).

Also, the same retention can be set across all the Kafka topics or it can be configured per topic, depending on the nature of the topic we can set the retention accordingly.

Conclusion

Apache Kafka is a powerful distributed streaming platform that combines messaging and storage capabilities. Its architecture, featuring brokers, topics, and partitions, delivers scalability and fault tolerance. Kafka's core concepts, such as consumer groups and offsets, enable efficient and reliable stream processing.

With its ability to handle high-volume data streams and support real-time applications, Kafka is a crucial component of modern data architectures. It empowers developers to build robust, data-driven applications that address the challenges of today's data-intensive world. To get started with Apache Kafka visit https://kafka.apache.org/quickstart