AI-Powered MongoDB ETL Without the Pain - Siamion Makarski

This talk explores the construction of MongoDB-driven Change Data Capture (CDC) pipelines utilizing artificial intelligence for enhanced agentic experiences. The speaker shares insights gained from over 13 years of developing MongoDB solutions, particularly in the realm of streaming data and pipeline management. Key challenges addressed include ensuring compliance and data governance when handling sensitive information, such as personally identifiable information in fintech applications. Through a live demonstration, the speaker showcases a pipeline built to mask sensitive user data while using MongoDB's aggregation capabilities and a custom transformation layer implemented in Rust. The session highlights the advantages of using an intelligent agent to streamline pipeline creation, manage data transformations, and maintain compliance throughout the process.

[music] >> So, today I will be trying to speak about the uh uh how to build MongoDB driven uh CDC pipelines with the help of AI and agentic experience. Um our company has been building MongoDB solutions for a while already, more than 13 years. And I personally have been working with the streaming solutions for a while and have experienced all the different pains of how to deal

with the pipelines and the detail solutions. And uh we we want to share our perspective of how can that can be done really in an easy way uh still with having all the compliance checks in place and the data governance also not affected. So, that would be in fact very interesting for such industries as fintech or any um application use case where there is the personally identifiable

information is being stored for the users. So, that's what we will focus today on. Hopefully, the live demo goes without any problems. Uh I have a backup solution just in case, but I hope that it also will work as intended. All right. Before we already started. Before we dive in into the live demo, I will just want to take you through a a use case uh with

how the data can drift uh in a live CDC pipeline. Let's let's imagine we have a a data pipeline that is working with the some personal user data. On the left-hand side, in the green, you have name and email and phone and the SSN. All of all of that is redacted, obfuscated, and masked. All is fine. And as long as it works, that that is However, if

it works without any problems for too long, then usually there is a question mark whether everything was just as At some point, analytics team or product team would want to have another feature added, which will also change your schema originally, and add a new field, which is a billing email. That billing email can be also a sensitive piece of that needs redaction, but it can just sleep

through and get unattended if you do not really have the right process to design the the pipelines in a really sustainable manner. Therefore, we want to introduce, and I want to share like a perspective, which is on the right hand side. That's our ETL bridge transforming engine in the future, but before I speak about that, just speaking about how that can be actually solved with the native

tools on the MongoDB side. There is a way to write an aggregation on the MongoDB side. They also have the streaming solution on Atlas, and it supports kind of an aggregation syntax. And that syntax is have a very hard to maintain and to write. And even in this example, it is 62 lines of code, which was not really accounting for that for that additional case with the

billing email data masking. It's really hard. I I personally have been doing that, and sometimes just sitting there stunned and frozen for 20 minutes and not knowing what to do next. In our transformation layer, which is supported by the Rye language, which is a subset of Rust, because the back end is implemented in Rust, it the whole similar operation will just take only 14 lines of code.

Uh and with all the native functions as hashing, masking phones, emails, addresses and uh the conditionals are really intuitive even if you never have seen Rust right before. that seems very compelling and too good to be true. So, uh let's see how we can actually experience that in real life. What is we could build such a pipeline which is taking the user data, uh masks it and

then also designs it that the data is being synced to the source destination of your choice if you just describe everything in plain English. So, that's the whole experience about the Agent AI, isn't it? So, that's the 26. Let's switch to the live demo. today I will try to take through the journey of building a pipeline that looks like this. This is my backup plan. So, we'll

try to replicate it using the Agent experience. On that on this pipeline we see the source data which is the database source, the transformation layer, and a few things. One is the another one is Kafka without any schema applied, and the third one is the Kafka with the Avro schema applied. So, uh one source, one transformation layer, three different destinations. >> [snorts] >> Let's dive in. Uh

we have in our tool the the Agent tab, obviously. Um that's will be powered by the Opus 4.6 in this demo. And I have a few prompts saved for myself, so uh I will copy I'll be copying the prompts and we'll be going through the experience of building the pipeline generally. Okay. So, I'm also very open with the agent. I'm telling it that I'm using actually doing

a live demo, so it needs really to take care of me. Okay. What I'm pasting in here that I'm doing a live demo at a conference, and I'm just for the first asking to list all the connections that I have predefined for our demo session, so we can see what we can work with. What are the tools available in there? So, what you can see right now,

I entered the prompt. The agent is executing a few tools that are available to it. It's listing the services. One of the services would be our MongoDB service, and it lists all the resources. So, as you can see, we have one collection of orders that I have picked up for our use case, and it asks me what uh whether I would like to continue or build something

with this collection of orders, and this is what we intend to do. Uh just a very uh very quick preview into what is in the data. This is synthetic data, but so that you understand what's in there. It's full of the sensitive information. Customer name, phone number, items, et cetera. So, we're going to be working against this source. So, I will be posting another prompt to the

AI agent and asking it to create a CDC pipeline, which is a with a specific name. And what I'm asking it to do, I need to mask email, phone fields for GDPR compliance. This is allowed in Europe for taking care of all the sensitive user data in there. And uh write the output into the new collection, which is orders clean, in in the specified sync that I

also prepared for this demo. And also each transformed document will be published to our Kafka topic at the same time. For now, we'll be targeting only two syncs, and then we'll add another sync afterwards at the latest stage. So, we'll see how this goes. So, again, the agent is executing the available services and the resources, does and checks, and what it already has figured out that origin

database available, that there is a middleware, which is a available for transformations, and the syncs. And it's asking me whether it wants me to deploy the pipeline. And I will say yes. Okay, now it claims it has created a pipeline, but how we can validate it. Everything seems to to go too smooth for now at the moment. The first thing that we can go to is go

to the pipeline step and check what's available in there. And as you can see, the this new job got created, which is running status, checkpoint enabled. Everything seems to be fine. Nothing has been running through it so far, and we have two syncs that have been configured. Good. For a brief light test, we want also we might just also to make sure on the human side to

to make sure that it's tested well, and then it behaves as expected. So, what I did right now, I clicked on the pipeline and to open it in our studio, which is kind of a kind of a workplace for human to validate what's been actually happening. uh on the left-hand side, you can see uh input data source that's been predefined and also the the modification script. Uh

if I clicked on run here, nothing will happen except for the simulation of how the data is going to be uh obfuscated This environment is uh very human-friendly and also in a way compliant cuz of what in here uh well, when I opened the um this in the studio, the tool went to the orders collection. It took a sample of 1,000 to 100 uh items from the

collection and it inferred the schema. And based on that schema, it prefilled it with a dummy data. So, not really actually real data that is available in the origin database got leaked to the tool itself even at the stage of the design of the pipeline. Just once again, and you can see the the the data has been also algorithmically shielding based on the shields. Uh but, the

actual real data looks looks And if I had to query by that, I wouldn't find that entry with the Cornelius there uh in the origin database. we also have AI tab available here. If, for example, I make a mistake, uh my application will not compile and I can talk to AI to fix it. Uh help me fix the error. Just in case, whenever you created something with

Agent Key AI experience and you want to modify something, then um this is kind of your tool to, you know, to share the context and it sees whatever you see also on the screen. And it says that it's missing a semicolon, so then identify where I made a mistake. And it suggests me to instruct. This is on purpose that in that UI, in that screen, you cannot

really have a AI agent to do anything for you. You'll have to consciously do it yourself. Okay, but seems to be fine that we have validated that case. We can also generate the our assertions based on this example and run the tests. Here the tests are failing because we have the assertion for the timeline for the timestamp. I can I can remove it. Because we are not

interested in asserting I can't I know Probably could have saved it. Okay, I didn't plan for showing this case with the timestamps, but there is one more timestamp that I cannot really guess right now. Let me just validate and I'll keep only a few fields for assertion. save it and run. And in fact, it will be what I did right modifying the in a strict mode. It

has sorted also for the fields that are not in the test or that are not being asserted. So that's kind of giving you additional check to make sure that there are no additional fields flying flying through and being propagated in the pipeline. Okay, good. Seems to be fine. Let's try to simulate how this works in in real life. I have here uh a few statements then I

will insert the um the document uh into the origin database and we'll need to see whether there is something already available in our sinks. As you can see uh I have inserted that document with the live postfix and we've got the the final target document uh already properly completely masked and it's available in the target system. If I go back to to the pipelines I'll see here

in the statistics that we got one event processed which is 300 bytes and no errors were spotted here. So so far so good. Uh we can also run a few more uh insertion state to make sure that it works. And we can go to the our And we have more like Fatima live and Raj as well live. So the data got propagated. When we show when we

have a look at this in the in the pipeline statistics we've got three events that got processed. Okay, but originally when I was starting talking about what we want to build we also wanted to build that third additional thing which would be supporting the um on the Kafka output. Uh let's try to do that. Now okay, that was in fact um one more additional step that I

I wanted to show that we can also uh add additional fields in a transformation layer. Let's do that first. So, what I'm asking is to evaluate the order and just append the data which is small, medium, or else based on the conditionals. and have additional information in the output document. Um I will go back to to show that to use to show that use case there, but

uh let let's uh step into step into the actually building the um the schema into the pipeline. So, I'm I'm telling the agent right now that I want to have the third sink added to our pipeline, which was on our original example, and use the the topic uh for the sink, the encoding Avro, and keep the keep the audit pipeline characteristics in intact and Okay, and it

says that it needs to stop, delete, and recreate the pipeline, so the changes have been And everything will be identical, and it will be safe because we have checkpoints enabled, so nothing will be lost. Okay, and I'm saying go because it's asked me to say go. Now, it stopped the running job, which had the three events processed. It creates a new job with the new sync available.

Okay, let's see. Okay, on our leg it seems fine. It seems to be running. We've got the Kafka demo uh appended as the additional sync. Let's see how this works actually. Like uh if if the any data gets processed. I will do the same exercise. I will just uh I will do a name life to Avro. And insert this data this document in the database on the

origin site. Very quick very quickly looking into Okay, seems good. Seems good at least we have uh the data being propagated to our database sync. We also had that additional transformation transformation applied because we wanted to estimate how big the order size in the human language was based on the based on the size and information it identified it as large. And uh what we want to have

a look at whether the data in fact landed in uh Kafka. So, here are my Kafka topics. I'll just renew. And here are my um order analytics Avro life demo topics. And order master JSON. So, let's have a look at JSON first and the messages. Uh so far we've been processing uh these four messages. So, all of them in there. And all of them just looks exactly

like we have in our market target. And the last one with the life to with Avro is also available. However, this topic is plain JSON. Let's check out how the uh Avro ones one looks. Yeah, it's been encoded in the binary Avro format. So, this is also one additional line of defense because you can define in Avro, use it as a as a mask to apply to

the origin document and it could only filter out two fields that you need. Filter them out and propagate in the Avro contract binary format. So, that you have your experience consistent across the syncs sync to sync systems. >> Um it's also possible to chain the pipelines if you want. So, in our pipeline, we could also use one of those as a source for the next pipeline that

you design. Um as a matter of fact, like like being conscious of time, uh I will not go into the details of that, but it will be kind of the same experience. Can be also identical one on the manual mode that I showed in the studio. Like similar to here, you have you can work with that as a kind of your workbench to to build all of

this. But, for now, I'll just want to very quickly wrap up uh before we go into the Q&A. If there are any questions of what's uh how this agent is being built. And what's been under the hood. So, with uh the in the heart of this uh the pipeline engine, there is a back-end application which is implemented in Rust. It has uh native tooling support with more

than 30 tools at the moment. These tools support the all the modifications that you have already seen so far. Building the pipelines, building the workspaces. Uh but, more to that, also you can interact purely with the AI agent to understand how your system performs. Yeah, all the job metrics available that are also exposed to Prometheus by endpoint also available to to the agent, so it can really

uh query those and give you a very quick analysis into what's going on. Uh we have the actually for I think I haven't seen any tool doing that, the horizontal scaling because the change screens are designed in a way that you have only one cursor running and you cannot really easily split it across multiple consumers. So, we have the logic that is allowing to have a leader

election that is being changed every 30 seconds. And you if for example you want to run 10 workers to horizontally split your processing, then there will be always for CDC pipeline one leader and then there will be nine workers that are picking up that job. So, as it looks right now, what what we are trying to do and have to is not to really build the tools

for humans in that in the in the direct sense. We are rather trying to build the tools for the AI and keeping the human in the loop as a decision maker. So, the main challenges like well working with the AI is governance and compliance and obviously data leakage. So, therefore the we have really put a lot of focus on on how the data is being exposed to

the AI. Therefore, while we are designing the pipeline originally, we do not really let the application to have access to the real data, which I have showed to you in the orders database originally. It's already kind of all being introspected based on the schema. Being shielded with a dummy data and then gives you an opportunity to really simulate and how the pipeline will work will look like.

and with the older test in place, we're trying to really have our defenses in place. Um, that's it uh from me for today, and I'll be happy to answer any questions in the last 5 minutes that we have. >> [music]