Build a Reddit Data Pipeline with Python and PostgreSQL

Data engineering teams often need to ingest public community discussions for sentiment analysis, brand monitoring, and market research. Building and maintaining custom scrapers against Reddit's layout is an expensive operational burden due to frequent structural updates and aggressive IP blocking.

This guide details how to build a production-ready data pipeline that uses the Apify platform and the reddit-scraper Actor to extract clean, structured post and comment records. We will walk through configuring the input JSON, executing the extraction via the Apify Python SDK, handling platform-specific limitations, and mapping the raw JSON into a relational PostgreSQL database.

Checked against the Actor's input schema and Apify docs on 2026-09-09.

How do you programmatically configure the Reddit Scraper?

You configure the Reddit Scraper by passing a structured JSON payload to the Apify API defining your target subreddits, date windows, and engagement filters. The underlying Actor routes all requests through high-quality residential proxies automatically, which avoids the HTTP 403 blocks that Reddit commonly throws at datacenter IP addresses.

The input schema requires the subreddits array, which accepts bare names like python, r/-prefixed names like r/python, or full URLs. To keep your database clean and avoid wasting compute resources on spam or stickied announcements, you can configure engagement thresholds like minScore and set excludeStickied to true.

Below is an input JSON configuration designed for a brand-monitoring and topic-extraction pipeline. It targets self-posts (text posts) with a minimum score of 10 upvotes, excludes stickied moderator threads, and filters for posts created within a specific date window.

{
  "subreddits": ["python", "datascience"],
  "maxPosts": 100,
  "sort": "new",
  "postType": "self",
  "postedAfter": "2026-09-01",
  "postedBefore": "2026-09-08",
  "minScore": 10,
  "excludeStickied": true,
  "excludeRemoved": true,
  "includeComments": false
}

How do you handle Apify API runs that exceed 300 seconds?

You handle long-running runs by initiating an asynchronous run and polling the status of the run, rather than using a synchronous call. The Apify synchronous run endpoint has a hard cap of 300 seconds, and if your scrape takes longer than 5 minutes, the synchronous API terminates the connection and returns an HTTP 408 error.

For data pipelines running in production, always run the Actor asynchronously. This involves making a POST request to initiate the run, which immediately returns a run ID and status. Your orchestrator then polls the run details until the status changes to SUCCEEDED.

Here is a Python script using the official apify-client library that demonstrates how to launch an asynchronous run, poll for completion, and safely download the resulting dataset items.

import os
import time
from apify_client import ApifyClient

# Initialize the client with your Apify API token
client = ApifyClient(os.getenv("APIFY_TOKEN"))

# Define the input configuration
actor_input = {
    "subreddits": ["python"],
    "maxPosts": 100,
    "sort": "new",
    "excludeStickied": True,
    "excludeRemoved": True
}

# Start the actor asynchronously. This does not block.
print("Starting the Reddit Scraper actor...")
run = client.actor("crawlerbros/reddit-scraper").call(
    run_input=actor_input,
    wait_secs=0  # Setting wait_secs to 0 forces an immediate asynchronous return
)

run_id = run["id"]
dataset_id = run["defaultDatasetId"]
print(f"Run started with ID: {run_id}")
print(f"Dataset ID: {dataset_id}")

# Poll the run status until it is finished
while True:
    run_details = client.run(run_id).get()
    status = run_details.get("status")
    print(f"Current run status: {status}")

    if status in ["SUCCEEDED", "FAILED", "ABORTED", "TIMED-OUT"]:
        break

    time.sleep(15)

if status == "SUCCEEDED":
    print("Extraction complete. Fetching dataset items...")
    dataset_items = client.dataset(dataset_id).list_items().items
    print(f"Retrieved items from the dataset.")
else:
    raise RuntimeError(f"Actor run failed with status: {status}")

How do you safely parse and clean nested JSON post data?

You parse and clean the output by defining a robust schema that accounts for omitted empty fields and maps the raw JSON key names into clean database columns. The reddit-scraper produces rich records containing up to 100 fields per post, but it omits empty fields entirely from the JSON payload to optimize bandwidth and storage.

Because fields like content (the markdown body of a self-post) or link_flair are omitted instead of set to null when they are absent, your ingestion script must use defensive dictionary access. If you write your extraction logic assuming every key in the schema is always present, your pipeline will crash with a KeyError.

The following Python snippet loads raw JSON records, defines a clean target structure, handles missing keys gracefully, and prepares a flat list of records for database insertion.

import pandas as pd

def clean_reddit_posts(raw_items):
    cleaned_records = []

    for item in raw_items:
        # Only process post records. If includeComments was enabled, 
        # comment records would have dataType 'comment'.
        if item.get("dataType") != "post":
            continue

        # Use defensive .get() calls with explicit fallbacks
        cleaned_post = {
            "post_id": item.get("post_id"),
            "subreddit": item.get("subreddit"),
            "title": item.get("title"),
            "author": item.get("author", "deleted_user"),
            "score": int(item.get("score", 0)),
            "upvote_ratio": float(item.get("upvote_ratio", 0.0)),
            "num_comments": int(item.get("num_comments", 0)),
            "content": item.get("content", ""), # Omitted if not a self-post
            "permalink": item.get("permalink"),
            "created_at": item.get("created_at"), # ISO-8601 string
            "link_flair": item.get("link_flair", None),
            "is_nsfw": bool(item.get("is_nsfw", False)),
            "is_original_content": bool(item.get("is_original_content", False))
        }

        # Guard against malformed records that lack a primary key
        if cleaned_post["post_id"]:
            cleaned_records.append(cleaned_post)

    # Convert to DataFrame for final validation and type coercion
    df = pd.DataFrame(cleaned_records)
    if not df.empty:
        df["created_at"] = pd.to_datetime(df["created_at"])
    return df

How do you stream cleaned records directly to PostgreSQL?

You stream cleaned records to PostgreSQL by establishing a database connection and executing an upsert statement that updates existing records based on a unique identifier. This ensures that any subsequent crawl of the same posts updates their scores and comment counts instead of throwing duplicate key errors.

Using SQLAlchemy, we can write a clean script that creates our target table if it does not exist, maps the keys from our pandas DataFrame to SQL parameters, and updates existing records using a PostgreSQL ON CONFLICT clause.

This script takes the cleaned data from our previous step and writes it into a target PostgreSQL table using SQLAlchemy.

import os
from sqlalchemy import create_engine, text

# Database connection configuration
DB_URI = os.getenv("DATABASE_URL", "postgresql://user:password@localhost:5432/reddit_db")
engine = create_engine(DB_URI)

# SQL statement to create the table if it does not exist
CREATE_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS reddit_posts (
    post_id VARCHAR(20) PRIMARY KEY,
    subreddit VARCHAR(50) NOT NULL,
    title TEXT NOT NULL,
    author VARCHAR(100),
    score INTEGER,
    upvote_ratio REAL,
    num_comments INTEGER,
    content TEXT,
    permalink TEXT,
    created_at TIMESTAMP WITH TIME ZONE,
    link_flair VARCHAR(100),
    is_nsfw BOOLEAN,
    is_original_content BOOLEAN,
    ingested_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);
"""

# Upsert query using PostgreSQL syntax
UPSERT_SQL = """
INSERT INTO reddit_posts (
    post_id, subreddit, title, author, score, upvote_ratio, 
    num_comments, content, permalink, created_at, link_flair, 
    is_nsfw, is_original_content
) VALUES (
    :post_id, :subreddit, :title, :author, :score, :upvote_ratio, 
    :num_comments, :content, :permalink, :created_at, :link_flair, 
    :is_nsfw, :is_original_content
) ON CONFLICT (post_id) DO UPDATE SET
    score = EXCLUDED.score,
    upvote_ratio = EXCLUDED.upvote_ratio,
    num_comments = EXCLUDED.num_comments,
    link_flair = EXCLUDED.link_flair,
    ingested_at = CURRENT_TIMESTAMP;
"""

def initialize_database():
    with engine.begin() as conn:
        conn.execute(text(CREATE_TABLE_SQL))
        print("Database schema verified.")

def upsert_records(df):
    if df.empty:
        print("No records to insert.")
        return

    records = df.to_dict(orient="records")

    # Execute batch upsert in a single transaction
    with engine.begin() as conn:
        for record in records:
            # Convert pandas Timestamp to Python datetime
            record["created_at"] = record["created_at"].to_pydatetime()
            conn.execute(text(UPSERT_SQL), record)

    print("Successfully upserted records to PostgreSQL.")

# Example execution flow
if __name__ == "__main__":
    # 1. Prepare database
    initialize_database()

    # 2. Mock raw data representing what the Apify Actor returned
    mock_raw_items = [
        {
            "dataType": "post",
            "post_id": "t3_16abcde",
            "subreddit": "python",
            "title": "Optimizing Django ORM Queries",
            "author": "django_dev",
            "score": 142,
            "upvote_ratio": 0.94,
            "num_comments": 23,
            "content": "Here is how I optimized my Django application database queries...",
            "permalink": "/r/python/comments/16abcde/optimizing_django_orm_queries/",
            "created_at": "2026-09-05T14:22:01.000Z",
            "is_nsfw": False,
            "is_original_content": True
        }
    ]

    # 3. Clean and insert
    cleaned_df = clean_reddit_posts(mock_raw_items)
    upsert_records(cleaned_df)

How do you handle deep comment thread extractions?

You handle deep comment thread extractions by enabling the comment-specific input parameters and iterating over both post and comment records in the returned dataset. The Actor can extract full comment threads with nested replies, depth, parent links, and OP flags when you set includeComments to true.

Because comment threads can become incredibly large, you must manage memory and execution time. You should configure maxCommentsPerPost to restrict the collection depth and avoid running into memory errors on the container.

The script below demonstrates how to handle an incoming dataset that contains both post and comment records, sorting them into separate tables or pandas dataframes based on the dataType column.

import pandas as pd

def process_combined_dataset(dataset_items):
    posts = []
    comments = []

    for item in dataset_items:
        data_type = item.get("dataType")

        if data_type == "post":
            posts.append({
                "post_id": item.get("post_id"),
                "subreddit": item.get("subreddit"),
                "title": item.get("title"),
                "author": item.get("author"),
                "score": item.get("score"),
                "num_comments": item.get("num_comments")
            })
        elif data_type == "comment":
            comments.append({
                "comment_id": item.get("comment_id"),
                "post_id": item.get("post_id"),
                "parent_id": item.get("parent_id"),
                "depth": item.get("depth", 0),
                "author": item.get("author"),
                "body": item.get("body"),
                "score": item.get("score")
            })

    df_posts = pd.DataFrame(posts)
    df_comments = pd.DataFrame(comments)

    return df_posts, df_comments

What are the limitations and failure modes of this pipeline?

There are several strict constraints on both the Reddit platform side and the Apify architecture that you must design your pipeline to tolerate.

Reddit's Hard Pagination Cap

Reddit limits public listings to approximately 1,000 posts. If your task requires collecting historical data, setting maxPosts to a very high number on a standard sort order like hot or top will still only yield about 1,000 items. To bypass this, you must run the Actor in chronological mode by setting postedAfter and postedBefore dates. This switches the internal collection to query the new listing, allowing you to walk back through historical time slices.

Residential Proxy Session Timeouts

The Actor enforces the use of Apify's residential proxies to bypass access blocks. However, residential proxy IP sessions are inherently unstable and expire after approximately 30 minutes. If you run a massive scrape that requests comments for hundreds of posts, the execution can take hours, and you will encounter transient network errors or proxy drops mid-run. For high-volume scrapes, always segment your runs by targeting smaller groups of subreddits or shorter time windows.

Memory Constraints with Nested Comments

Enabling full comment threads dramatically increases memory consumption and payload sizes. A single highly upvoted post can have thousands of nested comments. If you scrape many such posts with includeComments enabled and a high maxCommentsPerPost value, the container running the Actor can run out of memory. If you must extract comments, configure maxCommentsPerPost to a sensible cap and ensure the Actor is allocated sufficient RAM.

API Rate Limits on Storage

Apify's platform rate-limits dataset item pushes and Request Queue modifications to 400 requests per second per storage object. If you run multiple concurrent scraping runs trying to share a single Request Queue, your runs will experience rate-limiting errors. A request queue can only be processed by one Actor or task run at a time, though multiple runs may add to it. Fan-out across a single shared queue does not work. Always let each run write to its own default unnamed dataset, and read the items from that specific storage ID when the run finishes.

Furthermore, on the free plan, only the 10 most recent runs are retained, and they expire after 4 months. Unnamed storages also expire, whereas named storages are always exempt from deletion. For critical pipelines, ensure you save data before these thresholds are crossed.

How do you calculate and control Apify usage costs?

You calculate costs by summing the platform's compute consumption and proxy bandwidth charges based on the pricing tier you are subscribed to. The formulas are fixed, but the rates vary depending on whether you are on the Free, Starter, Scale, or Business plans.

Apify compute usage is billed in Compute Units (CUs). The formula to calculate compute unit consumption is:

CU = (memory_mb / 1024) * duration_hours

The price per CU and the cost per gigabyte of residential proxy usage are tied to your platform plan tier:

• Free Tier: $0.20 per CU, $8 per GB of residential proxy data.
• Starter Tier ($19/mo): $0.20 per CU, $8 per GB of residential proxy data.
• Scale Tier ($199/mo): $0.16 per CU, $7.50 per GB of residential proxy data.
• Business Tier ($999/mo): $0.13 per CU, $7 per GB of residential proxy data.

How Cost Scales with Pipeline Decisions

Your choice of input parameters has a massive impact on proxy data charges.

• Text-Only Post Scrapes: Scraping only post metadata (without comments and without media payload extraction) is highly efficient. A run yielding post records transfers small amounts of text data, representing negligible proxy costs.
• Deep Comment Scrapes: Enabling includeComments forces the crawler to load heavy JSON trees for every single post. This rapidly multiplies the number of HTTP requests and data transferred. Scraping posts along with their full comment threads can easily balloon the data transfer, adding significant proxy costs.

To protect against runaway costs caused by deep comment threads or extremely active subreddits, always configure the maxItems property in your input JSON. This acts as a global safety cap, and the Actor will gracefully terminate the run and output the collected data as soon as the total item count (posts and comments combined) hits your limit.

What is the most efficient way to schedule daily runs?

The most robust way to automate your daily data ingestion is to use Apify's native Scheduling tool combined with a clean-up API query. Schedules on Apify use a 6-field cron syntax (which optionally includes seconds).

To prevent empty runs from consuming your monthly resources, observe these platform constraints:

• Prior Run Requirement: You cannot schedule an Actor or task until it has successfully completed at least one manual run. Run your configured Actor once before attempting to automate it.
• Initial Status: All newly created schedules are set to DISABLED by default. You must explicitly toggle the schedule to active status in the console or via the API.
• Input Schema Prefills: The Apify Console UI displays a "prefill" section for inputs, but this prefill is completely ignored by API-triggered runs and scheduled runs. Only the explicit schema default values apply. Always define your payload parameters in the schedule's input JSON.

You can set up your daily run to post directly to your orchestration server when complete. Use a Webhook on your schedule or Actor task to send a POST payload containing the run ID to your database sync endpoint. Webhooks support exactly one action: POST to a URL. This is the only event-driven primitive on the platform. This event-driven architecture ensures your script only runs when the extraction is finished and verified, saving you from polling the Apify API continuously throughout the day.

How do you verify and debug your database schema with sample data?

You verify your relational schema by executing integration tests against local mock databases before running high-volume jobs on the Apify platform. This practices isolated testing and avoids racking up residential proxy bandwidth costs while debugging simple SQL syntax or parsing errors.

By setting up local unit tests, you can simulate different API payloads returned by the scraper. For instance, you should test how your pipeline handles self-posts, link-only posts, and posts that contain special characters or emojis in the title or content fields.

The following Python script illustrates how to run a complete end-to-end local validation test. It uses an in-memory SQLite database to mimic your PostgreSQL schema, processes sample records, and prints verification results to the console.

import sqlite3
import pandas as pd

# Sample raw records simulating actual JSON output from reddit-scraper
sample_api_response = [
    {
        "dataType": "post",
        "post_id": "t3_sample01",
        "subreddit": "datascience",
        "title": "Machine Learning Best Practices",
        "author": "ml_engineer",
        "score": 350,
        "upvote_ratio": 0.98,
        "num_comments": 45,
        "content": "Here is a list of production tips...",
        "permalink": "/r/datascience/comments/sample01/ml_best_practices/",
        "created_at": "2026-09-08T10:00:00.000Z",
        "is_nsfw": False,
        "is_original_content": True
    },
    {
        "dataType": "post",
        "post_id": "t3_sample02",
        "subreddit": "datascience",
        "title": "Interesting link on neural nets",
        "author": "researcher_12",
        "score": 85,
        "upvote_ratio": 0.88,
        "num_comments": 12,
        "permalink": "/r/datascience/comments/sample02/interesting_link/",
        "created_at": "2026-09-08T11:15:00.000Z",
        "is_nsfw": False,
        "is_original_content": False
    }
]

def run_local_pipeline_test():
    # 1. Initialize local in-memory SQLite database
    conn = sqlite3.connect(":memory:")
    cursor = conn.cursor()

    cursor.execute("""
    CREATE TABLE test_posts (
        post_id TEXT PRIMARY KEY,
        subreddit TEXT,
        title TEXT,
        author TEXT,
        score INTEGER,
        upvote_ratio REAL,
        num_comments INTEGER,
        content TEXT,
        permalink TEXT,
        created_at TEXT,
        is_nsfw INTEGER,
        is_original_content INTEGER
    );
    """)

    # 2. Extract and clean sample items
    cleaned_items = []
    for item in sample_api_response:
        if item.get("dataType") != "post":
            continue

        cleaned_items.append((
            item.get("post_id"),
            item.get("subreddit"),
            item.get("title"),
            item.get("author", "deleted"),
            int(item.get("score", 0)),
            float(item.get("upvote_ratio", 0.0)),
            int(item.get("num_comments", 0)),
            item.get("content", ""), # Handles missing content on link-only posts
            item.get("permalink"),
            item.get("created_at"),
            1 if item.get("is_nsfw") else 0,
            1 if item.get("is_original_content") else 0
        ))

    # 3. Insert and verify data
    cursor.executemany("""
    INSERT OR REPLACE INTO test_posts VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    """, cleaned_items)
    conn.commit()

    # 4. Query back and assert values match expectations
    df = pd.read_sql_query("SELECT * FROM test_posts", conn)
    print("Database verification complete. Records written successfully:")
    print(df[["post_id", "subreddit", "score", "num_comments"]])

    # Assertions to verify parsing logic was clean
    assert len(df) == 2, "Should have successfully parsed and written two records."
    assert df.loc[df["post_id"] == "t3_sample02", "content"].values[0] == "", "Link post body should default to empty string."
    print("All pipeline assertions passed.")

if __name__ == "__main__":
    run_local_pipeline_test()

The Actor's README is the source of truth for its inputs, outputs and limits. Written with AI assistance. Need a hand wiring this into your stack? Email [e-mail]

Comments

[Luis Cruz]

This is a great walkthrough of setting up a Reddit data pipeline using Apify and PostgreSQL! I particularly appreciate the emphasis on handling the asynchronous nature of API runs; it's essential for building resilient data pipelines. For future enhancements, consider implementing monitoring and alerting for scrape failures to quickly address issues that may arise from changes in Reddit's structure or API limitations. If you're looking for additional support in refining this pipeline or tackling related challenges, I’d be glad to discuss a paid collaboration. How have you found the performance of this setup in a production environment so far?