In this blog, we will walk through how to build realtime AI data pipeline with with Amazon S3 bucket, SQS and CocoIndex - an open source data ETL framework - <a href="https://github.com/cocoindex-io/cocoindex" target="_blank" rel="noopener nofollow">https://github.com/cocoindex-io/cocoindex</a> easily.
<img src="https://media.daily.dev/image/upload/s--9pSdUbYD--/f_auto/v1748590930/ugc/content_bed8821f-283d-4b7e-8f14-0a813b23c23b?_a=BAMClqUq0" alt="cover">
<h2>Why incremental processing?</h2>
Incremental processing is a data processing approach that only processes new or changed data since the last run,
rather than reprocessing the entire dataset each time.
This is particularly valuable for large datasets where reprocessing everything would be inefficient and time-consuming,
or when you require low latency on data freshness. In general, incremental processing is preferred under the following conditions:
<h3>High Freshness Requirement</h3>
For most user-facing applications this is needed. For example, when users update their documents, it’s unexpected if they see stale information in search results.
If the search result is fed into an AI agent, it may mean unexpected response to users (i.e. LLM generate output based on inaccurate information). It’s more dangerous and users may even take the unexpected response without noticing.
<h3>High Transformation Cost</h3>
Transformation cost is significantly higher than retrieval itself. This is especially true when using expensive AI models for embedding or processing.
<h3>Data is large scale</h3>
When dealing with large-scale datasets, reprocessing the entire dataset can be impractical or even impossible due to:
<ul>
<li>Limited computational resources</li>
<li>Time constraints</li>
<li>Cost considerations</li>
<li>Storage limitations</li>
</ul>
For example, if you have terabytes of data in S3, reprocessing everything every time a new file arrives would be extremely inefficient and costly.
Incremental processing allows you to handle these large datasets efficiently by only processing the changes.
Overall, say T is your most acceptable staleness. If you don’t want to recompute the whole thing repeatedly every cycle of T, you will need incremental processing more or less.
<h2>Setup Infra to handle incremental processing</h2>
<img src="https://media.daily.dev/image/upload/s--ZQLtW7ld--/f_auto/v1748570867/ugc/content_fce59c74-b028-462a-981b-ec6996678256?_a=BAMClqUq0" alt="aws-incremental-etl">
CocoIndex’s incremental processing provides several key technical advantages:
<ol>
<li>
Efficient Change Processing: Only processes new or modified S3 files, optimizing computational resource usage.
</li>
<li>
Real-time SQS Integration: Leverages AWS SQS for instant S3 change detection and near real-time processing.
</li>
<li>
Smart Caching System:
<ul>
<li>Automatically manages persisted states with data lineage tracking to ensure data consistency.</li>
<li>Optimizes performance through smart caching and minimal recomputation of only changed components.
The caching is at the granularity of your control on the transformation flow, beyond the scope of a file.
For example, if you have M chunks of a file, and N chunk changed and requires embedding,
only the N chunks will be recomputed if the transformation logic is the same.</li>
</ul>
</li>
</ol>
<h2>AWS SQS</h2>
<a href="https://aws.amazon.com/sqs/" target="_blank" rel="noopener nofollow">Amazon SQS</a> (Simple Queue Service) is a fully managed message queuing service that enables you to decouple and scale microservices, distributed systems, and serverless applications. It provides a reliable, highly-scalable hosted queue for storing messages as they travel between applications or microservices.
When files are uploaded to or modified in S3, SQS receives notifications about these changes and queues them as messages. Each message contains metadata about the S3 event, such as:
<ul>
<li>The type of event (e.g., ObjectCreated, ObjectRemoved)</li>
<li>The S3 bucket name</li>
<li>The object key (file path)</li>
<li>Timestamp of the event</li>
<li>Other relevant metadata</li>
</ul>
These messages remain in the queue until they are successfully processed, ensuring no events are lost even if the processing system is temporarily unavailable.
<h2>Live update out of the box with SQS</h2>
CocoIndex provides two modes to run your pipeline, one time update and live update, both leverage the incremental processing. Particularly with AWS SQS, you could leverage the live update mode -
where CocoIndex continuously monitors and reacts to the events in SQS, updating the target data in real-time. This is ideal for use cases where data freshness is critical.
<h2>How does it work?</h2>
Let’s take a look at simple example of how to build a real-time data transformation pipeline with S3 and CocoIndex. It builds a vector database of text embeddings from markdown files in S3. You could find a similar example to process local files in <a href="https://cocoindex.io/blog/text-embeddings-101" target="_blank" rel="noopener nofollow">this blog</a>.
<h3>S3 bucket and SQS setup</h3>
Please follow the <a href="https://cocoindex.io/docs/ops/sources#amazons3" target="_blank" rel="noopener nofollow">documentation here</a> to setup S3 bucket and SQS queue.
<h4>S3 bucket</h4>
<ul>
<li>Creating an AWS account.</li>
<li>Configuring IAM permissions.</li>
<li>Configure policies. You’ll need at least the <code>AmazonS3ReadOnlyAccess</code> policy, and if you want to enable change notifications, you’ll also need the <code>AmazonSQSFullAccess</code> policy.</li>
</ul>
<img src="https://media.daily.dev/image/upload/s--9Y6lf9VM--/f_auto/v1748570924/ugc/content_843fea27-b3db-447c-b6b2-24a126a4e026?_a=BAMClqUq0" alt="permission">
<h4>SQS queue</h4>
For real-time change detection, you’ll need to create an SQS queue and configure it to receive notifications from your S3 bucket.
Please follow the <a href="https://cocoindex.io/docs/ops/sources#optional-setup-sqs-queue-for-event-notifications" target="_blank" rel="noopener nofollow">documentation</a> to configure the S3 bucket to send event notifications to the SQS queue.
<img src="https://media.daily.dev/image/upload/s--yfPBC5Ux--/f_auto/v1748570965/ugc/content_15200a9a-ae22-4eb9-8701-05eabb026fdf?_a=BAMClqUq0" alt="sqs">
Particularly, the SQS queue needs a specific access policy that allows S3 to send messages to it.
<pre><code class="language-json">{
 ...
 &quot;Statement&quot;: [
 ...
 {
 &quot;Sid&quot;: &quot;__publish_statement&quot;,
 &quot;Effect&quot;: &quot;Allow&quot;,
 &quot;Principal&quot;: {
 &quot;Service&quot;: &quot;s3.amazonaws.com&quot;
 },
 &quot;Resource&quot;: &quot;${SQS_QUEUE_ARN}&quot;,
 &quot;Action&quot;: &quot;SQS:SendMessage&quot;,
 &quot;Condition&quot;: {
 &quot;ArnLike&quot;: {
 &quot;aws:SourceArn&quot;: &quot;${S3_BUCKET_ARN}&quot;
 }
 }
 }
 ]
}
</code></pre>
Then you can upload your files to the S3 bucket.
<img src="https://media.daily.dev/image/upload/s--vCmytFk2--/f_auto/v1748570978/ugc/content_6dbfaa42-fd23-4472-8812-c2446a509672?_a=BAMClqUq0" alt="s3">
<h2>Define Indexing Flow</h2>
<h3>Flow Design</h3>
<img src="https://media.daily.dev/image/upload/s--49xV-lSO--/f_auto/v1748571019/ugc/content_d6c83bc6-4487-4c11-a1b2-883b8af86d30?_a=BAMClqUq0" alt="flow">
The flow diagram illustrates how we’ll process our codebase:
<ol>
<li>Read text files from the Amazon S3 bucket</li>
<li>Chunk each document</li>
<li>For each chunk, embed it with a text embedding model</li>
<li>Store the embeddings in a vector database for retrieval</li>
</ol>
<h3>AWS File Ingestion</h3>
Define the AWS endpoint and the SQS queue name in <code>.env</code> file:
<pre><code class="language-bash"># Database Configuration
DATABASE_URL=postgresql://localhost:5432/cocoindex

# Amazon S3 Configuration
AMAZON_S3_BUCKET_NAME=your-bucket-name
AMAZON_S3-SQS_QUEUE_URL=https://sqs.us-west-2.amazonaws.com/123456789/S3ChangeNotifications
</code></pre>
Define indexing flow and ingest from Amazon S3 SQS queue:
<pre><code class="language-python">@cocoindex.flow_def(name=&quot;AmazonS3TextEmbedding&quot;)
def amazon_s3_text_embedding_flow(
 flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope
):
 bucket_name = os.environ[&quot;AMAZON_S3_BUCKET_NAME&quot;]
 prefix = os.environ.get(&quot;AMAZON_S3_PREFIX&quot;, None)
 sqs_queue_url = os.environ.get(&quot;AMAZON_S3_SQS_QUEUE_URL&quot;, None)

 data_scope[&quot;documents&quot;] = flow_builder.add_source(
 cocoindex.sources.AmazonS3(
 bucket_name=bucket_name,
 prefix=prefix,
 included_patterns=[&quot;*.md&quot;, &quot;*.mdx&quot;, &quot;*.txt&quot;, &quot;*.docx&quot;],
 binary=False,
 sqs_queue_url=sqs_queue_url,
 )
 )

</code></pre>
This defines a flow that reads text files from the Amazon S3 bucket.
<img src="https://media.daily.dev/image/upload/s--TmcPXOVd--/f_auto/v1748571035/ugc/content_6c959a34-5508-46ef-9a3a-f9ee8b41353a?_a=BAMClqUq0" alt="ingest">
<h3>Rest of the flow</h3>
For the rest of the flow, we can follow the tutorial <a href="https://cocoindex.io/blog/text-embeddings-101" target="_blank" rel="noopener nofollow">in this blog</a>.
The entire project is available <a href="https://github.com/cocoindex-io/cocoindex/tree/main/examples/amazon_s3_embedding" target="_blank" rel="noopener nofollow">here</a>.
<h2>Run the flow with live update</h2>
In the <code>main()</code> function, you can add the following code to run the flow with live update:
<pre><code class="language-python">def _main():
 with cocoindex.FlowLiveUpdater(amazon_s3_text_embedding_flow):
</code></pre>
and start the flow by running:
<pre><code class="language-bash">python3 main.py
</code></pre>
Note that the main purpose of putting the FlowLiveUpdater in <a href="http://main.py" target="_blank" rel="noopener nofollow">main.py</a> is that it could run query in the same process -
so one process for both live indexing and serving.
Other option to have long running process to keep index updated:
<pre><code class="language-bash">cocoindex update main.py -L
</code></pre>
And you will have a continuous long running process that will update the vector database with the new files in the S3 bucket.
<h2>Support us</h2>
We are constantly improving, and more features and examples are coming soon. If you love this article, please give us a star ⭐ at <a href="https://github.com/cocoindex-io/cocoindex" target="_blank" rel="noopener nofollow">GitHub</a> to help us grow.
Thanks for reading!

Squad for devs building apps with GenAI and LLMs! Connect, share, exchange ideas, and get feedback. 

Build With GenAI

CocoIndex enables building real-time RAG pipelines by processing only new or changed data from Amazon S3 using SQS for event notifications. The framework provides incremental processing with smart caching, automatically handling data lineage and minimizing recomputation. The tutorial demonstrates setting up S3 bucket notifications, configuring SQS queues, and implementing a live-updating vector database pipeline that processes markdown files into text embeddings for retrieval applications.

Build Real-Time RAG Pipelines with CocoIndex, Amazon S3, and AWS SQS