Creating Robust and Type-Safe Data Pipelines using strict-stream, JSONStream, and AWS Lambda

In today's data-centric landscape, the development of resilient and type-safe data pipelines has never been more essential. This article delves into the construction of such pipelines through the integration of the strict-stream and JSONStream libraries, coupled with AWS Lambda. This powerful combination empowers developers to establish data pipelines that ensure the integrity of data, adhere to type expectations, and harness the capabilities of AWS.

Prerequisites

Before we proceed, make sure you have the following prerequisites in place:

• An AWS Account
• Node.js installed
• AWS SDK for JavaScript (aws-sdk) installed (npm install aws-sdk)
• JSONStream and strict-stream libraries installed (npm install jsonstream-next strict-stream)

Setting Up AWS Services

Creating an S3 Bucket

1. Access the AWS Management Console.
2. Navigate to Amazon S3 and create a new bucket.
3. Upload a sample JSON file named sample.json to the bucket. Here's a snippet of the JSON content:

[
  {
    "id": 1,
    "name": "John",
    "age": 30
  },
  {
    "wrong": "record here"
  },
  {
    "id": 2,
    "name": "Jane",
    "age": 25
  }
]        

Creating an AWS Lambda Function

1. Go to the AWS Lambda service.
2. Create a new Lambda function from scratch.
3. Configure the function and replace the provided code in the "Function code" section with the AWS Lambda handler code.

Utilizing strict-stream, JSONStream, and AWS Lambda

Here is the TypeScript code for the AWS Lambda function that processes JSON data from an S3 bucket using the strict-stream and JSONStream libraries:

import AWS from 'aws-sdk';
import * as JSONStream from 'jsonstream-next';
import { of } from 'strict-stream';
import { nodeTransform } from 'strict-stream/nodeTransform';
import { filter } from 'strict-stream/filter';
import { map } from "strict-stream/map";
import { Readable } from 'stream';

// Initialize the AWS SDK
AWS.config.update({ region: 'your-region' }); // Replace with your AWS region
const s3 = new AWS.S3();

// Type definition for Customer
type Customer = {
  id: number;
  name: string;
  age: number;
};

// Validate function to ensure type safety
function validateCustomer(customer: Customer): customer is Customer {
  if (customer) {
    return 'name' in customer && 'age' in customer && customer.age >= 0;
  }
  return false;
}

type TODO = any;

// Lambda function handler
export const handler = async (event: TODO, context: TODO) => {
  const bucketName = 'your-bucket-name'; // Replace with your bucket name

  const getObjectParams: AWS.S3.GetObjectRequest = {
    Bucket: bucketName,
    Key: 'sample.json', // Replace with your file name
  };

  // Create a stream of JSON data from S3 object
  const s3ReadStream = s3.getObject(getObjectParams).createReadStream();
  const sampleStream = Readable.from(s3ReadStream);

  const stream = of(sampleStream)
    // parse binary stream to objects      
    .pipe(
      nodeTransform(
        JSONStream.parse('*')
      )
    )
    // filter out negative values      
    .pipe(filter(Boolean))
    // ensure type safety that customer is of type Customer      
    .pipe(filter(validateCustomer))
    .pipe(
      map(
        async (customer) => {
           // after validation it's a type-safe mapper as an example
           // validation ensures that the customer is of type Customer
           return customer;
        }
      )
    );


  try {
    // fetch data from stream and print results
    for await (const customer of stream) {

      // there is stream of type Customer
      // and `customer` could be safely used

      console.log(`Name: ${customer.name}, Age: ${customer.age}`);
      // Additional processing or storage can be performed here
    }
  } catch (error) {
    console.error('Error:', error);
    throw error;
  }
};        

Check the code example in the original post with better highlights.

Conclusion

This article has demonstrated the creation of robust and type-safe data pipelines through the integration of strict-stream, JSONStream, and AWS Lambda. By leveraging the capabilities of these tools, developers can establish pipelines that not only process data efficiently but also validate it against predefined structures, ensuring data integrity and reliability.

The utilization of TypeScript's type system adds an extra layer of security to the pipelines, allowing errors and mismatches to be caught during compile-time rather than runtime. This proactive approach fosters stability, predictability, and adherence to strict specifications.

By seamlessly integrating AWS Lambda and S3 with the meticulously crafted pipeline, developers can unlock the potential of scalable and serverless data processing. This integration, along with advanced streaming capabilities, facilitates efficient data handling while upholding high standards of type safety.

As organizations continue navigating complex data processing scenarios, prioritizing type safety becomes paramount. The tools explored in this article empower developers to construct pipelines that not only process data at scale but also ensure its quality. Type-safe pipelines contribute to improved code maintainability, fewer runtime errors, and an enhanced user experience.

In a data-driven world where data integrity is non-negotiable, the amalgamation of strict-stream, JSONStream, and AWS Lambda stands as a testament to the commitment to building dependable and type-safe data pipelines. Embracing these tools and practices equips developers to navigate evolving data processing landscapes with confidence, knowing that their pipelines are built on the foundations of accuracy, reliability, and adherence to type definitions.

For more information on the mentioned NPM packages, refer to:

1. strict-stream: [npm install strict-stream](https://www.npmjs.com/package/strict-stream)

2. JSONStream: [npm install jsonstream-next](https://www.npmjs.com/package/jsonstream-next)

3. AWS SDK for JavaScript (aws-sdk): [npm install aws-sdk](https://www.npmjs.com/package/aws-sdk)

Happy coding!

Reference to original article is here