Amazon S3

Integrating Amazon S3 as a Sink with GlassFlow

In this guide, you will learn how to integrate Amazon S3 cloud storage with GlassFlow using the GlassFlow Python SDK. You will build a custom integration and implement the sink logic in Python code to send the data to S3.

Prerequisites

Before you start, make sure you have the following:

• A GlassFlow account. Sign up for a free GlassFlow account.

• You have an AWS account.

• You installed AWS CLI.

• Python is installed on your machine.

• Pip is installed to manage project packages.

Step 1: AWS Configuration

We use Boto3 (AWS SDK for Python) to store processed streaming data in AWS S3.

Create IAM user

Before using Boto3, you need to set up authentication credentials for your AWS account using either the IAM Console or the AWS CLI. You can either choose an existing root user or create a new one.

For instructions about how to create a user using the IAM Console, see Creating IAM users. Once the user has been created, see Managing access keys to learn how to create and retrieve the keys used to authenticate the user.

Copy both generated aws_access_key_id and aws_secret_access_key, you will use them when you configure the local credentials file and to set environment variables.

Configure credentials file

Use AWS CLI installed to configure your credentials file by running the below command:

aws configure

Alternatively, you can create the credentials file yourself. By default, its location is ~/.aws/credentials. in MacOS/Linux operating systems At a minimum, the credentials file should specify the access key and secret access key. In this example, the key and secret key for the account are specified in the default profile:

[default]
aws_access_key_id = YOUR_ACCESS_KEY
aws_secret_access_key = YOUR_SECRET_KEY

You may also want to add a default region to the AWS configuration file, which is located by default at ~/.aws/config:

[default]
region=us-west-2

Alternatively, you can pass an AWS_REGION name as an environment variable when creating clients and resources.

You have now configured credentials for the default profile as well as a default region to use when creating connections. See Configuration for in-depth configuration sources and options.

Step 2: Create a GlassFlow Pipeline

1. Navigate to GlassFlow WebApp and log in with your credentials.

2. Click on Create New Pipeline.

3. Set the pipeline name to s3-data-pipeline.

4. Data Source: Select SDK or built-in integration as the data source. If using a built-in integration, follow the prompts to enter the required details (e.g. topic for Google Pub/Sub)

5. Transformation: Upload your transformation script (transform.py) to handle any data processing or enrichment. Or choose a default Echo function.

6. Data Sink: Select SDK as the data sink. You will configure the sink in your code to send the data to a specific S3 bucket.

7. Confirm Pipeline Creation: Copy the new Pipeline ID and Access Token.

Step 3: Setup a custom connector for S3

Clone the project

Start by cloning the glassflow-examples GitHub repository to your local machine.

git clone https://github.com/glassflow/glassflow-examples.git

Navigate to the project directory:

cd connectors/sink/aws-s3

Create a new virtual environment

Create a new virtual environment in the same folder and activate that environment:

python -m venv .venv && source .venv/bin/activate

Install libraries

Install the GlassFlow, AWS Boto3 Python SDK, and virtual environment package python-dotenvusing pip.

pip install glassflow python-dotenv boto3

Step 4: Create Environment Configuration File

Add a .env file in the project directory with the following configuration variables:

PIPELINE_ID=your_pipeline_id
PIPELINE_ACCESS_TOKEN=your_pipeline_access_token
AWS_REGION=your_aws_region
AWS_ACCESS_KEY_ID=your_aws_access_key_id
AWS_SECRET_ACCESS_KEY=your_aws_secret_access_key
S3_BUCKET_NAME=your_s3_bucket_name

Replace placeholders with the actual values from your GlassFlow account and AWS credentials.

Step 5: Push Data to Amazon S3

Check an example Python script sink_connector.py that pushes data to your S3 bucket. Customize it if needed.

https://github.com/glassflow/glassflow-examples/blob/main/connectors/sink/aws-s3/sink_connector.py

import glassflow
import boto3
import sys
import time
import json
import os
import uuid
from dotenv import load_dotenv
import random


class SinkConnectorS3:
    def __init__(self) -> None:
        self.bucket_name = os.environ["AWS_S3_BUCKET_NAME"]
        self.aws_region_name = os.environ["AWS_REGION"]
        self.firehose_uid = uuid.uuid4().hex[0:4]
        self.role_name = "firehose_delivery_role_%s" % self.firehose_uid
        self.firehose_delivery_stream_name = "glassflow-pipeline-%s" % self.firehose_uid
        self.aws_account_id = self._get_aws_account_id()
        self.firehose_client = boto3.client(
            "firehose", region_name=self.aws_region_name
        )

    def setup(self):
        if not self._check_firehose_stream():
            self._setup_iam_firehose_role()
            time.sleep(30)
            # waiting for role to be created before setting up firehose
            self._setup_aws_firehose()

        if not self._check_firehose_stream():
            print("Delivery stream still not exist. Exiting...")
            sys.exit(1)

        # Sleep for 30 seconds to allow the delivery stream to be created
        time.sleep(30)

    def _get_aws_account_id(self):
        try:
            # Create a new STS (Security Token Service) client
            sts_client = boto3.client("sts", region_name=self.aws_region_name)

            # Call the get_caller_identity API to retrieve account information
            response = sts_client.get_caller_identity()

            # Extract and return the account ID from the response
            aws_account_id = response["Account"]
            return aws_account_id
        except Exception as e:
            print(f"An error occurred while retrieving AWS account ID: {e}")

    def _setup_iam_firehose_role(self):
        # Create IAM client
        iam_client = boto3.client("iam")
        # Define IAM policy document
        policy_document = {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": "s3:PutObject",
                    "Resource": f"arn:aws:s3:::{self.bucket_name}/*",
                }
            ],
        }
        # Create IAM role if not exists
        try:
            iam_client.get_role(RoleName=self.role_name)
            print("IAM role already exists.")
        except iam_client.exceptions.NoSuchEntityException:
            print("IAM role does not exist. Creating...")
            response = iam_client.create_role(
                RoleName=self.role_name,
                AssumeRolePolicyDocument=json.dumps(
                    {
                        "Version": "2012-10-17",
                        "Statement": [
                            {
                                "Effect": "Allow",
                                "Principal": {"Service": "firehose.amazonaws.com"},
                                "Action": "sts:AssumeRole",
                            }
                        ],
                    }
                ),
            )
            print("IAM role created successfully.")

            # Attach IAM policy to role
            iam_client.put_role_policy(
                RoleName=self.role_name,
                PolicyName="firehose_delivery_policy",
                PolicyDocument=json.dumps(policy_document),
            )
            print("IAM policy attached to role.")
        except Exception as e:
            print("An error occurred:", e)
            sys.exit(1)

    def _check_firehose_stream(self):
        try:
            stream = self.firehose_client.describe_delivery_stream(
                DeliveryStreamName=self.firehose_delivery_stream_name
            )
            return True
        except self.firehose_client.exceptions.ResourceNotFoundException:
            return False

    def _setup_aws_firehose(self):
        # Create a Kinesis Firehose client
        firehose_client = boto3.client("firehose", region_name=self.aws_region_name)

        # Create Firehose delivery stream if not exists
        try:
            firehose_client.create_delivery_stream(
                DeliveryStreamName=self.firehose_delivery_stream_name,
                ExtendedS3DestinationConfiguration={
                    "RoleARN": f"arn:aws:iam::{self.aws_account_id}:role/{self.role_name}",  # Use the created IAM role ARN
                    "BucketARN": f"arn:aws:s3:::{self.bucket_name}",  # Use the created S3 bucket ARN
                    "Prefix": "glassflow_transforms/",  # Prefix for S3 objects
                    "ProcessingConfiguration": {
                        "Enabled": True,
                        "Processors": [
                            {
                                "Type": "AppendDelimiterToRecord",
                                "Parameters": [
                                    {
                                        "ParameterName": "Delimiter",
                                        "ParameterValue": "\\n",
                                    }
                                ],
                            }
                        ],
                    },
                },
            )
            print(
                f"Delivery stream: {self.firehose_delivery_stream_name} created successfully."
            )
        except Exception as e:
            print("An error occurred when creating stream:", e)
            sys.exit(1)

    def put(self, data):
        try:
            json_data = json.dumps(data)

            # Put record to Firehose delivery stream
            response = self.firehose_client.put_record(
                DeliveryStreamName=self.firehose_delivery_stream_name,
                Record={"Data": json_data.encode("utf-8")},
            )
            print(response)
        except Exception as e:
            print("An error occurred when putting record to Firehose:", e)
            # TODO decide what to do in case of error

    def cleanup(self):
        self.file.close()


def main():
    sink_connector = SinkConnectorS3()
    sink_connector.setup()

    pipeline_id = os.environ["PIPELINE_ID"]
    space_id = os.environ["SPACE_ID"]
    pipeline_access_token = os.environ["PIPELINE_ACCESS_TOKEN"]

    # Create GlassFlow client
    pipeline_client = glassflow.GlassFlowClient().pipeline_client(
        space_id=space_id,
        pipeline_id=pipeline_id,
        pipeline_access_token=pipeline_access_token,
    )

    retry_delay = 10
    while True:
        try:
            res = pipeline_client.consume()
            if res.status_code == 204:
                # No new events. sleep for a little bit
                time.sleep(retry_delay)
                retry_delay *= 2  # Double the delay for the next attempt
                retry_delay += random.uniform(0, 1)  # Add jitter
                continue
            if res.status_code == 200:
                record = res.json()
                sink_connector.put(record)
                print(
                    "Consumed transformed event from GlassFlow and sent to S3 via Firehose"
                )
                # set the retry delay back to original
                retry_delay = 10
        except KeyboardInterrupt:
            print("Exiting")
            sink_connector.cleanup()
            break


if __name__ == "__main__":
    load_dotenv()
    main()

Run the sink_connector.py script to push data to your S3 bucket:

python sink_connector.py

Summary

In this guide, you learned how to set up a data streaming pipeline with GlassFlow, integrating Amazon S3 using the GlassFlow custom connector. You configured the pipeline using SDKs for data sink, and you pushed sample data to S3 using a Python script.