Data Engineering

Data Ingestion

F1

Building a Formula 1 Data Pipeline: The Data Collection Layer

TL;DR: This article explores how I built the data collection layer for my F1 pipeline. I implemented Python-based collectors with async programming, robust error handling, and efficient connection pooling. While I’ve built the code for a full historical load, I plan to use Airflow in the future to fetch data session-by-session for each Grand Prix weekend.

In my previous article, I introduced the architecture for my Formula 1 data pipeline project. As promised, I’m now diving deeper into each component, starting with the data collection layer — the foundation that everything else is built upon.

Racing Against Data Challenges

I started this project because I wanted a real-world challenge that would push my data engineering skills beyond the typical e-commerce datasets used in tutorials. Formula 1 data is particularly interesting because it combines structured time-series data (lap times, positions, weather readings) with semi-structured information (race control messages, team strategies).

When I began building the collection layer, I immediately faced several challenges:

• Mixed data sources — some available through APIs, others requiring web scraping
• Inconsistent data formats — especially timestamps and identifiers across sources
• Performance bottlenecks — synchronous API calls were too slow for bulk collection
• Network reliability — API timeouts and rate limits required careful handling

I’ll walk through how I tackled each of these challenges and share the hard-won lessons from my failures along the way.

Architecture: Starting Grid

The data collection components in my pipeline include:

Architecture

• API Integration with OpenF1, providing current season telemetry
• Web Scraper for historical race results from Formula1.com

I initially tried using a cloud-based API gateway, but quickly realized a simpler approach would be more maintainable for my project scope. The current architecture is lightweight but robust, perfect for a personal project that needs to run reliably without breaking the bank.

Data Sources: The Starting Line

OpenF1 API: Real-time Racing Data

The OpenF1 API is the primary data source, offering detailed timing, position, and status information. It’s organized into endpoint categories:

• Meeting and session data — metadata about race weekends
• Driver information — numbers, names, teams
• Timing data — lap times, sector splits
• Position data — track position updates
• Weather conditions — temperature, wind, rainfall

Here’s a look at how the API data is structured:

# Configuration for different endpoint types
SESSION_BASED_ENDPOINTS = ['laps', 'position', 'race_control', 'stints']
GENERAL_ENDPOINTS = ['meetings', 'sessions', 'drivers', 'weather']

While the API is well-designed overall, I encountered two significant challenges:

First, the API sometimes returns inconsistent types for the same field. For example, timestamps might appear as ISO-formatted strings in one response and Unix microsecond timestamps in another. I had to build robust type handling to normalize these inconsistencies.

Second, there’s no bulk download capability — you need to fetch data race by race, session by session. This led me to implement a parallel fetching strategy (more on that later).

Web Scraping: Historical Race Data

The OpenF1 API doesn’t provide historical race results, which I needed for longitudinal analysis. After exploring several options, I decided to scrape the official Formula1.com website.

Web scraping introduced its own set of challenges. The site uses dynamic JavaScript rendering, inconsistent HTML structures between seasons, and occasionally changes its layout. I built a scraper using Beautiful Soup that’s resilient to these changes:

def scrape_session_results(self, url: str, session_type: str) -> pd.DataFrame:
    """Scrapes results from F1 website and returns them as a DataFrame"""
    try:
        response = requests.get(url, headers=self.headers)
        soup = BeautifulSoup(response.text, 'html.parser')
        # Processing logic varies by session type...
        
        # For qualifying sessions we target specific table elements
        if session_type == 'qualifying':
            table = soup.find('table', class_='f1-table')
            # Extract qualifying times from each round (Q1, Q2, Q3)
            
        # For race results, the structure is different
        elif session_type == 'race':
            cells = soup.find_all('p', class_='f1-text')
            # Process position, driver, time and points data

The scraper extracts data for race results, qualifying sessions, practice sessions, and sprint races, then normalizes the data into a consistent format before storing it in the database.

Async Programming: Engine Upgrade

One of my first implementations used synchronous API calls, which performed terribly. A full data collection run took more than an hour — completely impractical for a real-time pipeline.

The Problem with Synchronous Calls

With synchronous processing, my code would:

1. Request data from endpoint A
2. Wait for response
3. Process and store data
4. Move to endpoint B
5. Repeat…

This sequential approach wasted enormous amounts of time just waiting for API responses.

Asynchronous Solution

Switching to asynchronous programming with asyncio and aiohttp was a game-changer. Here's a simplified version of the approach I took:

async def fetch_and_store_session_based_data(self, session, session_keys):
    """Fetch and store session-based F1 data asynchronously"""
    for session_key in session_keys:
        for endpoint in SESSION_BASED_ENDPOINTS:
            try:
                # Fetch data without holding a database connection
                params = {"session_key": session_key}
                data = await self.api_fetcher.fetch_data(session, endpoint, params)
                
                # Process and store data only if we received something
                if data:
                    self.db_ops.insert_data(endpoint, data, unique_columns)
            except Exception as e:
                self.logger.error(f"Error processing {endpoint}: {e}")

This change reduced collection time by over 60%, bringing a full data run down to about 30 minutes. The key innovations were:

1. Separating connection acquisition from query execution
2. Using asynchronous API requests to eliminate blocking waits
3. Processing multiple endpoints concurrently

I later optimized this further by implementing connection pooling and batch inserts, which brought the full collection time down to just a few minutes.

Error Handling: Pit Stop Strategy

Data collection pipelines fail — it’s not a question of if, but when. External services go down, networks timeout, and data formats change unexpectedly. Rather than fighting against this reality, I embraced it by designing for failure from the start.

My error handling approach has three core components:

1. Exponential backoff for retries on failure
2. Circuit breaker patterns to prevent cascading failures (stopping requests to endpoints that repeatedly fail)
3. Detailed error logging to quickly identify root causes

Here’s a simplified version of my retry logic:

# Retry with exponential backoff
for attempt in range(self.MAX_RETRIES):
    try:
        # Make API call here...
        response = await session.get(url, params=params)
        if response.status == 200:
            return await response.json()
        
        # Handle specific status codes (404, 429, etc.)
        
    except Exception as e:
        if attempt < self.MAX_RETRIES - 1:
            # Calculate longer wait time for each retry
            backoff_time = self.BACKOFF_FACTOR ** attempt
            await asyncio.sleep(backoff_time)
            continue
        else:
            self.logger.error(f"Failed after {self.MAX_RETRIES} attempts")
            raise

This approach has been remarkably effective. My pipeline now recovers automatically from most transient errors, and when permanent failures occur, the detailed logs make troubleshooting straightforward.

Data Validation: Technical Inspection

Data quality issues can silently corrupt your entire pipeline if not caught early. I implemented validation at multiple levels:

1. Schema validation — ensuring required fields exist
2. Type checking — confirming data types match expectations
3. Relationship validation — verifying foreign key integrity
4. Deduplication — preventing duplicate records

One of the most valuable patterns I implemented was a “validate and transform” approach that combines validation with normalization:

def transform_data(self, data, endpoint):
    """Transform and validate API data"""
    allowed_fields = self._get_allowed_fields(endpoint)
    column_mappings = self._get_column_mappings(endpoint)
    
    # Validate and transform each record
    transformed_data = []
    for record in data:
        # Check for required fields based on endpoint
        if not self._has_required_fields(record, endpoint):
            continue
            
        # Apply field mappings and type conversions
        transformed_record = {}
        for key, value in record.items():
            # Map field names to our schema
            new_key = column_mappings.get(key, key)
            if allowed_fields is None or new_key in allowed_fields:
                transformed_record[new_key] = self._convert_type(value, new_key)
        
        transformed_data.append(transformed_record)
    
    return transformed_data

This ensures data consistency before it enters the database, preventing a whole class of potential issues downstream.

Connection Management: Pit Crew

In my initial implementation, I opened a new database connection for each API endpoint. This quickly led to connection exhaustion as PostgreSQL has limits on concurrent connections.

The Connection Problem

PostgreSQL typically defaults to a maximum of 100 connections. With multiple endpoints and concurrent processing, I was quickly hitting this limit and seeing errors like:

FATAL: sorry, too many clients already

Connection Pool Solution

The solution was a connection pool manager:

def get_connection(self) -> pg_connection:
    """Get a connection from the pool"""
    if self._pool is None:
        self.initialize_pool()
    
    try:
        connection = self._pool.getconn()
        return connection
    except Exception as e:
        self.logger.error(f"Error getting connection: {e}")
        raise

Combined with a context manager pattern, this approach:

• Reuses connections efficiently
• Properly releases connections back to the pool
• Handles connection failures gracefully
• Limits the total number of concurrent connections

This single change dramatically improved the reliability of the data collection process and eliminated a whole class of intermittent failures.

Lessons from the Track

Building this data collection layer taught me several valuable lessons I’d like to share:

1. Separate Fetching from Processing

My initial design mixed data fetching, transformation, and storage in the same functions. This made error handling complex and debugging difficult.

I refactored to create a clean separation of concerns:

• APIFetcher handles only API communication and retries
• DataManager coordinates the overall process
• PostgresOperations manages database operations

This separation made the code more maintainable and easier to test.

2. Design for Idempotency

Idempotency means running the pipeline multiple times should produce the same result without duplicates or errors. I achieved this with:

• Unique constraints in the database schema
• ON CONFLICT clauses in INSERT statements
• Consistent primary keys across runs

This approach lets me safely rerun the pipeline after failures without manual cleanup.

3. Monitor Everything

Comprehensive logging was critical for understanding pipeline behavior:

2025-04-19 18:24:37,769 - F1_Pipeline - INFO - data_manager.py:53 - Processing position for session 9687
2025-04-19 18:24:37,769 - F1_Pipeline - INFO - fetcher.py:31 - Fetching data from https://api.openf1.org/v1/position with params {'session_key': 9687}
2025-04-19 18:24:38,488 - F1_Pipeline - INFO - operations.py:51 - Fetching existing keys from position...
2025-04-19 18:24:38,775 - F1_Pipeline - INFO - operations.py:58 - Fetched 176484 existing keys from position
2025-04-19 18:24:38,811 - F1_Pipeline - INFO - operations.py:149 - Inserted 486 records into position
2025-04-19 18:24:38,812 - F1_Pipeline - INFO - data_manager.py:81 - Inserted 486 new records for position (session_key: 9687)

I log not just errors but also performance metrics, record counts, and processing times. This data proved invaluable for identifying optimization opportunities and troubleshooting issues.

Beyond Full Loads: Planning for Incremental Updates

While the current implementation performs full data loads, this approach isn’t ideal for a production system. Processing the entire dataset repeatedly is inefficient and puts unnecessary load on both the API and our database.

The Challenge with Incremental F1 Data

For my next iteration, I plan to implement incremental loading patterns, but F1 data presents unique challenges:

1. Race corrections — Official results can change after penalties are applied
2. Timing adjustments — Lap times occasionally get revised after review
3. Session reclassifications — Drivers may be disqualified after technical inspection

To address these challenges, I’ll implement a simpler and more efficient approach:

Session-based incremental loads

# Pseudocode for session-based approach
# The API provides a way to get the latest session
latest_sessions = api_fetcher.fetch_data("sessions", {"latest": True})

for session in latest_sessions:
    session_key = session['session_key']
    
    # Check if we've already processed this session
    if not is_session_processed(session_key):
        # Fetch and process all data for this session
        process_new_session(session_key)

Full refreshes for specific sessions when needed (e.g., after penalties or corrections)

Airflow for Orchestration

This is where Airflow will become essential. I will build DAGs that will:

• Detect new F1 sessions each race weekend
• Trigger session-specific data pulls
• Run post-race validation and cleanup

With this approach, I’ll get the best of both worlds: near real-time data during race weekends and reliable, consistent data for historical analysis.

The Finish Line

After several iterations, the data collection layer now reliably processes:

• 2000+ race laps per Grand Prix weekend
• ~20,000 position updates per race
• Hundreds of weather readings and race control messages
• Historical results going back multiple seasons

The performance improvements have been substantial:

• Collection runs complete in minutes instead of hours
• Recovery from API failures happens automatically

The most important takeaway from building this layer was that designing for failure and focusing on resilience from the start pays enormous dividends in data engineering projects.

What’s Next on the Calendar

The data collection layer is just the first part of our Formula 1 data pipeline. In my next article, I’ll dive into the streaming layer, where I’ll use Debezium and Kafka to implement Change Data Capture (CDC) for real-time data processing.

We’ll explore:

• Setting up Debezium connectors for PostgreSQL
• Configuring Kafka topics and consumer groups
• Processing CDC events efficiently
• Stream processing patterns for F1 data

Until then, happy coding and enjoy the race!

I’d love to hear your thoughts — did I miss any pitfalls in handling F1’s tricky session corrections or connection pooling? If you’ve tackled similar data challenges (or spotted areas I can improve), Don’t hesitate to share it in the comments!