Configure Kinesis Event Notifications for S3 Pipelines

Overview

S3 pipelines discover new files by periodically scanning the bucket using AWS ListObjectsV2 API calls, which return only 1,000 files at a time. For buckets with a large volume of files, this process can take several minutes.

With Kinesis-based file discovery:

1. S3 sends event notifications to Amazon EventBridge when files are created.

2. EventBridge routes events to a Kinesis Data Stream.

3. The SingleStore pipeline reads events from the stream and loads files immediately.

4. File discovery latency is reduced to approximately <1.5 seconds.

Prerequisites

Ensure the following:

• An AWS account with permissions to create and configure:

  • S3 buckets and event notifications

  • Amazon EventBridge rules

  • Amazon Kinesis Data Streams

  • IAM roles and policies

• An existing S3 bucket with data files

Step 1: Enable EventBridge Notifications on the S3 Bucket

1. Navigate to Amazon S3 in the AWS Management Console.

2. Select your S3 bucket.

3. Navigate to the Properties tab.

4. Navigate to the Amazon EventBridge section.

5. Select Edit.

6. On the Edit Amazon EventBridge page, select On.

7. Select Save changes.

Step 2: Create a Kinesis Data Stream

Create a Kinesis Data Stream to receive S3 event notifications.

1. Navigate to Amazon Kinesis in the AWS Management Console.

2. Select Create data stream.

3. Configure the stream:

   • Data stream name: Enter the name of your Kinesis data stream.

   • Capacity mode: Select Provisioned. This is for predictable workloads (start with 1 shard).

4. Select Create data stream.

5. Copy and securely store the stream ARN.

Step 3: Configure IAM Permissions

Configure permissions for the following:

1. EventBridge to write to Kinesis

2. SingleStore to read from Kinesis and S3

{
 "Version": "2012-10-17",
 "Statement": [
   {
     "Sid": "KinesisReadAccess",
     "Effect": "Allow",
     "Action": [
       "kinesis:DescribeStream",
       "kinesis:GetRecords",
       "kinesis:GetShardIterator",
       "kinesis:ListShards"
     ],
     "Resource": "arn:aws:kinesis:us-east-1:123456789012:stream/s3-file-notifications"
   },
   {
     "Sid": "S3ReadAccess",
     "Effect": "Allow",
     "Action": [
       "s3:GetObject",
       "s3:ListBucket",
       "s3:HeadObject"
     ],
     "Resource": [
       "arn:aws:s3:::your-bucket-name",
       "arn:aws:s3:::your-bucket-name/*"
     ]
   }
 ]
}

Step 4: Create an EventBridge Rule

1. Navigate to Amazon EventBridge in the AWS Management Console.

2. Select Create rule.

3. On the Create rule page, in Builder mode, select Advanced builder.

4. In Define rule detail, configure the following:

   1. Name: Enter the name of the EventBridge rule.

   2. Description (optional): Enter the description of the EventBridge rule.

   3. Event bus: Select default.

5. Enable the rule on the selected event bus and select Next.

6. On Build event pattern, configure the following:

   1. In Events, select the following:

      1. Event source: AWS events or EventBridge partner events.

   2. In Event pattern, select the following:

      1. Creation method: Select Custom pattern (JSON editor).

      2. Event pattern: Select Edit pattern. Enter the following event JSON and replace <YOUR_BUCKET_NAME> with your S3 bucket:

         JSON

         Copy

         {
           "source": ["aws.s3"],
           "detail-type": [
             "Object Created",
             "Object Deleted"
           ],
           "detail": {
             "bucket": {
               "name": ["<YOUR_BUCKET_NAME>"]
             }
           }
         }

7. In Select target(s), configure the following:

   1. Target type: Select AWS service.

   2. Select a Target: Select Kinesis stream.

   3. Target location: Select from the following:

      1. Target in this location

      2. Target in another AWS account

   4. Stream: Based on the selected target location, enter or select the following:

      1. If Target in this location is selected, select your Kinesis stream.

      2. If Target in another AWS account is selected, enter the stream ARN.

   5. Partition key (optional): Enter JSON path as partition key.

   6. Execution role: Select from the following:

      1. Create a new role for this specific resource

      2. Use existing role

   7. Role Name: Based on the selected execution role, enter or select the following:

      1. If Create a new role for this specific resource is selected, enter the name of the role.

      2. If Use existing role is selected, select the role name.

      Select Next.

8. (Optional) In Configure tags, select Add new tag. Enter Key and Value. Select Next.

9. In Review and Create, review the configurations and select Create rule.

Step 5: Test the Configuration

Verify that events flow correctly before creating the pipeline.

echo "test data" > test-file.txt
aws s3 cp test-file.txt s3://your-bucket-name/test-file.txt

SHARD_ITERATOR=$(aws kinesis get-shard-iterator \
 --stream-name <your-stream-name> \
 --shard-id shardId-000000000000 \
 --shard-iterator-type TRIM_HORIZON \
 --query 'ShardIterator' \
 --output text \
 --region us-east-1)

aws kinesis get-records \
 --shard-iterator $SHARD_ITERATOR \
 --region us-east-1

Step 6: Create the SingleStore Pipeline

Create the S3 pipeline with Kinesis-based file discovery enabled. The following examples demonstrates how to create SingleStore pipeline with different credential modes.

CREATE PIPELINE fast_s3_pipeline AS
LOAD DATA S3 'your-bucket-name/data/'
CONFIG '{
 "region": "us-east-1",
 "file_notifications_kinesis_stream_arn": "arn:aws:kinesis:us-east-1:123456789012:stream/s3-file-notifications"
}'
CREDENTIALS '{
 "aws_access_key_id": "YOUR_ACCESS_KEY",
 "aws_secret_access_key": "YOUR_SECRET_KEY"
}'
INTO TABLE my_table
FORMAT JSON;

CREATE PIPELINE fast_s3_pipeline AS
LOAD DATA S3 'your-bucket-name/data/'
CONFIG '{
 "region": "us-east-1",
 "file_notifications_kinesis_stream_arn": "arn:aws:kinesis:us-east-1:123456789012:stream/s3-file-notifications",
 "role_arn": "arn:aws:iam::123456789012:role/singlestore-pipeline-role"
}'
INTO TABLE my_table
FORMAT JSON;

CREATE PIPELINE fast_s3_pipeline AS
LOAD DATA S3 'your-bucket-name/data/'
CONFIG '{
 "region": "us-east-1",
 "file_notifications_kinesis_stream_arn": "arn:aws:kinesis:us-east-1:ACCOUNT_A:stream/s3-file-notifications",
 "role_arn": "arn:aws:iam::ACCOUNT_A:role/cross-account-pipeline-role"
}'
CREDENTIALS '{
 "aws_access_key_id": "YOUR_ACCESS_KEY",
 "aws_secret_access_key": "YOUR_SECRET_KEY"
}'
INTO TABLE my_table
FORMAT JSON;