DEV Community: Rugved

Building a Scalable Reddit Data Pipeline with Apache Airflow 3.1.0

Rugved — Thu, 23 Oct 2025 15:51:15 +0000

A modern Apache Airflow 3.1.0 template for scraping and analyzing Reddit trending posts with PostgreSQL storage, built with Docker Compose for easy deployment.

🚀 Features

This Reddit trending posts pipeline offers:

Airflow 3.1.0 with TaskFlow API and modern architecture
Reddit API Integration for scraping trending posts from multiple subreddits
PostgreSQL Database with comprehensive schema for post metadata
Docker Compose setup for easy deployment and scaling
Celery Executor for distributed task processing
Redis for task queue management
PgAdmin for database administration
Flower for Celery monitoring
Rate Limiting and error handling
Configurable Subreddits via environment variables

🏗️ Architecture Overview

🛠️ Tech Stack

Apache Airflow 3.1.0: Modern workflow orchestration
PostgreSQL: Robust data storage
Docker Compose: Container orchestration
Celery Executor: Distributed task processing
Redis: Message broker
Python 3.11+: Core programming language

📊 Pipeline Capabilities

DAG Structure

The reddit_trending_pipeline DAG includes:

get_config: Retrieves configuration and timezone settings
get_subreddits: Reads subreddit list from environment variables
fetch_subreddit: Scrapes posts from each subreddit (parallel execution)
combine_results: Merges results from all subreddits
persist_and_output: Saves data to PostgreSQL and outputs JSON

Key Features

@dag(
    dag_id="reddit_trending_pipeline",
    description="Scrape Reddit posts from last 24 hours and save to PostgreSQL",
    start_date=datetime.datetime(2025, 10, 1),
    schedule="0 6 * * *",  # every day at 6 AM
    catchup=False,
    default_args={
        "owner": "airflow",
        "retries": 0,
        "retry_delay": timedelta(minutes=5),
    },
    tags=["reddit", "scraping", "taskflow"],
)
def reddit_trending_pipeline():
    # Task implementations...

Smart Data Filtering

The pipeline includes intelligent filtering:

Time-based: Only posts from the last 24 hours
Quality-based: Minimum 100 upvotes threshold
Rate limiting: 2-second delay between requests
Media handling: Extracts images, videos, and gallery URLs

# Only last 24 hours
if created_utc <= args["cutoff_ts"]:
    continue
if int(p.get('ups', 0)) < 100:
    continue

🗄️ Database Schema

The pipeline creates a comprehensive reddit_posts table with the following fields:

CREATE TABLE reddit_posts (
    id SERIAL PRIMARY KEY,
    post_id VARCHAR(32) UNIQUE NOT NULL,
    subreddit VARCHAR(100) NOT NULL,
    title TEXT,
    description TEXT,
    upvotes INT,
    downvotes INT,
    score INT,
    url TEXT,
    author VARCHAR(255),
    created_at TIMESTAMP WITH TIME ZONE,
    num_comments INT,
    over_18 BOOLEAN,
    media_urls TEXT,
    upvote_ratio REAL,
    scraped_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

🚀 Getting Started

Prerequisites

Docker and Docker Compose
At least 4GB RAM
At least 2 CPU cores
10GB free disk space

Option A: Use This Template

git clone https://github.com/rugvedp/apache-airflow-3-template.git
cd apache-airflow-3-template

Option B: Start Fresh with Official Template

If you want to start from the official Airflow 3.1.0 template:

# Download the official Airflow 3.1.0 docker-compose.yaml
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/3.1.0/docker-compose.yaml'

# Create required directories
mkdir -p dags logs plugins config

2. Initialize Environment

make env

This creates the required directories and .env file with your user ID.

3. Initialize Airflow

make init

This initializes the Airflow database and creates the default admin user (username: airflow, password: airflow).

4. Start Services

make upd

This starts all services in detached mode.

5. Access the Web Interface

Airflow UI: http://localhost:8080 (airflow/airflow)
PgAdmin: http://localhost:5050 (admin@admin.com/admin)
Flower: http://localhost:5555 (optional, for Celery monitoring)

Airflow Dashboard

DAG Execution View

⚙️ Configuration

Environment Variables

Create a .env file or set these environment variables:

# Required
AIRFLOW_UID=50000

# Optional Reddit Configuration
REDDIT_SUBREDDITS='["MusicIndia", "IndiaTech", "worldnews", "programming"]'
REDDIT_REQUEST_DELAY_S=2
REDDIT_DB_CONN=postgresql://airflow:airflow@postgres:5432/airflow

# Optional Airflow Configuration
_AIRFLOW_WWW_USER_USERNAME=airflow
_AIRFLOW_WWW_USER_PASSWORD=airflow

DAG Configuration

The DAG runs daily at 6 AM (Asia/Kolkata timezone) and:

Scrapes posts from the last 24 hours
Filters posts with at least 100 upvotes
Includes rate limiting (2-second delay between requests)
Handles media URLs (images, videos, galleries)
Supports conflict resolution for duplicate posts

🎯 Usage

Manual DAG Execution

Access the Airflow UI at http://localhost:8080
Navigate to the reddit_trending_pipeline DAG
Click "Trigger DAG" to run manually
Monitor task execution in the Graph or Tree view

Customizing Subreddits

Set the REDDIT_SUBREDDITS environment variable:

export REDDIT_SUBREDDITS='["MusicIndia", "IndiaTech", "worldnews", "programming", "MachineLearning"]'

Database Queries

Connect to PostgreSQL via PgAdmin or directly:

-- View recent posts
SELECT subreddit, title, upvotes, created_at 
FROM reddit_posts 
ORDER BY created_at DESC 
LIMIT 10;

-- Top posts by subreddit
SELECT subreddit, COUNT(*) as post_count, AVG(upvotes) as avg_upvotes
FROM reddit_posts 
GROUP BY subreddit 
ORDER BY post_count DESC;

-- Trending posts (high upvote ratio)
SELECT title, subreddit, upvotes, upvote_ratio
FROM reddit_posts 
WHERE upvote_ratio > 0.8 
ORDER BY upvotes DESC;

🛠️ Development

Makefile Commands

make help          # Show all available commands
make env           # Create directories and .env file
make init          # Initialize Airflow database
make up            # Start services (foreground)
make upd           # Start services (detached)
make down          # Stop services
make restart       # Restart all services
make clean         # Remove containers, images, and volumes
make shell         # Open bash shell in worker container
make cli ARGS="..." # Run Airflow CLI commands

Adding New DAGs

Place your DAG file in the dags/ directory
The DAG will be automatically detected by the DAG processor
Use the TaskFlow API for modern Python DAGs:

from airflow.decorators import dag, task
from datetime import datetime

@dag(
    dag_id="my_custom_dag",
    start_date=datetime(2025, 1, 1),
    schedule="@daily",
    catchup=False,
)
def my_dag():
    @task
    def my_task():
        return "Hello World"

    my_task()

dag = my_dag()

📈 Monitoring

Airflow UI

DAGs: Overview of all DAGs and their status
Graph: Visual representation of task dependencies
Tree: Timeline view of task executions
Gantt: Task duration analysis

Flower (Celery Monitoring)

Access at http://localhost:5555
Monitor worker status and task queues
View task execution statistics

PgAdmin

Access at http://localhost:5050
Database administration and query interface
Monitor database performance

📚 Additional Resources

🎉 Conclusion

This Reddit trending posts pipeline demonstrates the power of modern Apache Airflow 3.1.0 with Docker Compose. It provides:

Scalable: Handle multiple subreddits simultaneously
Reliable: Robust error handling and monitoring
Maintainable: Clean, readable code with TaskFlow API
Observable: Comprehensive monitoring and logging

Ready to build your own data pipeline? Clone the repository and start scraping Reddit data today!

git clone https://github.com/rugvedp/apache-airflow-3-template.git
cd apache-airflow-3-template

Have questions about building data pipelines with Airflow? Drop a comment below! 🚀

Building a Hindi-Speaking AI Career Counseling Mentor with Twilio and Groq's LLaMA Models

Rugved — Mon, 10 Feb 2025 17:53:32 +0000

In this post, I'll walk you through how to create a voice-based AI career counseling mentor that speaks Hindi using Twilio for voice calls and Groq's LLaMA models for generating AI responses. This bot will act as a friendly and knowledgeable mentor, guiding students through career-related queries in Hindi. We'll also use ngrok to expose our local Flask server to the internet.

Overview of the Tech Stack

Twilio: Handles voice calls and speech recognition.
Groq's LLaMA Models: Powers the AI mentor with natural language understanding and response generation.
Flask: A lightweight Python web framework to handle HTTP requests and responses.
ngrok: Exposes your local Flask server to the internet so Twilio can interact with it.

How It Works

Incoming Call: Twilio forwards the call to your Flask app's /voice endpoint.
Speech Recognition: The app uses Twilio's Gather to capture the user's speech.
AI Response: The app sends the speech text to Groq's LLaMA model, which generates a response in Hindi.
Text-to-Speech: Twilio's Polly.Aditi converts Hindi text into speech and plays it back.
Continuous Conversation: The bot continues the conversation by prompting the user for more input.

Step-by-Step Implementation

1. Set Up Your Environment

Before diving into the code, ensure you have the following:

Python 3.10+ installed.
A Twilio account with a phone number capable of making voice calls.
A Groq API key (sign up at Groq).
Install the required Python libraries:

pip install flask twilio groq

2. Write the Code

a. Initialize Flask and Dependencies

from flask import Flask, request, Response, send_from_directory
from twilio.twiml.voice_response import VoiceResponse, Gather
from twilio.rest import Client
from groq import Groq
import os

# Initialize Flask
app = Flask(__name__)

# Twilio Credentials
TWILIO_ACCOUNT_SID = os.getenv('TWILIO_ACCOUNT_SID')
TWILIO_AUTH_TOKEN = os.getenv('TWILIO_AUTH_TOKEN')
TWILIO_PHONE_NUMBER = os.getenv('TWILIO_PHONE_NUMBER')
MY_PHONE_NUMBER = os.getenv('MY_PHONE_NUMBER')

twilio_client = Client(TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN)

# Initialize Groq client
client = Groq(api_key=os.getenv('GROQ_API_KEY'))  # Add your API key here
ngrok_url = 'https://your-ngrok-url.ngrok.io'  # Replace with your ngrok URL

messages = [
    {
        "role": "system",
        "content": """You are a highly knowledgeable and professional mentor female voice assistant who can solve any query regarding to career choice. Provide insightful, personalized, and practical guidance to students regarding their career and academic path. Focus on delivering clear, concise, expert advice using natural, conversational language suitable for Hindi translation and speech.  Be encouraging and supportive. Keep responses short and to the point. Start with the greeting and flow should be human-like. Behave like you are his friend and guide."""
    }
]

b. Handle Incoming Voice Calls

@app.route("/voice", methods=['POST'])
def voice():
    """Handles incoming voice calls from Twilio."""
    response = VoiceResponse()
    response.say("Namaste, aap kaise hai?", voice="Polly.Aditi", language="hi-IN")
    gather = Gather(input="speech", action="/process_voice", method="POST", language="hi-IN", speechTimeout="auto")
    gather.say("कृपया अपनी समस्या बताएं।", voice="Polly.Aditi", language="hi-IN")
    response.append(gather)
    return Response(str(response), content_type='text/xml')

This function greets the user in Hindi and prompts them to describe their career-related query using Twilio's Gather verb for speech input.

c. Process User Speech and Generate AI Response

@app.route("/process_voice", methods=['POST'])
def process_voice():
    """Processes user Hindi speech input, generates a response, and returns a spoken reply."""
    response = VoiceResponse()
    speech_text = request.form.get("SpeechResult", "")

    if not speech_text:
        response.say("मुझे आपकी आवाज़ सुनाई नहीं दी, कृपया फिर से बोलें।", voice="Polly.Aditi", language="hi-IN")
        response.redirect("/voice")  # Retry input
        return Response(str(response), content_type='text/xml')

    print(f"User said: {speech_text}")

    # Generate AI response using Groq
    ai_response = generate_response(speech_text)
    print(f"AI response: {ai_response}")

    # Stream the response back to the user
    response.say(ai_response, voice="Polly.Aditi", language="hi-IN")

    # Continue the conversation
    gather = Gather(input="speech", action="/process_voice", method="POST", language="hi-IN", speechTimeout="auto")
    response.append(gather)
    return Response(str(response), content_type='text/xml')

d. Generate AI Response Using Groq

def generate_response(prompt):
    """Uses Groq to generate AI responses in Hindi based on conversation context."""
    messages.append({"role": "user", "content": prompt})

    try:
        ai_response = client.chat.completions.create(
            model="llama-3.3-70b-versatile",  # Replace with your preferred Groq model
            messages=messages,
            max_tokens=32768,
            temperature=0.5,
            n=1
        )
        res = ai_response.choices[0].message.content.strip()
        messages.append({"role": "assistant", "content": res})
        return res
    except Exception as e:
        print(f"Error generating response: {e}")
        return "मुझे क्षमा करें, मैं आपकी सहायता नहीं कर सकता।"

3. Set Up ngrok

Run:

ngrok http 5000

Copy the HTTPS URL provided by ngrok and update ngrok_url in the Flask app.

4. Configure Twilio

Log in to your Twilio console.
Go to Phone Numbers > Manage > Active Numbers and select your Twilio phone number.
Under the Voice & Fax section, set the Webhook for incoming calls to https://your-ngrok-url.ngrok.io/voice.

Conclusion

By combining Twilio and Groq's LLaMA models, you can create a powerful Hindi-speaking AI career counseling mentor. This project demonstrates the potential of AI in providing personalized guidance in regional languages, making it accessible to a wider audience.

Let me know if you have any questions or suggestions!

GitHub Profile

LinkedIn Profile