Welcome, good morning everybody.
So today I want to talk a little bit about change data capture, CDC,
stream processing, with the Patrick link.
This talk is split into three parts.
The first part is for people that have never heard of link before.
The second part is maybe a little bit more deep, but I think it's really, really deep.
And then the third part, we could dive really, really deep into under the hood.
So just to make you particularly interested in the software and what we are doing here.
So yeah, I already got an introduction, but just to summarize it.
So like from the open source side, part of the link even before it became part of the Patrick software foundation in 2014.
I'm a member of the management committee of the Patrick link.
In the years I also made it to the top contributors according to additions in the top one.
I don't know which refactoring I did to the top one contributor.
Yeah, and among the core people that try to design things equal every day in the world.
Can you realize, yeah, I went through a couple of companies.
The latest one where I was a co-founder was in Maroc.
Maroc got acquired by Confluent beginning of this year.
And now I'm a principal software engineer at Confluent.
So let's talk about it.
Before I start with an introduction to the link, I would actually like to talk about stream processing in general.
Because when you do stream processing, you basically always can identify roughly like four building blocks.
So let's talk about those building blocks first.
So first of all, you need streams, right? You want to have data, you maybe want to create some pipeline from source to sync.
You might want to distribute your streams because you have a lot of data.
So maybe you want to scale out and scale in depending on the load.
You want to join streams together.
You want to enrich streams.
Maybe there is a control stream and the main stream.
So you want to dynamically modify the behavior of the application while the application is running.
And yeah, sometimes there is a bug in your application.
Or you just want to trace certain behavior, then you also want to be in play streams.
Time, working with time is also a very, very important concept.
Because on one side, you want to make progress in your pipeline.
But at some points, you also want to sync for an ISO if you have two streams.
Maybe you want to wait for the other stream.
Maybe you want to block or you want to buffer some of the streams.
Maybe if the second event doesn't come in, you want to time out after some time.
Maybe you also want to replay historical data.
So you want to fast forward the time.
You don't want to wait another hour to fire an hour window.
No, this should be quicker.
Then when we talk about buffering, what I just said, or in general, storing data state is a very important component for processing.
State can be, for example, a machine learning model that is updated from time to time to classify your incoming streams.
It can be a cache if you don't want to look up in the database for every record that comes in.
In a general state, a low state can be an order of classified.
And state also, at some point, needs to be acquired.
If you have a state full streaming application, a very useful component is helped by actually making sure that I can create this natural of my streaming repository.
So I wanted a state backup to my streaming application.
I wanted to make a version of it, so every night I want to create a screenshot version.
Maybe I want a full streaming application in a staging cluster, in a gelatin cluster, and play around with the data in the state.
Maybe I want to do some testing, or I just want to time track the process of my application.
So let's talk a little bit about what makes Flink unique compared to other competitors.
First of all, what I just showed is that Flink is one of the best stream processors for all of these use cases and building blocks.
So when you design a streaming application, you can start with a whiteboard and you just draw some circles.
What do you actually want to do?
Maybe you want to reform some sources.
Maybe you want to normalize your data.
You want to filter some data out.
You want to join the data and in the end you really want to sync it somewhere else.
But this is how it starts.
And this is also how you have to reason about when you're creating a pipeline.
And what Flink does under the hood is it has a parallelism and scalability built in.
So you don't have to think of threading or network transfers or anything like that.
Under the hood there are sharks, there are petitions depending on the connectors.
There are sub-tasks that in parallel execute operations.
Each of these tasks, some of these tasks can stay full and can have some storage local to the operator.
Very important.
So the state basically scales out and scales in with the operator.
You don't need to vary to a database which would increase data.
And then, of course, it travels and then the whole pipeline runs.
And now comes the important part.
What Flink explained really unique is that this possibility of creating an existence snapshot of your entire stream of technology.
So there are, like, what we call the checkpoint barriers which are traveling through the topology
and make a backup of each state of the operator.
And then this snapshot is then persisted on a long-term storage like S3 or HDFS or some other distributed kind of system.
When we talk about use cases, there are plenty of use cases.
We have the process transactions, docs, IOTs, any kind of events, user interactions.
People use it for broad detection for machine learning, for event-driven applications, for ETL, for data integration, for analytics.
So Flink has become, over the last 10 years, it has become like a very large platform.
You can connect various connectors from business stream systems, you can read and write files, databases, key value stores.
And as I said, like also event-driven applications where maybe you want to send out an email or don't need a connector.
You can also implement something custom that talks to somewhere as the API.
So let's look at this scene.
So I also want to quickly talk about Flink's API.
So this is the API stack.
The two main APIs are data stream API, table API or table table API.
And there is also stateful functions.
Stateful functions is a sub-project that tries to execute an actor model on a page of Flink, but will not go into detail here.
So first of all, like all the APIs are built on a data flow runtime, so there's no batching or anything involved under the hood.
It's really a data flow runtime.
Whatever the result is ready, it will be streamed to the next operator.
On top of that, there is a loader, the stream operator API, which you can use.
But yeah, this is for X, of course, I would say.
And then we have the mainstream APIs on top.
And the specialty about table table API is that there is an optimizer on a planning stage in between.
So the C helps you when you're not creating the most efficient pipelines.
The optimizer will make sure that the streaming will be executed more efficiently.
Yeah, let's also quickly look at the APIs.
So this is like a basic example of creating a stream for just three elements.
And then you're executing this on a cluster or in your IDE, then you're retrieving the result back.
And you have an iterator locally, you can just print it locally.
It's not very useful, but this is a minimal example of the Java API.
The important thing is that the stream API basically exposes all the building blocks that I mentioned on my previous slide.
So you can have very abstract operator typologies, and you can use built-in functions like map, process, and connect.
Which each of them takes different functions, and then you can really define your business logic in those.
They use different functions, and you can also use completely arbitrary Java records.
For example, Python records that flow between the operator and conceptually.
This is interesting when we talk about change data capture.
Conceptually, the data stream API does not know about changes.
It only knows about records, so there is no change flag or anything like that.
So conceptually, the data stream API is an app that can only or insert only as long.
And also when you look at the output, one, two, three, four, five, six.
So let's take a look at table API and simple API.
So usually you just say to this table API, order in SQL, because it's a unified API.
You can decide whether you want to electrify your pipeline programmatically,
or whether you want to use standard SQL for defining your topology.
In the end, you also execute and you can also print locally in your IDP.
Here, this API abstracts all building blocks.
So you have no access to timers or state or anything like this.
This will be on the foot.
Also the operator topology is determined by the planner, not by you.
The nice thing here is you can focus on your business logic.
And you do this declaratively to optimize your business durations
and make something out of it.
Internally, it uses highly efficient records, also up to the engine, not to you.
What you will see maybe is like a road type.
If you really want to go out of table API, then you see a road type,
which can work as a business program.
And the interesting thing here is that conceptually, we are working with tables here,
tables and views, you know, databases.
But under the hood, there is actually a change level.
And that's what I want to show in the following slides.
But you can also see that, like for example, if you're disappearing here,
you will get this output when you run it in the IDQ.
And you already see that there is, of course, an F0 column with the 123 output.
But there is an additional column, first column,
which already shows that there is some change like attached to every record.
In this case, it's just insert.
The nice thing about Linux APIs is that you can mix and match them.
So you can, for example, start with the S3 API,
and then you go to table API, or the other way around.
If you have SQL, you can do the detail in SQL first.
And then if you have some more complex logic,
like timer services, or like a very complex state,
or whatever, then you can go to the S3 API, do it there,
and then you can switch back to SQL, or you can just use the SQL for Nectar.
But you find the entire pipeline in the S3 API.
So that is up to you.
But yeah, the APIs for that are present to go back and forth between those two.
So now let's really talk about change.s3 processing.
If you think about data processing,
like in most of the cases,
the main data processing is always consuming a stream of changes.
Because if this would not be like a continuous input stream,
then your company, your project, whatever it would actually be,
right?
So like it is actually very common that data flows in continuously.
And the Flinky APIs, the Flink runtime sees everything basically as a stream.
And it just distinguishes between a bounded stream and unbounded stream.
So bounded means you define a start and an end,
and the end was coming there.
Unbounded means you start somewhere,
and now it can be somewhere in the past,
and then you start processing the future.
So this is up to you.
Yeah, if you really think a bit about this,
actually batch processing is just a special case of stream processing.
So batch processing means that through the bounded nature of the stream,
I can maybe do some more specialized operators like sorting, for example.
It's easier in such a thing.
And you can also use different algorithms if you have sorted like this to a sort of a join,
or something like this.
So that the runtime has special operators and special handling of bounded streams.
But in general, you can process everything for the stream.
So both bounded and unbounded data.
So how does actually things look like?
So how can I work with streams and things like that?
So the first answer to this, or I mentioned before,
you actually don't work with streams.
So what you work with is dynamic tables.
So this is just a concept we call the dynamic tables.
It's a concept similar to materialized views and materialized view maintenance.
So what you do as a user is you define your tables.
So on the left side, we have transactions on the right side.
We have maybe revenue.
And then you define, in the middle, you define a standing, running SQL31,
which gets translated into a pipeline methodology next to the private branch.
So then the question is, OK, if we have this big SQL kind of a database,
and the answer to that is no, it's not a database,
because we are not in charge of the data.
So you can bring your own data and your own systems.
So it's more like a process.
It leads from all different kinds of systems.
So if a table is not a stream, or if I don't work with streams,
how does that actually relate with each other?
And an interesting piece of interesting term here is called stream table duality.
So you can basically see a stream as the change log of a continuously changing table.
So it is possible, and I will also show an example shortly,
that you can convert from a table into a stream and from a stream into a table.
You can do a back and forth mapping after this possible.
Usually, as a user, you don't see that.
Under the hood, the runtime, all the sources, all the things, all the operators,
they work with change logs under the hood.
So in Flink, we have four different kinds of change tags for each record.
So we have insertions, insertions are also the default input and output for bounded hash queries.
And then we have update four, which basically removes a previously inputted result.
Then we have update after to update something.
And then we have to feed one of the last results.
When we see only insert only in a log, then we call this an only or insert only log.
If it contains some kind of division or update before,
we call this updating table or an updating stream.
And if it never contains an update before, but only update afters,
then there's a primary key involved, and then we call this absurdity.
So let me make a quick example.
So again, we have on the left side those actions on the right side with value,
and in the middle we have summing and sumproving by name of different sections.
So what happens now is, like in the logical table, there is a new record coming in called Alice.
This is how it will be represented in the change tag under the hood.
And then this is what comes out.
So we are summing here.
So 256 is the first result that we are already looking at.
So now the next variable comes in.
Again, it will be added also to the last table.
But now it comes in.
There's another Alice, and we want to move like this.
So that means the sum is not updated or we need to update the sum to the newest number.
That means, first, we have to remove the old record.
And if we want to materialize our change log into a table,
so we also have to remove the row in the table.
And then we can finally add the updated row in the table.
And this is what the change log looks like.
And if you would apply this change log to a SQL or to some key values there,
or like the search or so, then the result would be there.
And if we would define a primary key on the sync table,
actually we don't need this update before,
because then it would be now searching operation.
And yeah, we can basically save 50% of traffic
if we do not want to support that with the rows in the sync table.
So I already mentioned that each sync and each source,
that they declare a change log model which changes they and they can't consume.
And yeah, I give like a quick example of various connectors.
So when we, for example, read from a file system,
this is usually a scan operation.
So it is very common that when you read from a file that this is just insert only.
There are no updates coming through the file system.
Sometimes they do, but in the general case,
you just scan through the Kafka file for example.
Okay, so Kafka, in the early days,
Kafka was actually just as a log for every N record
that came in through Kafka was also considered like an insert only record.
Then later, Kafka also added some absurd functionality.
So we also have a connector called Kafka Absurd for that.
That means when a value in Kafka is null, it means addition.
So the Kafka Absurd connector, for example,
would produce insertions and divisions.
If you define the JVC connector,
JVC also doesn't have this concept of updates.
So in the same case, we would scan the entire table
and just scan to only produce insertions.
So we have all the insertions for JVC.
But that comes like the most complex case.
What happens if you use, for example, an easy one?
You connect it to the database to consume the change of this particular from the database.
You put this into Kafka and then you consume from Kafka.
In this case, for example, this could, for example,
all kinds of changes that can then be evaluated by the end.
The optimizer basically tracks the changes through the entire topology
and the sync prepares what it can digest.
The optimizer could react sometimes with an error message,
but sometimes there's more to it.
So let's quickly also talk about these two different modes.
So I already said that sometimes you can do upserts
where there's no update before.
And sometimes you need all four kind of changes.
And this is called like retract versus absurd.
So retract has this nice property that there is no primary key required.
This works for almost all external systems, which is great.
You can also support up with the pros, which you cannot support in the upsert.
The table is called an absurd table.
And interestingly, also retracts,
so like this retracting of the previous admitted record
is actually often required in distributed systems.
And I also have a little example on the right side, but I will show shortly.
So let me definitely explain this first.
So this is a count of a count.
So we are creating an histogram.
The lexical variable itself is not so important.
What is important?
What is actually flowing in the cluster through the operators?
So whatever record comes in,
the first operator will identify this.
Okay, this is the first time, so the count is one.
And then since we want to do the count of a count, the next operator
will also count this as a one, and it will keep some state.
How many records have I seen for this particular count?
So now comes the second record in.
And we have to after the count.
So now the count is two, but one anymore.
And interestingly, if we do a hash partition for some collectors,
it might be that the count ends up at a completely different operator.
But what happens with the old count?
So now you have two threads or two operators,
parallel instances of the operator that have a count
and that they need to remove the count in the other operator.
This is why this case rejection is required,
because the update before needs to go to the subclass one
and remove the man outdated record.
But in general, absurd is an optimization.
It reduces traffic, reduces computation.
And if it's possible, it's great,
but usually there is a lot of reflections flowing on in the...
And also have some examples here.
Like if you would do an explain on some SQL query in the SQL,
the bottom part is what you would see.
So let's assume we have a table of transactions and a table of payment
and a table of result.
The table of result can consume all kinds of changes.
I just took my table also.
And you join transactions and payments.
And in the explain, you also see that there is...
You can get information in the explain about the change of mode.
For example, if the input here is insert only,
insert only, then also the join will produce insert only result.
And for example, if we do an outer join,
in this case the left outer join,
then things become a bit more complex.
Here you have insert only, insert only,
but since that outer join will emit another first,
like if there's one thing, like one record comes in,
there's no matching record so far,
and you have to emit another first for the other side,
and then when the other side is coming in,
then you have to remove another again
and emit the final result.
And that's why, for example, here,
you have all kinds of changes coming out of the join.
And then we can even make it more complicated.
What happens if we define a primary key
on transactions and payments?
Then the optimizer will recognize,
okay, that input spec and the right input spec
will contain now a key key.
That is great.
So I can remove the update before,
so you can see that there is one,
that particular is not necessary anymore,
because we can do upserts on the results.
So this query is obviously more efficient than the other one.
And the other good optimizer can also range
between those different modes.
I don't want to get details here,
but if it's possible, like it's necessary,
you can go from updating to the collection.
But that's not also under the course of the
information.
And depending on the operators,
you also can switch between these modes.
So for example, if you have a regular join,
to append only tables,
then also the resulting table will be append only.
And I showed already that
if there's one of the tables updating,
the results will be updating.
And if there's some outer join,
then the result is always updating.
And now comes the interesting part.
If you have append only table,
and you join it to an updating table,
there is a special kind of join,
which we call temporal join.
A temporal join will actually produce
an append only table,
because it looks at the table
at a point, a specific point in time.
That's a very interesting operator.
Unfortunately, we don't have enough time,
but I just want to show you an example
of this very, very useful join operator.
So let's assume we have some orders table,
and orders have a currency,
and there is a currency rates table.
And obviously, you don't want to join those two tables
with the latest currency rates,
but you actually want to know
what was the currency rate at the time
when the order was created.
And this syntax here with the persistent time as of
actually allows you to consume
all the changes from the rates table
and join it with orders
at the point on the order.
This is just one example
of a very sophisticated join operation.
And by the way, for system time as of this season,
we're going to have to be able to understand
so carefully how we see what we're going to do.
So I also have prepared a demo.
I think we still have seven minutes left,
so it should be good to see.
So I also want to show you some of the CPC capabilities.
So I will run everything in my IDE.
I will use Java for this example.
I have a MySQL container,
and I'm running it, so we'll start with a SQL container.
I'm processing it.
And this container will create a MySQL instance,
and it will also be filled already
with, I think, three or four rows.
I have a few examples here.
I can simply run the examples
and the main method of the IDE.
So what I'm doing here
is I'm creating different tables to connect to a SQL.
One is a JPC one,
which fully screens the table once,
and the other one is a CPC one,
which, like, continuously
monitors the tables and the JPC.
So let's just run this.
So here we see the first three results.
As you can see, the application has not stopped.
So it is waiting for more records to come,
and now I want to insert more values into MySQL.
I could have used MySQL to see a line for that,
but I can also use my SQL to see what I see.
So I am having a regular,
I can set into transaction JPC, values, blah, blah, blah,
and I can run this main method here.
So it's a bit overkill to use Spring for that,
for just one value,
but I think it's flexible enough
you can also use it for the hash query
of just setting one record into the database,
and as you can see,
we can show my SQL and from my SQL,
via CPC to the link,
and then to the next CPC.
We have also more sophisticated examples here,
and I don't think that we have more time for that,
but we can do a lot of things with Spring SQL.
I think I could spend a day
and talk a little bit more about the way this works.
So, put on the grid.
Yeah, Spring SQL and it is very powerful,
has been crafted over years and years and years,
by many, many companies, many teams.
It's very flexible for integration,
for integrating various systems of different semantics,
and there is way more.
So, I just showed some operators,
but we have a large, large coverage of SQL standard,
so over-windows support for aggregating,
for Spring, we support the recognized laws
for pattern matching and complex plan processing.
We have time for obsessions, windows for like,
cutting your screen into pieces.
Then there is a huge CPC connector ecosystem,
not part of the fourth thing,
but also quite useful as a little think-of-stars already.
Then, something new, fatal store,
which tries to be like the first streaming data warehouse
kind of thing.
It's not a very early version, but it's very promising.
So, yeah, I would recommend to, yeah,
maybe look into one of these sub-projects as well.
Not only things, it's big, but also the ecosystem,
around the thing, the growths and growths and growths.
I'm happy to take questions.
I think we have three minutes left,
but otherwise I will also speak outside for any questions.
Thank you very much.
Thank you.
Yeah, so like...
Can you please repeat the question?
The question is like, how does it, like,
handle like, transactions that also take a lot of time
before the transaction has ended.
So, in general, I think we are not very good at
transaction handling in things,
but like you have with Data Stream API,
you have a lot of possibilities to,
like for example, you can buffer all the data from this
transaction and stay at it after terrible time you want to,
and then just wait until the transaction closes,
and then you're creating the execution of the transaction,
and that is possible.
So, yeah, personally, I would maybe do some stuff
in the Data Stream API first until the transaction ended,
and then push that.
Thank you.
Yeah, thank you both.
So, in terms of running times,
do I just think from bottom all,
if it's an hybrid, I shift my interpretation?
No, I think it's creating its own,
its own runtime.
So, you will be able to do mutual apps,
some types of software,
and then just look at the,
like, I don't know if there's a lot of data in the stream.
No, I don't.
Is that all there?
Okay.
Let me check.
There's a library that's called the HGIS,
which is not for you to be used,
but we can also buy it,
and that's the idea for a line.
But in general, yeah,
the platform, it's rather a platform,
that's where it looks.
Now, the next logical question,
how far does it scale?
I would say,
as the question was,
how far does it scale,
I can tell you that
probably from the system,
and like,
there's single state,
there's a billion,
all of us,
and all of us in the day or so.
Yeah, and there is Apple,
that processes things,
like all the big banks
use it for credit card fraud detection
and stuff like that.
So, I don't think that,
like most companies in this room,
they will not reach the scalability limits of Link,
because yeah, we are not at,
unless here some Apple or Alibaba people
are in this room,
then maybe, I don't know.
You said that the frame does not own data,
but then there is this table store project,
so is it like move towards,
you know, more like ownership?
Yeah, this table store is a very,
very interesting approach,
like it started last year,
or two years ago,
it's rather new.
I think it was last year, early last year.
It doesn't really fit to Apache Flink,
but it's still, it's very useful,
and yeah, we will see,
maybe it will leave the,
maybe it will leave the Apache Software Foundation soon,
but yeah, not allowed to,
and not the Software Foundation,
but the Flink project itself,
we will see,
because it doesn't fit really well,
but it's in general,
like we still have this vision of Flink as a database.
Yeah, we will see.
Thank you.
Sir, a question.
This is about big states,
because you mentioned that you can have like
terabytes of state,
but when you create a checkpoint,
and if this checkpoint will be very big,
and storage of it can be long,
is it like a huge DC post to write into the store?
Yeah, so the question was,
like how can we actually snapshot large state in general?
And this is exactly where Flink distinguishes,
like where it differs from competitors,
because there is a lot of engineering involved
to make this as efficient as possible.
I'm sure there's even more to do,
it's still not perfectly efficient,
there's more optimizations that you can do,
but for example, there is like differential snapshots
involved, there is local recovery involved,
or like there are many, many algorithms under the hood
to make it as quickly as possible.
But yeah, of course,
like if you have terabytes of state,
and the machine has died completely,
and then you obviously need to restore these terabytes
of state from S3 into your task manager again,
and this can take time.
So like it tries its best,
but yeah, of course,
you need to do benchmarks for your use case.
One last question.
Yeah, so we guarantee exactly once and to end
if the connectors, source and sync support that,
like especially for state,
so there are no duplicates in state,
we might need to reprocess data during a failure,
but yeah, like end to end,
exactly once semantics are possible.
Okay, then yeah, thank you very much,
and I'm waiting outside.