Syncing MLB Stats to pandas Without Double Processing

Orchestrating Real-Time Sports Data Pipelines

Running a scraper inside a scheduled cron job is only ten percent of a production data engineering workflow. The real complexity lies in what happens downstream: moving that data into your analytical tools, warehouse, or automated workflows without missing updates or processing the same records twice. When working with high-volume sports data, your ingestion pipeline must be idempotent, resilient to network timeouts, and structured to handle state transitions smoothly.

We will focus on wiring the mlb-stats-scraper into an automated data pipeline. Instead of looking at this tool as a standalone utility, we will treat it as an upstream data source feeding into a pandas analytics environment and an n8n automation flow. We will solve the state-management problem that causes pipeline re-runs to double-process records, handle the platform-level limits of the Apify execution model, and construct a robust synchronization loop.

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

How do you prevent duplicate data on scheduled pipeline runs?

To prevent duplicate data, you must maintain a state file in a named Key-Value store that tracks the last successfully processed timestamp or game primary key. Your orchestrator queries this state before executing the run, adjusts the input parameters dynamically, and updates the state only after downstream ingestion succeeds. This mechanism ensures that if a run fails mid-process, the subsequent execution resumes from the last known good record rather than re-importing identical rows.

When running scheduled updates, developers often pass a static date range or pull the current season's entire dataset. This practice creates redundant write operations downstream. Because unnamed storages on the Apify free plan expire, retaining only the 10 most recent runs for a duration of 4 months, storing state inside default run storages is unreliable. Instead, you must use a named Key-Value store, which is exempt from deletion, to keep a persistent record of the last processed game date or ID.

The following Python script demonstrates how to fetch the last processed date from a named Key-Value store, use it to construct the input payload for the scraper, execute the run, and update the state upon completion.

import os
from apify_client import ApifyClient

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

# Define a named Key-Value store to persist pipeline state
store_name = "mlb-pipeline-state"
store = client.key_value_stores().get_or_create(name=store_name)

# Retrieve the last processed date; default to a safe start date if empty
state_record = client.key_value_store(store["id"]).get_record("last_processed_date")
last_date = state_record["value"] if state_record else "2024-07-01"

# Construct the input payload dynamically using the state value
# Note: Always pass an explicit input dict; 'prefill' in Console UI is not applied to API calls
run_input = {
    "mode": "schedule",
    "fromDate": last_date,
    "toDate": "2024-07-10",
    "gameType": "R",
    "maxItems": 100
}

# Start the Actor run on the Apify platform
# We target crawlerbros/mlb-stats-scraper specifically
run = client.actor("crawlerbros/mlb-stats-scraper").call(run_input=run_input)

# Check if the run completed successfully and fetch data
if run.get("status") == "SUCCEEDED":
    dataset_id = run.get("defaultDatasetId")
    items = client.dataset(dataset_id).list_items().items

    if items:
        # Extract the latest game date from the retrieved dataset items
        # Dates are returned in ISO 8601 format: YYYY-MM-DD
        dates = [item["gameDate"][:10] for item in items if "gameDate" in item]
        if dates:
            new_last_date = max(dates)

            # Update our persistent named Key-Value store with the new high-water mark
            client.key_value_store(store["id"]).set_record(
                "last_processed_date", 
                new_last_date
            )
            print(f"Pipeline executed successfully. State updated to: {new_last_date}")
    else:
        print("No new records returned. State remains unchanged.")
else:
    print(f"Actor run failed with status: {run.get('status')}")

How to handle the 300-second synchronous API execution limit?

You handle the timeout limit by initiating runs asynchronously to obtain a run ID, then polling the run status or registering a webhook. This avoids the HTTP 408 error triggered when synchronous requests exceed the platform's hard 300-second execution cap. By shifting to an asynchronous pattern, you can manage long-running extractions of historical statistics or massive schedule ranges without dropping connections.

When calling an Actor via a synchronous POST request, the platform forces a hard timeout at 300 seconds. If the API source experiences latency or if you request a large volume of historical statistics, your connection will drop with an HTTP 408 response. For production orchestrations, you must decouple the execution request from the retrieval phase.

This is accomplished by POSTing to the run endpoint to trigger an asynchronous run, which instantly returns a run object containing the run ID, and then monitoring the status. Below is a Node.js implementation of this asynchronous execution and polling pattern.

const axios = require('axios');

async function triggerMlbPipeline() {
  const apifyToken = process.env.APIFY_TOKEN;
  const actorId = 'crawlerbros/mlb-stats-scraper';

  const runInput = {
    mode: 'playerStats',
    playerId: 660271, // Juan Soto
    season: 2024,
    statGroup: 'hitting'
  };

  try {
    // Trigger the run asynchronously. This returns immediately without waiting for completion.
    const startResponse = await axios.post(
      `https://api.apify.com/v2/acts/${actorId}/runs?token=${apifyToken}`,
      runInput,
      { headers: { 'Content-Type': 'application/json' } }
    );

    const runId = startResponse.data.data.id;
    console.log(`Successfully started run. Run ID: ${runId}`);

    // Poll the run status until it reaches a terminal state
    const maxPollAttempts = 30;
    const pollIntervalMs = 10000; // 10 seconds

    for (let attempt = 1; attempt <= maxPollAttempts; attempt++) {
      await new Promise(resolve => setTimeout(resolve, pollIntervalMs));

      const statusResponse = await axios.get(
        `https://api.apify.com/v2/acts/${actorId}/runs/${runId}?token=${apifyToken}`
      );

      const status = statusResponse.data.data.status;
      console.log(`Attempt ${attempt}: Run status is ${status}`);

      if (status === 'SUCCEEDED') {
        const datasetId = statusResponse.data.data.defaultDatasetId;
        console.log(`Run finished! Dataset ID: ${datasetId}`);
        return datasetId;
      } else if (['FAILED', 'ABORTED', 'TIMED-OUT'].includes(status)) {
        throw new Error(`Actor run ended with terminal status: ${status}`);
      }
    }
    throw new Error('Polling timed out before Actor execution completed.');
  } catch (error) {
    console.error('Pipeline orchestration error:', error.message);
    throw error;
  }
}

triggerMlbPipeline();

Connecting Scraper Outputs to n8n Webhook Receivers

While polling works well for scripts, real-time enterprise pipelines should rely on event-driven execution. The platform provides exactly one event-driven primitive: webhooks that POST a payload to a target URL when a specific run event occurs. You can leverage this to push data straight to an n8n workflow.

Because n8n has a specialized Apify Trigger node, you can easily configure self-hosted or cloud instances to wake up on run completion. If you are self-hosting, authenticate using your standard API key; if you are on n8n Cloud, you can use OAuth2 credentials. The trigger node eliminates the need for polling loops by responding instantly to platform events.

Below is an example of the webhook payload configuration you can set on your Actor task. This JSON is evaluated when the run finishes and is POSTed directly to your webhook handler.

{
  "eventName": "ACTOR.RUN.SUCCEEDED",
  "payload": {
    "userId": "{{userId}}",
    "actorId": "{{actorId}}",
    "actorRunId": "{{resource.id}}",
    "datasetId": "{{resource.defaultDatasetId}}",
    "metadata": {
      "mode": "schedule",
      "processedAt": "{{createdAt}}"
    }
  }
}

When this payload hits your n8n workflow, the first node extracts the datasetId and retrieves the list of items from the dataset endpoint. The downstream nodes can then map these objects directly into database columns or push notifications through your custom messaging channels.

Feeding JSON Datasets Directly Into pandas DataFrames

For analytics and machine learning applications, your pipeline needs to transition data from JSON payloads to memory-optimized pandas DataFrames. The mlb-stats-scraper outputs clean JSON objects with defined schemas for teams, players, schedule, standings, and player stats.

When fetching data from the default dataset, you can request the output directly in CSV or JSON format. Reading directly into pandas via the platform's API endpoint is the most efficient method, as it bypasses the need to write intermediate files to disk.

import pandas as pd
import requests

def load_mlb_data_to_pandas(dataset_id: str, apify_token: str) -> pd.DataFrame:
    """
    Fetches clean JSON items from a completed scraper dataset
    and loads them directly into a structured pandas DataFrame.
    """
    # Use the clean format endpoint to avoid platform wrapper keys
    api_url = f"https://api.apify.com/v2/datasets/{dataset_id}/items?token={apify_token}&format=json&clean=true"

    try:
        response = requests.get(api_url)
        response.raise_for_status()
        data = response.json()

        if not data:
            print("Dataset is empty. Returning empty DataFrame.")
            return pd.DataFrame()

        # Convert JSON array to structured DataFrame
        df = pd.DataFrame(data)

        # Enforce standard datatypes and handle schema anomalies
        if "gameDate" in df.columns:
            df["gameDate"] = pd.to_datetime(df["gameDate"])
        if "gamePk" in df.columns:
            df["gamePk"] = df["gamePk"].astype("Int64")  # Nullable integer type

        return df

    except requests.exceptions.RequestException as e:
        print(f"Failed to fetch dataset: {e}")
        raise

How do you protect pipelines from API rate limits and data failures?

To protect pipelines, you must implement exponential backoff on data storage calls, enforce a hard spending cap using maxTotalChargeUsd, and validate that downstream systems do not flood the storage endpoints beyond their limits. By utilizing retry logic and programmatic cost caps, you ensure that sporadic network anomalies or runaway code blocks do not crash your orchestrations or lead to unexpected bills.

The data storage system imposes specific rate limits: 60 requests per second per storage object, and 400 requests per second for dataset item pushes. If your downstream worker parses a large dataset and makes concurrent reads or writes to the same Key-Value store, the platform will rate-limit your requests. To handle this, implement a sleep delay with exponential backoff on any storage operations.

Additionally, to prevent runaway costs from unexpected execution loops, always pass the maxTotalChargeUsd parameter on your run endpoints. This parameter is exposed inside the execution environment as the ACTOR_MAX_TOTAL_CHARGE_USD environment variable. When this spending cap is tripped, the run terminates. However, keep in mind that termination is not an instant kill: the container continues consuming resources briefly during the shutdown phase, so you should monitor your usage metrics.

The following Python example shows how to safely handle potential errors, implement basic retries, and validate the structural shape of your incoming data before running database update commands.

import time
import requests
from typing import Dict, Any

def fetch_dataset_with_backoff(dataset_id: str, apify_token: str, max_retries: int = 5) -> Dict[str, Any]:
    """
    Retrieves dataset items using an exponential backoff loop to respect 
    the platform storage request rate limits.
    """
    url = f"https://api.apify.com/v2/datasets/{dataset_id}/items?token={apify_token}&clean=true"
    delay = 1.0  # Initial delay in seconds

    for attempt in range(1, max_retries + 1):
        try:
            response = requests.get(url)

            # If rate-limited (HTTP 429), trigger backoff
            if response.status_code == 429:
                print(f"Rate limited (429). Attempt {attempt} of {max_retries}. Waiting {delay}s...")
                time.sleep(delay)
                delay *= 2  # Double the backoff duration
                continue

            response.raise_for_status()
            return response.json()

        except requests.exceptions.RequestException as e:
            if attempt == max_retries:
                print("Max retries reached. Pipeline ingestion failed.")
                raise e
            time.sleep(delay)
            delay *= 2

    raise RuntimeError("Failed to fetch dataset due to persistent errors.")

A Worked Example of Incremental Standings Updates

To see these pieces working together, consider a complete Python synchronization script. The objective is to pull the division standings for the American League and National League, verify that new standings data exists, compare it against our local pandas-based cache, and append any changes to our long-term database.

This script uses the standings mode of the scraper to pull real-time standings data. It pulls the data, reads our cached records from a named Key-Value store, compares the win/loss records, and stores the updated table back to the platform.

import os
import pandas as pd
from apify_client import ApifyClient

# Initialize client
client = ApifyClient(token=os.environ.get("APIFY_TOKEN"))

# Define store and cache keys
store_name = "mlb-standings-cache-store"
cache_key = "last_standings_cache"
store = client.key_value_stores().get_or_create(name=store_name)

# 1. Fetch previous standings cache
cache_record = client.key_value_store(store["id"]).get_record(cache_key)
previous_df = pd.DataFrame(cache_record["value"]) if cache_record else pd.DataFrame()

# 2. Run the Actor in standings mode
run_input = {
    "mode": "standings",
    "season": 2024,
    "leagueId": "both",
    "maxItems": 100
}

print("Executing standings update run...")
run = client.actor("crawlerbros/mlb-stats-scraper").call(run_input=run_input)

if run.get("status") == "SUCCEEDED":
    dataset_id = run.get("defaultDatasetId")
    fresh_items = client.dataset(dataset_id).list_items().items

    if fresh_items:
        current_df = pd.DataFrame(fresh_items)

        # Select crucial fields to determine if standings shifted
        comparison_cols = ["teamName", "wins", "losses", "pct"]

        if not previous_df.empty and set(comparison_cols).issubset(previous_df.columns):
            # Check if there are differences between old and new records
            merged = current_df[comparison_cols].merge(
                previous_df[comparison_cols],
                on="teamName",
                suffixes=("_new", "_old")
            )

            # Identify teams where wins or losses changed
            changed_mask = (merged["wins_new"] != merged["wins_old"]) | (merged["losses_new"] != merged["losses_old"])
            changed_teams = merged[changed_mask]

            if not changed_teams.empty:
                print(f"Standings updated! Detected changes for {len(changed_teams)} teams.")
                print(changed_teams[["teamName", "wins_new", "losses_new"]])
            else:
                print("Standings data received, but no records have changed since last run.")
        else:
            print("First run complete. Creating initial standings cache.")

        # 3. Update the persistent named Key-Value cache with fresh data
        # We convert the DataFrame back to a JSON-compatible dict structure
        client.key_value_store(store["id"]).set_record(
            cache_key,
            current_df.to_dict(orient="records")
        )
    else:
        print("No standings records found in the dataset.")
else:
    print(f"Standings scraper run failed with status: {run.get('status')}")

Platform Resource Requirements and Cost Analysis

To run your orchestrations cost-effectively, you must understand how your resource consumption translates into actual billing. Compute consumption on the platform is measured in Compute Units (CUs), calculated with the formula:

CU = (Memory in MB / 1024) * Duration in Hours

Doubling the memory allocation is CU-neutral only for autoscaling runs. However, autoscaling only applies to complex solutions running multiple tasks or URLs for at least 30 seconds each. Because the MLB Stats Scraper executes single-purpose API fetches to the official MLB Stats API, it operates on a standard container. This means choosing a larger memory setting arbitrarily will directly increase your CU consumption rate without providing any execution speed advantage.

Let us look at the billing rates across the different subscription tiers to calculate the exact resource costs:

• Free ($0): Compute Unit rate is $0.20 per CU. Residential proxy rate is $8 per GB. Only the 10 most recent runs are retained, for 4 months.
• Starter ($19/mo): Compute Unit rate is $0.20 per CU. Residential proxy rate is $8 per GB.
• Scale ($199/mo): Compute Unit rate is $0.16 per CU. Residential proxy rate is $7.50 per GB.
• Business ($999/mo): Compute Unit rate is $0.13 per CU. Residential proxy rate is $7 per GB.

Because the upstream MLB Stats API is public and does not require authentication, the scraper does not need complex residential proxies. It runs entirely on standard datacenter proxies, which are included in the platform's subscription tiers. This eliminates the $7 to $8 per gigabyte residential proxy charge entirely from your operational costs.

Your primary cost driver is container run time. If you run a standard container to retrieve stats, your cost scale is determined entirely by how long your container runs. For instance, at the Free and Starter tier rate of $0.20 per CU, every run represents a small fraction of a cent. Even when scheduled to run hourly throughout the regular season, the total cost remains extremely low. This allows you to allocate your monthly budget to downstream database ingestion, storage, and processing tasks.

Real-World Limitations and Caveats

While the scraper provides direct access to real-time baseball data, building an enterprise-grade pipeline around it requires accounting for several platform and upstream limitations.

• Single-Consumer Request Queues: A request queue can only be processed by one Actor or task run at a time. If you design a fanned-out pipeline architecture where multiple parallel runs attempt to consume from or write to a single shared queue, the runs will block or throw concurrency errors. If you need to distribute workloads, you must partition your inputs into separate runs with individual queues.
• Historical Data Inconsistencies: Although the underlying MLB Stats API supports historical seasons going back as far as 1876, structural changes in baseball tracking systems mean older seasons lack the detailed stat columns (like On-base plus slugging or WHIP) that are standard for modern players. Downstream pandas parsers must be designed defensively using techniques like .get() or check-for-null calls to prevent missing-key exceptions.
• Synchronous Execution Constraints: Any pipeline using synchronous calls must complete within 300 seconds. If an analytical query requires pulling player stats for multiple teams across multiple seasons in a single execution, it will hit the HTTP 408 timeout. Your orchestration logic must split massive historical retrievals into smaller, asynchronous batches.
• Platform Schedule Execution Safeguards: Platform schedules use a standard 6-field cron format where the seconds field is optional, and the minimum interval is 10 seconds. However, you cannot schedule a newly created Actor task until it has successfully run at least once manually. Additionally, any newly created schedule is disabled by default. If your deployment scripts configure pipelines programmatically via the API, they must explicitly toggle the schedule to active and ensure a baseline execution has completed.

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