Automating Data Pipelines: From Scraping to Dashboard

Contains affiliate links. Disclosure.

You set up a scraper that works perfectly — but you run it manually, copy the CSV into a spreadsheet, and forget to check it for two weeks. That's not a data pipeline. A real automated scraping pipeline runs on a schedule, cleans the data, stores it, renders a live dashboard, and pages you when something goes wrong.

This guide walks through every stage of that system using a concrete example: daily e-commerce price monitoring, from scheduled Apify actor to Grafana dashboard with Slack alerts.

The Five Stages of a Scraping-to-Dashboard Pipeline

Before touching code, map your pipeline as five sequential stages. Each stage has one job and one output:

Stage	Job	Output
1. Schedule	Trigger the scraper reliably	Raw JSON/HTML
2. Transform	Clean and enrich raw data	Normalised rows
3. Store	Persist structured records	Database table
4. Visualise	Render queryable charts	Live dashboard
5. Alert	Detect anomalies and notify	Slack/email message

Keeping stages decoupled matters. If your storage schema changes, only stage 3 needs updating. If you swap Grafana for Metabase, stages 1–3 stay untouched.

---

Stage 1 — Scheduled Scraping with Apify

Apify is the fastest way to schedule cloud scraping without managing infrastructure. You deploy an Actor (a containerised scraping function), attach a Schedule, and Apify handles retries, proxy rotation, and result storage.

Set up the Actor

For this example, use the Amazon Product Scraper Actor from Apify Store. It accepts a list of product ASINs or search keywords and returns structured product data including title, price, rating, and availability.

Configure the Actor input via the Apify Console or the API:

{
  "searchKeywords": ["running shoes men", "trail running shoes"],
  "maxItemsPerSearch": 50,
  "countryCode": "US"
}

Attach a schedule

In the Apify Console, open the Actor → Schedules → New Schedule. Set it to run daily at 06:00 UTC:

0 6 * * *

Apify runs the Actor, stores results in a Dataset (append mode), and provides a REST endpoint to retrieve new rows since the last run.

Retrieve results via the API

Every Apify Dataset has a REST endpoint. Fetch only the rows added in the last 24 hours using the clean=true and offset parameters:

import httpx
import os
from datetime import datetime, timedelta, timezone

APIFY_TOKEN = os.environ["APIFY_TOKEN"]
DATASET_ID = os.environ["APIFY_DATASET_ID"]

def fetch_new_rows(hours_back: int = 25) -> list[dict]:
    """Fetch rows added to the Apify Dataset in the last `hours_back` hours.

    The default is 25 (not 24) to account for scheduling jitter — if the
    Actor fires slightly late, a 24-hour window could miss the newest rows.
    """
    since = datetime.now(timezone.utc) - timedelta(hours=hours_back)
    url = f"https://api.apify.com/v2/datasets/%7BDATASET_ID%7D/items?fpr=use-apify"
    params = {
        "token": APIFY_TOKEN,
        "clean": "true",
        "format": "json",
    }
    response = httpx.get(url, params=params, timeout=30)
    response.raise_for_status()
    rows = response.json()
    # Filter to rows from the last N hours
    return [
        r for r in rows
        if datetime.fromisoformat(r.get("scrapedAt", "1970-01-01T00:00:00+00:00")) >= since
    ]

For more on running concurrent scraping jobs to feed this pipeline, see Async Web Scraping: Concurrency Patterns.

---

Stage 2 — Data Transformation

Raw scraper output is noisy. Prices arrive as strings ("$89.99"), titles have trailing whitespace, and some rows have missing fields. The transformation stage normalises everything before it touches your database.

Cleaning function

import re
from decimal import Decimal, InvalidOperation
from dataclasses import dataclass
from typing import Optional

@dataclass
class ProductPrice:
    asin: str
    title: str
    price_usd: Optional[Decimal]
    rating: Optional[float]
    review_count: int
    availability: str
    scraped_at: str

def parse_price(raw: str) -> Optional[Decimal]:
    """Extract numeric price from strings like '$89.99', '89,99', etc."""
    if not raw:
        return None
    cleaned = re.sub(r"[^\d.]", "", raw.replace(",", "."))
    try:
        return Decimal(cleaned)
    except InvalidOperation:
        return None

def transform_row(raw: dict) -> Optional[ProductPrice]:
    """Normalise a single scraper output row. Returns None if critical fields missing."""
    asin = raw.get("asin") or raw.get("id", "")
    if not asin:
        return None

    return ProductPrice(
        asin=asin.strip(),
        title=(raw.get("title") or "").strip()[:500],
        price_usd=parse_price(raw.get("price", "")),
        rating=float(raw["rating"]) if raw.get("rating") else None,
        review_count=int(raw.get("reviewsCount", 0) or 0),
        availability=(raw.get("availability") or "unknown").strip().lower(),
        scraped_at=raw.get("scrapedAt", ""),
    )

def transform_batch(rows: list[dict]) -> list[ProductPrice]:
    results = [transform_row(r) for r in rows]
    return [r for r in results if r is not None]

Enrichment

Beyond cleaning, enrichment adds computed columns. For price monitoring, the most useful enrichment is day-over-day delta: how much did this product's price change since yesterday?

def enrich_with_delta(
    today: list[ProductPrice],
    yesterday_map: dict[str, Decimal],
) -> list[dict]:
    enriched = []
    for p in today:
        row = p.__dict__.copy()
        if p.price_usd is not None and p.asin in yesterday_map:
            prev = yesterday_map[p.asin]
            row["price_delta_usd"] = float(p.price_usd - prev)
            row["price_delta_pct"] = float((p.price_usd - prev) / prev * 100) if prev else None
        else:
            row["price_delta_usd"] = None
            row["price_delta_pct"] = None
        enriched.append(row)
    return enriched

---

Stage 3 — Storage

You have two practical options for storing pipeline data: Apify Datasets (for lightweight, query-free storage) or a relational database (for complex queries and dashboard connectors).

Option A: Apify Datasets

Apify Datasets are append-only JSON stores. They require no infrastructure and integrate directly with Apify webhooks. Use them when:

• You need results accessible to other Apify Actors
• Your dashboard tool can consume the Apify REST API (e.g. via a custom connector)
• You want to avoid database maintenance entirely

Option B: PostgreSQL on Liquid Web

For a production dashboard with historical queries, a managed PostgreSQL database is the right call. Liquid Web's managed databases give you a fully-managed PostgreSQL instance — backups, failover, and connection pooling handled for you.

Create the table schema:

CREATE TABLE product_prices (
    id           BIGSERIAL PRIMARY KEY,
    asin         VARCHAR(20)    NOT NULL,
    title        VARCHAR(500),
    price_usd    NUMERIC(10,2),
    rating       NUMERIC(3,1),
    review_count INTEGER        DEFAULT 0,
    availability VARCHAR(100),
    price_delta_usd NUMERIC(10,2),
    price_delta_pct NUMERIC(6,2),
    scraped_at   TIMESTAMPTZ    NOT NULL,
    inserted_at  TIMESTAMPTZ    NOT NULL DEFAULT NOW(),
    UNIQUE (asin, scraped_at)
);

CREATE INDEX idx_product_prices_asin      ON product_prices (asin);
CREATE INDEX idx_product_prices_scraped_at ON product_prices (scraped_at DESC);

Insert transformed rows using psycopg (the modern PostgreSQL adapter for Python 3):

import psycopg
import os
from typing import Iterable

DB_URL = os.environ["DATABASE_URL"]  # postgres://user:pass@host:5432/dbname

def upsert_prices(rows: Iterable[dict]) -> int:
    row_list = list(rows)
    if not row_list:
        return 0
    with psycopg.connect(DB_URL) as conn:
        with conn.cursor() as cur:
            cur.executemany(
                """
                INSERT INTO product_prices
                    (asin, title, price_usd, rating, review_count,
                     availability, price_delta_usd, price_delta_pct, scraped_at)
                VALUES
                    (%(asin)s, %(title)s, %(price_usd)s, %(rating)s, %(review_count)s,
                     %(availability)s, %(price_delta_usd)s, %(price_delta_pct)s, %(scraped_at)s)
                ON CONFLICT (asin, scraped_at) DO NOTHING
                """,
                row_list,
            )
        conn.commit()
    return len(row_list)

For a full walkthrough of combining Firecrawl with a Supabase/PostgreSQL backend, see Firecrawl + Supabase: Store Scraped Data in PostgreSQL.

Storage comparison

Option	Setup effort	Query flexibility	Cost at 1M rows/month
Apify Dataset	Zero	Low (REST only)	Included in Apify plan
PostgreSQL (Liquid Web)	15 min	Full SQL	~$15–$30/month
SQLite (local)	Zero	Full SQL	Free

Pricing verified June 2026. Plans change frequently — check the vendor’s official pricing page for current rates.

Best for small projects: Apify Datasets — no ops overhead.
Best for production dashboards: PostgreSQL on Liquid Web — full SQL, reliable connectors.

---

Stage 4 — Visualization

Two open-source tools dominate self-hosted dashboards for scraped data: Grafana and Metabase.

Grafana vs Metabase at a glance

Feature	Grafana	Metabase
Primary strength	Time-series and metrics	Business intelligence / ad-hoc SQL
Data source	PostgreSQL, MySQL, InfluxDB, 50+ more	PostgreSQL, MySQL, BigQuery, 20+ more
Chart types	Graphs, gauges, heatmaps	Tables, charts, pivot tables, maps
SQL required	Sometimes (for complex panels)	Optional (has GUI query builder)
Self-hosted	Free (OSS)	Free (OSS)
Best for	Ops/monitoring dashboards	Business reporting

For an e-commerce price pipeline, Grafana is the better fit: prices over time are a time-series problem, and Grafana's panel library handles it well.

Grafana setup with Docker

# docker-compose.yml
services:
  grafana:
    image: grafana/grafana:11.0.0
    ports:
      - "3000:3000"
    environment:
      GF_SECURITY_ADMIN_PASSWORD: changeme
    volumes:
      - grafana_data:/var/lib/grafana

volumes:
  grafana_data:

docker compose up -d

Open http://localhost:3000, add a PostgreSQL data source pointing to your Liquid Web database, then create a dashboard panel with this SQL query:

SELECT
    scraped_at                                    AS "time",
    asin,
    title,
    price_usd
FROM product_prices
WHERE
    asin IN ($asin_filter)          -- Grafana variable
    AND scraped_at >= NOW() - INTERVAL '30 days'
ORDER BY scraped_at ASC;

Enable the asin_filter as a Grafana variable (Query type, data source = PostgreSQL) to let users pick which products to compare.

Key panels to build

1. Price over time — time-series line chart per ASIN
2. Price delta heatmap — which products moved the most this week
3. Availability tracker — bar chart showing "in stock" percentage over time
4. Top movers table — products with the biggest price drop in the last 24 hours, sorted by price_delta_pct

---

Stage 5 — Alerting

A dashboard you have to open manually is a passive tool. An alerting layer turns your pipeline into an active system that tells you when something interesting happens.

Grafana Alerts

Grafana's built-in alerting evaluates SQL queries on a schedule and fires when a condition is met. Create a Grafana Alert rule:

1. Query: SELECT MAX(ABS(price_delta_pct)) FROM product_prices WHERE scraped_at >= NOW() - INTERVAL '2 hours'
2. Condition: IS ABOVE 15 (trigger if any product moved more than 15% in 2 hours)
3. Contact point: Slack webhook URL (paste into Grafana → Alerting → Contact points)

Slack webhook notification

You can also fire alerts directly from your Python pipeline without Grafana. This gives you more control over message formatting:

import httpx
import os

SLACK_WEBHOOK_URL = os.environ["SLACK_WEBHOOK_URL"]

def send_price_alert(asin: str, title: str, delta_pct: float, new_price: float) -> None:
    direction = "📉 dropped" if delta_pct < 0 else "📈 jumped"
    message = {
        "text": (
            f"*Price alert* — {title} ({asin})\n"
            f"Price {direction} by *{abs(delta_pct):.1f}%* → now *${new_price:.2f}*"
        )
    }
    response = httpx.post(SLACK_WEBHOOK_URL, json=message, timeout=10)
    response.raise_for_status()

def check_and_alert(enriched_rows: list[dict], threshold_pct: float = 10.0) -> None:
    for row in enriched_rows:
        delta = row.get("price_delta_pct")
        if delta is not None and abs(delta) >= threshold_pct:
            send_price_alert(
                asin=row["asin"],
                title=row["title"],
                delta_pct=delta,
                new_price=float(row["price_usd"]),
            )

Email alerts via SendGrid

For email-based alerts, swap the Slack call for a SendGrid API request. The same check_and_alert function structure works — just replace the HTTP call:

def send_email_alert(to: str, subject: str, body: str) -> None:
    headers = {
        "Authorization": f"Bearer {os.environ['SENDGRID_API_KEY']}",
        "Content-Type": "application/json",
    }
    payload = {
        "personalizations": [{"to": [{"email": to}]}],
        "from": {"email": "alerts@yourdomain.com"},
        "subject": subject,
        "content": [{"type": "text/plain", "value": body}],
    }
    httpx.post(
        "https://api.sendgrid.com/v3/mail/send",
        headers=headers,
        json=payload,
        timeout=10,
    ).raise_for_status()

---

Wiring It All Together

Here's the complete orchestration script that ties all five stages into one runnable pipeline:

#!/usr/bin/env python3
"""
pipeline.py — run once daily from a cron job or Apify webhook.
Environment variables required:
  APIFY_TOKEN, APIFY_DATASET_ID, DATABASE_URL,
  SLACK_WEBHOOK_URL (optional)
"""

import logging
from decimal import Decimal

# Import stage modules (shown in full above)
from pipeline.fetch    import fetch_new_rows
from pipeline.transform import transform_batch, enrich_with_delta
from pipeline.store    import upsert_prices, fetch_yesterday_prices
from pipeline.alert    import check_and_alert

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger(__name__)

def run() -> None:
    # Stage 1 — Fetch
    log.info("Fetching new rows from Apify Dataset…")
    raw_rows = fetch_new_rows(hours_back=25)
    log.info("Fetched %d raw rows", len(raw_rows))

    # Stage 2 — Transform
    cleaned = transform_batch(raw_rows)
    yesterday_map: dict[str, Decimal] = fetch_yesterday_prices()
    enriched = enrich_with_delta(cleaned, yesterday_map)
    log.info("Transformed %d rows (%d dropped)", len(enriched), len(raw_rows) - len(enriched))

    # Stage 3 — Store
    inserted = upsert_prices(enriched)
    log.info("Inserted %d rows into PostgreSQL", inserted)

    # Stage 5 — Alert (dashboard is always live — no code needed to "run" it)
    check_and_alert(enriched, threshold_pct=10.0)
    log.info("Alert check complete")

if __name__ == "__main__":
    run()

Schedule this script with a cron job on your Liquid Web server, or trigger it via an Apify webhook that fires after each successful Actor run.

Triggering via Apify webhook

In the Apify Console, go to your Actor → Webhooks → New Webhook:

• Event type: Actor run succeeded
• Request URL: https://your-server.com/pipeline/trigger
• Payload template: leave default (sends run metadata)

Your server receives a POST request after every successful scrape, triggers pipeline.py, and the dashboard updates within seconds of new data arriving.

---

Scaling and Production Considerations

As your pipeline grows beyond one Actor and one table, a few practices prevent it from becoming fragile:

Idempotency: Make stage 3 safe to re-run. If the pipeline crashes mid-insert and retries, you should not get duplicate rows. The schema above includes a UNIQUE constraint on (asin, scraped_at) and the insert uses ON CONFLICT DO NOTHING, so re-runs are safe by default.

Dead-letter logging: Log rows that fail transformation to a separate table (pipeline_errors) instead of silently dropping them. You'll thank yourself when a data format change silently zeroes out your price columns.

Schema migrations: Use Alembic (Python) or Flyway to version your database schema. Never alter production tables by hand.

Monitor the pipeline itself: Add a Grafana panel that charts inserted_at row counts per hour. A flat line means the scraper stopped running — catch it before your dashboard goes stale.

For more on e-commerce price monitoring strategies and Actor selection, see Automated Price Monitoring for Ecommerce: 2026 Blueprint.

---

FAQ

How do I automate a scraping pipeline?

Use a scheduling tool to trigger your scraper on a cron-style schedule (Apify Schedules, a server cron job, or a workflow tool like GitHub Actions). After each run, a webhook or polling script fetches the new data, transforms it, and writes it to your storage layer. The dashboard reads from storage in real time — no manual steps needed.

What tools do I need for a data pipeline?

A minimal scraping-to-dashboard pipeline needs four components: a scraper with scheduling (Apify), a transformation script (Python), a database (PostgreSQL), and a visualisation tool (Grafana or Metabase). For alerting, add Slack webhooks or an email API like SendGrid. All of these have free tiers or open-source versions.

How do I build a dashboard from scraped data?

Connect Grafana or Metabase directly to your PostgreSQL database. Both tools have native PostgreSQL connectors. Write SQL queries against your data tables and use the tools' panel editors to turn query results into time-series charts, tables, and gauges. No intermediary export step is needed — the dashboard queries live data on page load or on a refresh interval.

---

Start Your Pipeline Today

The scaffolding above gives you a working end-to-end system in an afternoon. The biggest ROI comes from Stage 1 — once Apify is running on a schedule and data is accumulating in PostgreSQL, stages 2–5 are incremental improvements.

Start scraping with Apify for free — the free plan includes enough compute for a daily price-monitoring Actor. Pair it with a Liquid Web managed database when you're ready for a production-grade storage layer.