AWS Lambda provides serverless computing capabilities, and it can be used for performing validation or light processing/transformation of data. Moreover, with its integration with more than 140 AWS services, it facilitates building complex systems employing event-driven architectures. There are many ways to build serverless applications and one of the most efficient ways is using specialised frameworks such as the AWS Serverless Application Model (SAM) and Serverless Framework. In this post, I’ll demonstrate how to build a serverless data processing application using SAM.


When we create an application or pipeline with AWS Lambda, most likely we’ll include its event triggers and destinations. The AWS Serverless Application Model (SAM) facilitates building serverless applications by providing shorthand syntax with a number of custom resource types. Also, the AWS SAM CLI supports an execution environment that helps build, test, debug and deploy applications easily. Furthermore, the CLI can be integrated with full-pledged IaC tools such as the AWS Cloud Development Kit (CDK) and Terraform - note integration with the latter is in its roadmap. With the integration, serverless application development can be a lot easier with capabilities of local testing and building. An alternative tool is the Serverless Framework. It supports multiple cloud providers and broader event sources out-of-box but its integration with IaC tools is practically non-existent.

In this post, we’ll build a simple data pipeline using SAM where a Lambda function is triggered when an object (CSV file) is created in a S3 bucket. The Lambda function converts the object into parquet and AVRO files and saves to a destination S3 bucket. For simplicity, we’ll use a single bucket for the source and destination.

SAM Application

After installing the SAM CLI, I initialised an app with the Python 3.8 Lambda runtime from the hello world template (sam init --runtime python3.8). Then it is modified for the data pipeline app. The application is defined in the template.yaml and the source of the main Lambda function is placed in the _transform _folder. We need 3rd party packages for converting source files into the parquet and AVRO formats - AWS Data Wrangler and fastavro. Instead of packaging them together with the Lambda function, they are made available as Lambda layers. While using the AWS managed Lambda layer for the former, we only need to build the Lambda layer for the _fastavro _package, and it is located in the _fastavro _folder. The source of the app can be found in the GitHub repository of this post.

 2└── requirements.txt
 6└── requirements.txt
 9└── unit
10    ├──
11    └──

In the resources section of the template, the Lambda layer for AVRO transformation (FastAvro), the main Lambda function (TransformFunction) and the source (and destination) S3 bucket (SourceBucket) are added. The layer can be built simply by adding the pip package name to the requirements.txt file. It is set to be compatible with Python 3.7 to 3.9. For the Lambda function, its source is configured to be built from the _transform _folder and the ARNs of the custom and AWS managed Lambda layers are added to the layers property. Also, an S3 bucket event is configured so that this Lambda function is triggered whenever a new object is created to the bucket. Finally, as it needs to have permission to read and write objects to the S3 bucket, its invocation policies are added from ready-made policy templates - _S3ReadPolicy _and S3WritePolicy.

 1# template.yaml
 2AWSTemplateFormatVersion: "2010-09-09"
 3Transform: AWS::Serverless-2016-10-31
 4Description: >
 5  sam-for-data-professionals
 7  Sample SAM Template for sam-for-data-professionals  
10  Function:
11    MemorySize: 256
12    Timeout: 20
15  FastAvro:
16    Type: AWS::Serverless::LayerVersion
17    Properties:
18      LayerName: fastavro-layer-py3
19      ContentUri: fastavro/
20      CompatibleRuntimes:
21        - python3.7
22        - python3.8
23        - python3.9
24    Metadata:
25      BuildMethod: python3.8
26  TransformFunction:
27    Type: AWS::Serverless::Function
28    Properties:
29      CodeUri: transform/
30      Handler: app.lambda_handler
31      Runtime: python3.8
32      Layers:
33        - !Ref FastAvro
34        - arn:aws:lambda:ap-southeast-2:336392948345:layer:AWSDataWrangler-Python38:8
35      Policies:
36        - S3ReadPolicy:
37            BucketName: sam-for-data-professionals-cevo
38        - S3WritePolicy:
39            BucketName: sam-for-data-professionals-cevo
40      Events:
41        BucketEvent:
42          Type: S3
43          Properties:
44            Bucket: !Ref SourceBucket
45            Events:
46              - "s3:ObjectCreated:*"
47  SourceBucket:
48    Type: AWS::S3::Bucket
49    Properties:
50      BucketName: sam-for-data-professionals-cevo
53  FastAvro:
54    Description: "ARN of fastavro-layer-py3"
55    Value: !Ref FastAvro
56  TransformFunction:
57    Description: "Transform Lambda Function ARN"
58    Value: !GetAtt TransformFunction.Arn

Lambda Function

The transform function reads an input file from the S3 bucket and saves the records as the parquet and AVRO formats. Thanks to the Lambda layers, we can access the necessary 3rd party packages as well as reduce the size of uploaded deployment packages and make it faster to deploy it.

 1# transform/
 2import re
 3import io
 4from fastavro import writer, parse_schema
 5import awswrangler as wr
 6import pandas as pd
 7import boto3
 9s3 = boto3.client("s3")
11avro_schema = {
12    "doc": "User details",
13    "name": "User",
14    "namespace": "user",
15    "type": "record",
16    "fields": [{"name": "name", "type": "string"}, {"name": "age", "type": "int"}],
20def check_fields(df: pd.DataFrame, schema: dict):
21    if schema.get("fields") is None:
22        raise Exception("missing fields in schema keys")
23    if len(set(df.columns) - set([f["name"] for f in schema["fields"]])) > 0:
24        raise Exception("missing columns in schema key of fields")
27def check_data_types(df: pd.DataFrame, schema: dict):
28    dtypes = df.dtypes.to_dict()
29    for field in schema["fields"]:
30        match_type = "object" if field["type"] == "string" else field["type"]
31        if, str(dtypes[field["name"]])) is None:
32            raise Exception(f"incorrect column type - {field['name']}")
35def generate_avro_file(df: pd.DataFrame, schema: dict):
36    check_fields(df, schema)
37    check_data_types(df, schema)
38    buffer = io.BytesIO()
39    writer(buffer, parse_schema(schema), df.to_dict("records"))
41    return buffer
44def lambda_handler(event, context):
45    # get bucket and key values
46    record = next(iter(event["Records"]))
47    bucket = record["s3"]["bucket"]["name"]
48    key = record["s3"]["object"]["key"]
49    file_name = re.sub(".csv$", "", key.split("/")[-1])
50    # read input csv as a data frame
51    input_path = f"s3://{bucket}/{key}"
52    input_df = wr.s3.read_csv([input_path])
53    # write to s3 as a parquet file
54    wr.s3.to_parquet(df=input_df, path=f"s3://{bucket}/output/{file_name}.parquet")
55    # write to s3 as an avro file
56    s3.upload_fileobj(generate_avro_file(input_df, avro_schema), bucket, f"output/{file_name}.avro")

Unit Testing

We use a custom function to create AVRO files (generate_avro_file) while relying on the AWS Data Wrangler package for reading input files and writing to parquet files. Therefore, unit testing is performed for the custom function only. Mainly it tests whether the AVRO schema matches the input data fields and data types.

 1# tests/unit/
 2import pytest
 3import pandas as pd
 4from transform import app
 8def input_df():
 9    return pd.DataFrame.from_dict({"name": ["Vrinda", "Tracy"], "age": [22, 28]})
12def test_generate_avro_file_success(input_df):
13    avro_schema = {
14        "doc": "User details",
15        "name": "User",
16        "namespace": "user",
17        "type": "record",
18        "fields": [{"name": "name", "type": "string"}, {"name": "age", "type": "int"}],
19    }
20    app.generate_avro_file(input_df, avro_schema)
21    assert True
24def test_generate_avro_file_fail_missing_fields(input_df):
25    avro_schema = {
26        "doc": "User details",
27        "name": "User",
28        "namespace": "user",
29        "type": "record",
30    }
31    with pytest.raises(Exception) as e:
32        app.generate_avro_file(input_df, avro_schema)
33    assert "missing fields in schema keys" == str(e.value)
36def test_generate_avro_file_fail_missing_columns(input_df):
37    avro_schema = {
38        "doc": "User details",
39        "name": "User",
40        "namespace": "user",
41        "type": "record",
42        "fields": [{"name": "name", "type": "string"}],
43    }
44    with pytest.raises(Exception) as e:
45        app.generate_avro_file(input_df, avro_schema)
46    assert "missing columns in schema key of fields" == str(e.value)
49def test_generate_avro_file_fail_incorrect_age_type(input_df):
50    avro_schema = {
51        "doc": "User details",
52        "name": "User",
53        "namespace": "user",
54        "type": "record",
55        "fields": [{"name": "name", "type": "string"}, {"name": "age", "type": "string"}],
56    }
57    with pytest.raises(Exception) as e:
58        app.generate_avro_file(input_df, avro_schema)
59    assert f"incorrect column type - age" == str(e.value)

Build and Deploy

The app has to be built before deployment. It can be done by sam build.

The deployment can be done with and without a guide. For the latter, we need to specify additional parameters such as the Cloudformation stack name, capabilities (as we create an IAM role for Lambda) and a flag to automatically determine an S3 bucket to store build artifacts.

1sam deploy \
2  --stack-name sam-for-data-professionals \
3  --capabilities CAPABILITY_IAM \
4  --resolve-s3

Trigger Lambda Function

We can simply trigger the Lambda function by uploading a source file to the S3 bucket. Once it is uploaded, we are able to see that the output parquet and AVRO files are saved as expected.

1$ aws s3 cp test.csv s3://sam-for-data-professionals-cevo/input/
2upload: ./test.csv to s3://sam-for-data-professionals-cevo/input/test.csv
4$ aws s3 ls s3://sam-for-data-professionals-cevo/output/
52022-07-17 17:33:21        403 test.avro
62022-07-17 17:33:21       2112 test.parquet


In this post, it is illustrated how to build a serverless data processing application using SAM. A Lambda function is developed, which is triggered whenever an object is created in a S3 bucket. It converts input CSV files into the parquet and AVRO formats before saving into the destination bucket. For the format conversion, it uses 3rd party packages, and they are made available by Lambda layers. The application is built and deployed and the function triggering is checked.