Building an Event-Driven ETL Data Pipeline with EMR Serverless

Build an automated, serverless weather data pipeline using S3, Lambda, and EMR Serverless. A new CSV upload triggers a PySpark job that cleans, transforms, and writes the processed data back to S3. All with scalable and managed compute.
Ghita

Software engineer | Data engineer | 8x AWS Certified
Published Sep 2, 2026
Last modified Sep 2, 2026
IntroductionEvery day, weather stations, IoT sensors, and external data providers generate CSV files containing temperature, humidity, wind, pressure, rainfall, and other measurements. Once these files land in S3, the workflow sounds straightforward: read the data, clean it, transform it, and make it available for downstream use.
The challenge is everything around that simple workflow.
You could keep EC2 instances or an EMR cluster running continuously, waiting for the next file to arrive. Or you could schedule a batch job to run every hour, even when there is nothing new to process. Both approaches come with trade-offs: unnecessary compute costs, infrastructure to manage, and a delay between the moment data arrives and the moment processing starts.
In this project, we’ll take a different approach: an event-driven, serverless ETL pipeline built with Amazon EMR Serverless and PySpark.
When a new weather CSV lands in S3, the event triggers a Lambda function that submits a PySpark job to EMR Serverless. The job cleans and transforms the data, writes the results back to S3, and then the compute resources can scale back down.
We’ll also use EMR Serverless Pre-Initialized Capacity to keep a small amount of capacity ready, reducing the startup latency you would normally expect from a serverless Spark workload. The infrastructure will be deployed using AWS CDK with Python.
In this article, we’ll build the pipeline step by step:
Step 1: Create the S3 Bucket
Step 2: Configure the S3 Event Notification
Step 3: Create the Lambda Function
Step 4: Create the EMR Serverless Application
Step 5: Create the PySpark ETL Job
Step 6: End-to-end testing
The complete project is available on GitHub.  HERE 
﻿
ArchitectureThe AWS solution we are building provides:
Operational Excellence: The pipeline is fully automated and event-driven, minimizing manual intervention and simplifying operations.
Performance Efficiency: EMR Serverless Pre-Initialized Capacity keeps baseline compute ready to process incoming jobs, reducing startup latency for frequent workloads.
Cost Optimization: The architecture eliminates the need to run traditional EMR clusters or EC2 instances continuously. A small pre-initialized capacity handles baseline workloads, while additional resources scale based on demand. S3 Lifecycle Rules can also be applied to prefixes to automate data retention and reduce storage costs.
Here’s an architecture diagram of what we’ll implement:
﻿
﻿
StepsStep 1: Create the S3 BucketThe first step is to create an S3 bucket that will act as the landing zone, output store, and script/log repository for the whole pipeline, organized into six key prefixes:
weather-data/raw/: Where new weather CSV files land and wait to be processed.
weather-data/processed/<input-file>/: where cleaned CSV output is written, partitioned by year/month/day, under a sub-folder named after the input file so each output traces back to its source upload.
weather-data/rejected/<input-file>/: where invalid rows are quarantined as CSV, tagged with a reason, under the same input-file-named sub-folder.
weather-data/scripts/: where the PySpark job itself lives, so EMR Serverless can find and execute it.
weather-data/logs/: the destination for EMR Serverless driver/executor logs.
weather-data/status/: one JSON job-status recap object per input file, written directly by the Spark job on completion. This is the final notification artifact in the whole pipeline.
We’ll also configure a prefix-scoped lifecycle rules:
raw files transition to Glacier after 90 days (they’re rarely re-read once processed),
rejected records expire after 180 days (useful for a data-quality audit window, but not worth keeping forever).
This keeps temporary or invalid data from accumulating indefinitely while allowing processed data, scripts, and logs to follow their own retention policies.
Also note that the Server-side encryption, versioning, and Block Public Access are enabled bucket-wide as a baseline security posture.
> CDK (Python):
self.data_bucket = s3.Bucket(
    self,
    "WeatherDataBucket",
    bucket_name=None,
    block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
    encryption=s3.BucketEncryption.S3_MANAGED,
    versioned=True,
    removal_policy=RemovalPolicy.DESTROY, 
    lifecycle_rules=[
        # Only the raw/ prefix ages out to Glacier -- processed/rejected/
        # scripts/logs/status are left untouched by this rule because
        # the prefix filter scopes it to weather-data/raw/ only.
        s3.LifecycleRule(
            id="ArchiveRawAfter90Days",
            enabled=True,
            prefix=RAW_PREFIX,
            transitions=[
                s3.Transition(
                    storage_class=s3.StorageClass.GLACIER,
                    transition_after=Duration.days(90),
                )
            ],
        ),
        # Only the rejected/ prefix expires after 180 days.
        s3.LifecycleRule(
            id="ExpireRejectedAfter180Days",
            enabled=True,
            prefix=REJECTED_PREFIX,
            expiration=Duration.days(180),
        ),
    ],
)
Tags.of(self.data_bucket).add("pipeline", "weather-etl")
> Result:
﻿
﻿
Step 2: Configure the S3 Event NotificationWe configure an S3 NotificationConfiguration directly on the data bucket that fires on s3:ObjectCreated:*, filtered down to files landing under the weather-data/raw/ prefix with a .csv suffix, and points straight at the Lambda function as its target. With a matching resource-based Lambda permission (Principal: s3.amazonaws.com) that grants S3 the right to invoke it.
> CDK (Python):
self.data_bucket.add_event_notification(
    s3.EventType.OBJECT_CREATED,
    s3n.LambdaDestination(self.start_emr_job_fn),
    s3.NotificationKeyFilter(prefix=RAW_PREFIX, suffix=".csv"),
)
﻿
> Result:
﻿
﻿
Step 3: Create the Lambda FunctionThe next component is a lightweight AWS Lambda function responsible for starting the EMR Serverless ETL job.
When a new weather CSV lands in S3, the Lambda function follows a simple sequence:
Validate the event (correct prefix, .csv suffix) and skip anything that doesn't look like a real upload, without retrying or erroring.
Generate a unique run id (a fresh UUID per invocation) used only for logging and correlation.
Start exactly one job, with the input path as an explicit argument.
The Lambda remains intentionally lightweight: it validates the event, prepares the job configuration, and submits the Spark workload. The actual data processing happens entirely in EMR Serverless.
> CDK (Python):
lambda_log_group = logs.LogGroup(
    self,
    "StartEmrJobLambdaLogGroup",
    log_group_name="/aws/lambda/weather-start-emr-job",
    retention=logs.RetentionDays.ONE_MONTH,
    removal_policy=RemovalPolicy.DESTROY,
)

self.lambda_role = iam.Role(
    self,
    "StartEmrJobLambdaRole",
    assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
    description="Least-privilege execution role for the start_emr_job Lambda function",
)
lambda_log_group.grant_write(self.lambda_role)
self.lambda_role.add_to_policy(
    iam.PolicyStatement(
        sid="StartAndTrackEmrServerlessJob",
        effect=iam.Effect.ALLOW,
        actions=[
            "emr-serverless:StartJobRun",
            "emr-serverless:TagResource",
            "emr-serverless:GetJobRun",
        ],
        resources=[
            f"arn:aws:emr-serverless:{self.region}:{self.account}:/applications/{self.emr_application.attr_application_id}*"
        ],
    )
)
self.lambda_role.add_to_policy(
    iam.PolicyStatement(
        sid="PassEmrJobExecutionRoleOnly",
        effect=iam.Effect.ALLOW,
        actions=["iam:PassRole"],
        resources=[self.emr_job_role.role_arn],
        conditions={"StringEquals": {"iam:PassedToService": "emr-serverless.amazonaws.com"}},
    )
)

self.start_emr_job_fn = _lambda.Function(
    self,
    "StartEmrJobFunction",
    function_name="weather-start-emr-job",
    runtime=_lambda.Runtime.PYTHON_3_12,
    handler="start_emr_job.handler",
    code=_lambda.Code.from_asset("../lambda"),
    role=self.lambda_role,
    timeout=Duration.seconds(30),
    memory_size=256,
    log_group=lambda_log_group,
    environment={
        "EMR_APPLICATION_ID": self.emr_application.attr_application_id,
        "EMR_EXECUTION_ROLE_ARN": self.emr_job_role.role_arn,
        "SPARK_SCRIPT_S3_PATH": spark_script_s3_path,
        "DATA_BUCKET": self.data_bucket.bucket_name,
        "RAW_PREFIX": RAW_PREFIX,
        "PROCESSED_PREFIX": PROCESSED_PREFIX,
        "REJECTED_PREFIX": REJECTED_PREFIX,
        "LOGS_PREFIX": LOGS_PREFIX,
        "STATUS_PREFIX": STATUS_PREFIX,
        "LOG_LEVEL": "INFO",
    },
)
self.start_emr_job_fn.node.add_dependency(script_deployment)
> Result:
﻿
﻿
Step 4: Create the EMR Serverless ApplicationNow we’ll create the Amazon EMR Serverless application (Spark runtime, emr-7.13.0 release) that will run our PySpark ETL jobs.
This is where we configure two important aspects of the architecture:
Pre-Initialized Capacity: keeps a baseline of compute ready to run jobs and helps reduce startup latency.
Autoscaling: allows EMR Serverless to provision additional capacity when the workload increases.
For this project, we’ll configure the app with a Pre-Initialized Capacity to one standing driver and one standing executor kept warm at all times, with autoscaling enabled up to 2 executors. The workers are deliberately small (2 vCPU each) and the executor ceiling is low so peak concurrent usage stays around 6 vCPU. Comfortably under a low account-level “Max concurrent vCPUs” service quota to avoid ServiceQuotaExceededException.
NOTE: 
Amazon EMR versions 7.0, 7.1, and 7.2 reach End of Support on July 31, 2026. Check the Amazon EMR Serverless release versions:  here 
> EMR Serverless Configuration Summary:
Configuration
Value
Release label
EMR 7.13.0
Application type
Spark
Pre-Initialized Capacity
1 Driver + 1 Executor
Driver (pre-init)
1 × 2 vCPU, 8 GB RAM, 20 GB disk
Executor (pre-init)
1 × 2 vCPU, 8 GB RAM, 20 GB disk
Auto Scaling
Enabled
Max capacity (executors)
16 vCPU, 64 GB RAM, 200 GB disk
Idle timeout
15 minutes
> CDK (Python):
self.emr_application = emrserverless.CfnApplication(
    self,
    "WeatherEmrServerlessApp",
    name="weather-etl-application",
    release_label="emr-7.13.0",
    type="SPARK",
    auto_start_configuration=emrserverless.CfnApplication.AutoStartConfigurationProperty(
        enabled=True
    ),
    auto_stop_configuration=emrserverless.CfnApplication.AutoStopConfigurationProperty(
        enabled=True, idle_timeout_minutes=15
    ),
    # --- Pre-Initialized Capacity: keeps 1 driver + 1 executor warm
    #     at all times, eliminating cold-start latency for the common
    #     case of a single small file upload. Small 2 vCPU workers to
    #     stay under a low concurrent-vCPU account quota. ---
    initial_capacity=[
        emrserverless.CfnApplication.InitialCapacityConfigKeyValuePairProperty(
            key="Driver",
            value=emrserverless.CfnApplication.InitialCapacityConfigProperty(
                worker_count=1,
                worker_configuration=emrserverless.CfnApplication.WorkerConfigurationProperty(
                    cpu="2vCPU", memory="8GB", disk="20GB"
                ),
            ),
        ),
        emrserverless.CfnApplication.InitialCapacityConfigKeyValuePairProperty(
            key="Executor",
            value=emrserverless.CfnApplication.InitialCapacityConfigProperty(
                worker_count=1,
                worker_configuration=emrserverless.CfnApplication.WorkerConfigurationProperty(
                    cpu="2vCPU", memory="8GB", disk="20GB"
                ),
            ),
        ),
    ],
    # --- Ceiling for autoscaling beyond the pre-initialized pool.
    #     Kept low to respect the account's concurrent-vCPU quota. ---
    maximum_capacity=emrserverless.CfnApplication.MaximumAllowedResourcesProperty(
        cpu="16vCPU",
        memory="64GB",
        disk="200GB",
    ),
    tags=[CfnTag(key="pipeline", value="weather-etl")],
)
> Result:
﻿
﻿
Step 5: Create the PySpark ETL JobOnce Lambda submits the EMR Serverless job, the PySpark application processes the single CSV file through the following steps:
Read the one CSV file.
Standardize types: trim whitespace, cast the timestamp string to a real TimestampType.
Deduplicate exact-duplicate rows and duplicate sensor readings (same station, same timestamp).
Split off rows missing required fields (station ID, timestamp, or any core measurement) into a "missing data" rejection bucket.
Validate physical plausibility on everything else: temperature between -60°C and 60°C, humidity between 0–100%, non-negative wind speed, pressure between 800–1100 hPa, non-negative rainfall. Anything outside those bounds gets tagged with a specific rejection_reason and routed to the rejected output instead of failing the job.
Derive partition columns (year, month, day, hour) from the timestamp for the valid data.
Write the valid rows as date-partitioned header CSV, and the rejected rows as CSV with their rejection reasons attached. Both under a sub-folder named after the input file, so every output object is traceable back to the upload that produced it.
> CDK (Python):
# Upload the PySpark ETL script to the data bucket at deploy time
script_deployment = s3deploy.BucketDeployment(
    self,
    "DeploySparkScript",
    sources=[s3deploy.Source.asset("../spark")],
    destination_bucket=self.data_bucket,
    destination_key_prefix=SCRIPTS_PREFIX,
    prune=False,
)
spark_script_s3_path = f"s3://{self.data_bucket.bucket_name}/{SCRIPTS_PREFIX}etl_job.py"
> Result:
﻿
﻿
Step 6: End-to-end testingThe end-to-end test validates the complete data processing workflow, from uploading the input CSV file to S3 through Lambda-triggered EMR processing and generation of the output files.
uploading the csv file to S3:** weather-data/raw/weather_20260726.csv
station_id,timestamp,temperature,humidity,wind_speed,pressure,rainfall
ST001,2026-07-26T00:00:00,22.5,65,3.2,1013.2,0.0
ST002,2026-07-26T00:00:00,19.8,70,2.1,1011.5,0.4
ST001,2026-07-26T00:00:00,22.5,65,3.2,1013.2,0.0
ST003,2026-07-26T00:05:00,,58,1.8,1009.8,0.0
ST004,2026-07-26T00:05:00,21.0,,2.5,1012.0,0.1
ST005,2026-07-26T00:10:00,85.0,60,2.0,1010.0,0.0
ST006,2026-07-26T00:10:00,18.4,105,1.5,1008.5,0.2
ST007,2026-07-26T00:15:00,17.2,72,-1.5,1011.0,0.0
ST008,2026-07-26T00:15:00,20.1,55,2.8,750.0,0.0
ST009,2026-07-26T00:20:00,16.5,80,3.0,1250.0,0.0
,2026-07-26T00:20:00,15.0,66,2.2,1009.0,0.3
ST010,,19.5,63,2.6,1010.5,0.0
ST011,2026-07-26T00:25:00,-25.0,40,4.1,1015.0,0.0
ST012,2026-07-26T00:25:00,23.0,68,1.9,1013.8,-0.5
ST013,2026-07-26T00:30:00,24.1,71,2.4,1012.9,0.0
ST013,2026-07-26T00:30:00,24.1,71,2.4,1012.9,0.0
ST014,2026-07-26T00:35:00,22.9,69,3.5,1011.7,1.2
ST015,2026-07-26T00:35:00,-65.0,50,2.0,1009.5,0.0
ST016,2026-07-26T00:40:00,21.5,62,2.1,1013.0,0.0
ST017,2026-07-26T00:40:00,20.8,64,1.7,1012.4,0.0
Lambda logs confirm the start of the processing:
﻿
A new job is submitted to EMR
﻿
 Job completed successfully:
﻿
S3 outputs review: 
﻿
job run recap file: weather-data/status/weather_20260726.json
{
  "status": "SUCCESS",
  "subject": "[weather-etl] SUCCESS: weather_20260726.csv",
  "message": "Weather ETL job SUCCEEDED\n----------------------------------------\nInput file : weather_20260726.csv\nSource     : s3://weatherpipelinestack-weatherdatabucket5fce382e-jrafmsqoopgl/weather-data/raw/weather_20260726.csv\nJob run id : weather-etl-ce304084cb4e41909067062ca0c86cf1\n\nResults by target:\n  processed -> s3://weatherpipelinestack-weatherdatabucket5fce382e-jrafmsqoopgl/weather-data/processed/weather_20260726\n               7 rows\n  rejected  -> s3://weatherpipelinestack-weatherdatabucket5fce382e-jrafmsqoopgl/weather-data/rejected/weather_20260726\n               11 rows\n\nRows read (raw)     : 20\nDuplicates removed  : 2\nProcessing duration : 40.72 s\n",
  "input_file_stem": "weather_20260726",
  "job_run_id": "weather-etl-ce304084cb4e41909067062ca0c86cf1",
  "written_at": "2026-08-26T23:10:52.678573+00:00",
  "raw_count": 20,
  "valid_count": 7,
  "rejected_count": 11,
  "duration_s": 40.72
}
﻿
ConclusionThis solution demonstrates how Amazon S3, AWS Lambda, and Amazon EMR Serverless can be combined to build a fully automated and scalable weather data processing pipeline.
Instead of manually starting compute resources or managing Spark clusters, the workflow reacts to new data as it arrives. A CSV uploaded to the S3 folder triggers a Lambda function, which submits a PySpark job to EMR Serverless. The job then cleans and transforms the weather data before writing the processed results back to S3.
The use of EMR Serverless removes much of the infrastructure management overhead while allowing compute capacity to scale based on workload. With pre-initialized capacity, small workloads can also benefit from reduced startup latency, while auto-stop helps avoid keeping resources running when they are no longer needed.
Overall, this event-driven architecture provides a simple, scalable, and cost-conscious approach to weather data processing.
﻿
ReferencesWhat is AWS Lambda?
 https://docs.aws.amazon.com/lambda/latest/dg/welcome.html 
Create your first Lambda function
 https://docs.aws.amazon.com/lambda/latest/dg/getting-started.html 
What is Event-Driven Architecture?
 https://aws.amazon.com/event-driven-architecture/ 
What is Amazon EMR Serverless?
 https://docs.aws.amazon.com/emr/latest/EMR-Serverless-UserGuide/emr-serverless.html 
EMR Serverless: Running jobs
 https://docs.aws.amazon.com/emr/latest/EMR-Serverless-UserGuide/jobs.html 
Manage applications from the EMR Studio console
 https://docs.aws.amazon.com/emr/latest/EMR-Serverless-UserGuide/studio-manage-app.html 
Configuration	Value
Release label	EMR 7.13.0
Application type	Spark
Pre-Initialized Capacity	1 Driver + 1 Executor
Driver (pre-init)	1 × 2 vCPU, 8 GB RAM, 20 GB disk
Executor (pre-init)	1 × 2 vCPU, 8 GB RAM, 20 GB disk
Auto Scaling	Enabled
Max capacity (executors)	16 vCPU, 64 GB RAM, 200 GB disk
Idle timeout	15 minutes