Design a Data Pipeline to Process Logs from Web Servers (AWS, PySpark, Snowflake)

Processing web server logs (such as those from Nginx or Apache) is a classic data engineering problem. Logs are continuously generated and contain valuable information about user behavior, system health, and potential security threats.

This document outlines a modern, scalable architecture specifically using AWS for infrastructure, PySpark for data processing, and Snowflake as the cloud data warehouse.

1. High-Level Architecture Diagram

flowchart LR
    subgraph "Data Generation"
        A1["EC2 Web Server 1"]
        A2["EC2 Web Server 2"]
    end

    subgraph "AWS Ingestion Layer"
        B1["Kinesis Agent"]
        C1["Amazon Kinesis Data Streams"]
    end

    subgraph "Processing Layer (PySpark)"
        D1["Amazon EMR / AWS Glue<br>(PySpark Streaming)"]
    end

    subgraph "Data Lake Storage"
        E1[("Amazon S3 - Raw Bucket")]
        E2[("Amazon S3 - Processed Bucket")]
    end

    subgraph "Data Warehouse (Snowflake)"
        F1["Snowpipe"]
        F2[("Snowflake Tables")]
    end

    subgraph "Consumption"
        G1["BI Dashboards<br>(Tableau/Looker)"]
    end

    A1 -->|Logs| B1
    A2 -->|Logs| B1
    B1 -->|Push| C1

    C1 -->|Consume| D1

    D1 -->|Write Raw JSON| E1
    D1 -->|Write Clean Parquet| E2

    E2 -->|S3 Event Notification| F1
    F1 -->|Auto-Ingest| F2

    F2 --> G1

2. Component Breakdown

Step 1: Data Generation & Ingestion (AWS Kinesis)

Web servers running on Amazon EC2 instances generate continuous log files.

• Tool: Amazon Kinesis Agent.
• Function: Installed directly on the EC2 web servers. It continuously monitors the log files (e.g., /var/log/nginx/access.log), handles log rotation, and reliably pushes the data to Kinesis Data Streams.
• Message Broker: Amazon Kinesis Data Streams acts as the high-throughput buffering layer, decoupling the web servers from the processing engine and preventing data loss during traffic spikes.

Step 2: Stream Processing (PySpark on EMR/Glue)

This is the core compute layer where the raw log text is transformed into structured, actionable data.

• Tool: PySpark Structured Streaming running on Amazon EMR or AWS Glue.
• Operations Performed:
  • Read Stream: PySpark continuously reads micro-batches from the Kinesis Data Stream.
  • Parsing: Extracts fields using regex or JSON parsing (e.g., IP address, timestamp, HTTP method, request URL, status code).
  • Enrichment: Maps IP addresses to Geolocation data (Country, City) or parses the User-Agent string to determine the device type and browser.
  • Filtering: Drops malformed records or filters out known bot IP addresses.
  • Write Stream: The processed DataFrames are written out to the storage layer.

Step 3: Data Lake Storage (Amazon S3)

The PySpark application writes data to Amazon S3, dividing it into different zones:

• Raw Zone (Bronze): A raw backup of the logs (often compressed JSON or text) in case reprocessing is needed.
• Processed Zone (Silver): PySpark writes the parsed, enriched data into S3 as Parquet files. Parquet is a columnar storage format that is highly compressed and optimized for analytical querying. The data is partitioned by year/month/day/hour.

Step 4: Data Warehousing (Snowflake & Snowpipe)

Once the data lands in S3, it needs to be made available for high-performance querying and analytics.

• Tool: Snowflake (Cloud Data Warehouse) and Snowpipe (Continuous Data Ingestion Service).
• Workflow:
  1. When PySpark writes a new Parquet file to the Processed S3 Bucket, an S3 Event Notification (via Amazon SQS) is triggered.
  2. This notification tells Snowpipe that a new file is ready.
  3. Snowpipe automatically wakes up and ingests the Parquet file into the designated Snowflake tables without requiring manual COPY INTO commands.
• Benefits: This creates a near real-time pipeline. As soon as PySpark finishes processing a micro-batch and writes it to S3, it is available in Snowflake within minutes.

Step 5: Data Consumption

• Data Analysts & Business Intelligence: Analysts use tools like Tableau, Power BI, or Looker connected directly to Snowflake to build dashboards tracking daily active users, error rates, geographic traffic distribution, and conversion metrics.
• Performance: Snowflake's scalable compute warehouses handle concurrent dashboard queries efficiently, separating storage from compute.

3. Key Pipeline Considerations

• Fault Tolerance:
  • Kinesis stores data across multiple Availability Zones for up to 365 days.
  • PySpark uses checkpointing to S3 to track its offset in the Kinesis stream. If the EMR cluster crashes, a new cluster can resume exactly where the last one left off (Exactly-Once processing semantics).
• Schema Evolution: If the log format changes, PySpark handles schema validation. Snowflake handles schema evolution easily, allowing you to add new columns to the target tables as new fields appear in the logs.
• Cost Optimization: Storing the bulk of historical data in S3 (Data Lake) is cost-effective. Snowflake is only used for the hot/warm data needed for active analytics, taking advantage of its auto-suspend feature to save compute costs when no queries are running.

4. Code Snippets

PySpark: Read from Kinesis and Write to S3

Here is a simplified example of using PySpark Structured Streaming to read JSON logs from Kinesis, parse them, and write Parquet files to S3.

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StructField, StringType, TimestampType, IntegerType

# Initialize Spark Session
spark = SparkSession.builder \
    .appName("WebServerLogProcessor") \
    .getOrCreate()

# Define schema for the incoming JSON logs
log_schema = StructType([
    StructField("ip_address", StringType(), True),
    StructField("timestamp", TimestampType(), True),
    StructField("method", StringType(), True),
    StructField("endpoint", StringType(), True),
    StructField("status_code", IntegerType(), True),
    StructField("user_agent", StringType(), True)
])

# Read stream from Amazon Kinesis
kinesis_df = spark.readStream \
    .format("kinesis") \
    .option("streamName", "web-server-logs-stream") \
    .option("region", "us-east-1") \
    .option("initialPosition", "LATEST") \
    .load()

# The data from Kinesis comes as a binary 'data' column. We cast it to string and parse the JSON.
parsed_df = kinesis_df \
    .selectExpr("CAST(data AS STRING)") \
    .select(from_json(col("data"), log_schema).alias("logs")) \
    .select("logs.*")

# Filter out successful health checks
filtered_df = parsed_df.filter(~( (col("endpoint") == "/health") & (col("status_code") == 200) ))

# Write stream to Amazon S3 in Parquet format
query = filtered_df.writeStream \
    .format("parquet") \
    .option("path", "s3://my-company-data-lake/processed-logs/web-logs/") \
    .option("checkpointLocation", "s3://my-company-data-lake/checkpoints/web-logs/") \
    .trigger(processingTime="1 minute") \
    .start()

query.awaitTermination()

Snowflake: Setup Snowpipe for Auto-Ingestion

This SQL script sets up an external stage pointing to the S3 bucket, creates the destination table, and configures Snowpipe to automatically load new Parquet files.

-- 1. Create a database and schema
CREATE DATABASE web_analytics_db;
CREATE SCHEMA logs;

-- 2. Create the target table in Snowflake
CREATE OR REPLACE TABLE web_analytics_db.logs.web_logs (
    ip_address STRING,
    log_timestamp TIMESTAMP,
    http_method STRING,
    endpoint STRING,
    status_code NUMBER,
    user_agent STRING
);

-- 3. Create an integration to securely connect Snowflake to S3
CREATE STORAGE INTEGRATION s3_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::123456789012:role/snowflake_role'
  STORAGE_ALLOWED_LOCATIONS = ('s3://my-company-data-lake/processed-logs/web-logs/');

-- 4. Create a File Format for Parquet
CREATE OR REPLACE FILE FORMAT my_parquet_format
  TYPE = PARQUET;

-- 5. Create an External Stage
CREATE OR REPLACE STAGE my_s3_stage
  STORAGE_INTEGRATION = s3_int
  URL = 's3://my-company-data-lake/processed-logs/web-logs/'
  FILE_FORMAT = my_parquet_format;

-- 6. Create the Snowpipe
-- This pipe continuously loads data from the external stage into the table.
-- You would configure SQS Event Notifications on your S3 bucket to trigger this pipe.
CREATE OR REPLACE PIPE web_analytics_db.logs.web_logs_pipe
  AUTO_INGEST = TRUE
AS
  COPY INTO web_analytics_db.logs.web_logs
  FROM @my_s3_stage
  MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;