Add one more message, I have already updated the FLIP[1].
[1]
https://cwiki.apache.org/confluence/display/FLINK/FLIP-149%3A+Introduce+the+upsert-kafka+Connector
Shengkai Fang <fskm...@gmail.com> 于2020年10月23日周五 下午2:55写道：
Hi, all.
It seems we have reached a consensus on the FLIP. If no one has other
objections, I would like to start the vote for FLIP-149.
Best,
Shengkai
Jingsong Li <jingsongl...@gmail.com> 于2020年10月23日周五 下午2:25写道：
Thanks for explanation,
I am OK for `upsert`. Yes, Its concept has been accepted by many
systems.
Best,
Jingsong
On Fri, Oct 23, 2020 at 12:38 PM Jark Wu <imj...@gmail.com> wrote:
Hi Timo,
I have some concerns about `kafka-cdc`,
1) cdc is an abbreviation of Change Data Capture which is commonly
used
for
databases, not for message queues.
2) usually, cdc produces full content of changelog, including
UPDATE_BEFORE, however "upsert kafka" doesn't
3) `kafka-cdc` sounds like a natively support for `debezium-json`
format,
however, it is not and even we don't want
"upsert kafka" to support "debezium-json"
Hi Jingsong,
I think the terminology of "upsert" is fine, because Kafka also uses
"upsert" to define such behavior in their official documentation [1]:
a data record in a changelog stream is interpreted as an UPSERT aka
INSERT/UPDATE
Materialize uses the "UPSERT" keyword to define such behavior too [2].
Users have been requesting such feature using "upsert kafka"
terminology in
user mailing lists [3][4].
Many other systems support "UPSERT" statement natively, such as impala
[5],
SAP [6], Phoenix [7], Oracle NoSQL [8], etc..
Therefore, I think we don't need to be afraid of introducing "upsert"
terminology, it is widely accepted by users.
Best,
Jark
[1]:
https://kafka.apache.org/20/documentation/streams/developer-guide/dsl-api.html#streams_concepts_ktable
[2]:
https://materialize.io/docs/sql/create-source/text-kafka/#upsert-on-a-kafka-topic
[3]:
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/SQL-materialized-upsert-tables-td18482.html#a18503
[4]:
http://apache-flink.147419.n8.nabble.com/Kafka-Sink-AppendStreamTableSink-doesn-t-support-consuming-update-changes-td5959.html
[5]:
https://impala.apache.org/docs/build/html/topics/impala_upsert.html
[6]:
https://help.sap.com/viewer/7c78579ce9b14a669c1f3295b0d8ca16/Cloud/en-US/ea8b6773be584203bcd99da76844c5ed.html
[7]: https://phoenix.apache.org/atomic_upsert.html
[8]:
https://docs.oracle.com/en/database/other-databases/nosql-database/18.3/sqlfornosql/adding-table-rows-using-insert-and-upsert-statements.html
On Fri, 23 Oct 2020 at 10:36, Jingsong Li <jingsongl...@gmail.com>
wrote:
The `kafka-cdc` looks good to me.
We can even give options to indicate whether to turn on compact,
because
compact is just an optimization?
- ktable let me think about KSQL.
- kafka-compacted it is not just compacted, more than that, it still
has
the ability of CDC
- upsert-kafka , upsert is back, and I don't really want to see it
again
since we have CDC
Best,
Jingsong
On Fri, Oct 23, 2020 at 2:21 AM Timo Walther <twal...@apache.org>
wrote:
Hi Jark,
I would be fine with `connector=upsert-kafka`. Another idea would
be to
align the name to other available Flink connectors [1]:
`connector=kafka-cdc`.
Regards,
Timo
[1] https://github.com/ververica/flink-cdc-connectors
On 22.10.20 17:17, Jark Wu wrote:
Another name is "connector=upsert-kafka', I think this can solve
Timo's
concern on the "compacted" word.
Materialize also uses "ENVELOPE UPSERT" [1] keyword to identify
such
kafka
sources.
I think "upsert" is a well-known terminology widely used in many
systems
and matches the
behavior of how we handle the kafka messages.
What do you think?
Best,
Jark
[1]:
https://materialize.io/docs/sql/create-source/text-kafka/#upsert-on-a-kafka-topic
On Thu, 22 Oct 2020 at 22:53, Kurt Young <ykt...@gmail.com>
wrote:
Good validation messages can't solve the broken user
experience,
especially
that
such update mode option will implicitly make half of current
kafka
options
invalid or doesn't
make sense.
Best,
Kurt
On Thu, Oct 22, 2020 at 10:31 PM Jark Wu <imj...@gmail.com>
wrote:
Hi Timo, Seth,
The default value "inserting" of "mode" might be not suitable,
because "debezium-json" emits changelog messages which include
updates.
On Thu, 22 Oct 2020 at 22:10, Seth Wiesman <
s...@ververica.com>
wrote:
+1 for supporting upsert results into Kafka.
I have no comments on the implementation details.
As far as configuration goes, I tend to favor Timo's option
where
we
add
a
"mode" property to the existing Kafka table with default
value
"inserting".
If the mode is set to "updating" then the validation changes
to
the
new
requirements. I personally find it more intuitive than a
seperate
connector, my fear is users won't understand its the same
physical
kafka
sink under the hood and it will lead to other confusion like
does
it
offer
the same persistence guarantees? I think we are capable of
adding
good
valdiation messaging that solves Jark and Kurts concerns.
On Thu, Oct 22, 2020 at 8:51 AM Timo Walther <
twal...@apache.org>
wrote:
Hi Jark,
"calling it "kafka-compacted" can even remind users to
enable
log
compaction"
But sometimes users like to store a lineage of changes in
their
topics.
Indepent of any ktable/kstream interpretation.
I let the majority decide on this topic to not further block
this
effort. But we might find a better name like:
connector = kafka
mode = updating/inserting
OR
connector = kafka-updating
...
Regards,
Timo
On 22.10.20 15:24, Jark Wu wrote:
Hi Timo,
Thanks for your opinions.
1) Implementation
We will have an stateful operator to generate INSERT and
UPDATE_BEFORE.
This operator is keyby-ed (primary key as the shuffle key)
after
the
source
operator.
The implementation of this operator is very similar to the
existing
`DeduplicateKeepLastRowFunction`.
The operator will register a value state using the primary
key
fields
as
keys.
When the value state is empty under current key, we will
emit
INSERT
for
the input row.
When the value state is not empty under current key, we
will
emit
UPDATE_BEFORE using the row in state,
and emit UPDATE_AFTER using the input row.
When the input row is DELETE, we will clear state and emit
DELETE
row.
2) new option vs new connector
We recently simplified the table options to a minimum
amount of
characters to be as concise as possible in the DDL.
I think this is the reason why we want to introduce a new
connector,
because we can simplify the options in DDL.
For example, if using a new option, the DDL may look like
this:
CREATE TABLE users (
user_id BIGINT,
user_name STRING,
user_level STRING,
region STRING,
PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
'connector' = 'kafka',
'model' = 'table',
'topic' = 'pageviews_per_region',
'properties.bootstrap.servers' = '...',
'properties.group.id' = 'testGroup',
'scan.startup.mode' = 'earliest',
'key.format' = 'csv',
'key.fields' = 'user_id',
'value.format' = 'avro',
'sink.partitioner' = 'hash'
);
If using a new connector, we can have a different default
value
for
the
options and remove unnecessary options,
the DDL can look like this which is much more concise:
CREATE TABLE pageviews_per_region (
user_id BIGINT,
user_name STRING,
user_level STRING,
region STRING,
PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
'connector' = 'kafka-compacted',
'topic' = 'pageviews_per_region',
'properties.bootstrap.servers' = '...',
'key.format' = 'csv',
'value.format' = 'avro'
);
When people read `connector=kafka-compacted` they might
not
know
that
it
has ktable semantics. You don't need to enable log
compaction
in
order
to use a KTable as far as I know.
We don't need to let users know it has ktable semantics, as
Konstantin
mentioned this may carry more implicit
meaning than we want to imply here. I agree users don't
need
to
enable
log
compaction, but from the production perspective,
log compaction should always be enabled if it is used in
this
purpose.
Calling it "kafka-compacted" can even remind users to
enable
log
compaction.
I don't agree to introduce "model = table/stream" option,
or
"connector=kafka-table",
because this means we are introducing Table vs Stream
concept
from
KSQL.
However, we don't have such top-level concept in Flink SQL
now,
this
will
further confuse users.
In Flink SQL, all the things are STREAM, the differences
are
whether
it
is
bounded or unbounded,
whether it is insert-only or changelog.
Best,
Jark
On Thu, 22 Oct 2020 at 20:39, Timo Walther <
twal...@apache.org>
wrote:
Hi Shengkai, Hi Jark,
thanks for this great proposal. It is time to finally
connect
the
changelog processor with a compacted Kafka topic.
"The operator will produce INSERT rows, or additionally
generate
UPDATE_BEFORE rows for the previous image, or produce
DELETE
rows
with
all columns filled with values."
Could you elaborate a bit on the implementation details in
the
FLIP?
How
are UPDATE_BEFOREs are generated. How much state is
required to
perform
this operation.
From a conceptual and semantical point of view, I'm
fine
with
the
proposal. But I would like to share my opinion about how
we
expose
this
feature:
ktable vs kafka-compacted
I'm against having an additional connector like `ktable`
or
`kafka-compacted`. We recently simplified the table
options
to
a
minimum
amount of characters to be as concise as possible in the
DDL.
Therefore,
I would keep the `connector=kafka` and introduce an
additional
option.
Because a user wants to read "from Kafka". And the "how"
should
be
determined in the lower options.
When people read `connector=ktable` they might not know
that
this
is
Kafka. Or they wonder where `kstream` is?
When people read `connector=kafka-compacted` they might
not
know
that
it
has ktable semantics. You don't need to enable log
compaction
in
order
to use a KTable as far as I know. Log compaction and table
semantics
are
orthogonal topics.
In the end we will need 3 types of information when
declaring a
Kafka
connector:
CREATE TABLE ... WITH (
connector=kafka -- Some information about the
connector
end-offset = XXXX -- Some information about the
boundedness
model = table/stream -- Some information about
interpretation
)
We can still apply all the constraints mentioned in the
FLIP.
When
`model` is set to `table`.
What do you think?
Regards,
Timo
On 21.10.20 14:19, Jark Wu wrote:
Hi,
IMO, if we are going to mix them in one connector,
1) either users need to set some options to a specific
value
explicitly,
e.g. "scan.startup.mode=earliest",
"sink.partitioner=hash",
etc..
This makes the connector awkward to use. Users may face
to
fix
options
one
by one according to the exception.
Besides, in the future, it is still possible to use
"sink.partitioner=fixed" (reduce network cost) if users
are
aware
of
the partition routing,
however, it's error-prone to have "fixed" as default for
compacted
mode.
2) or make those options a different default value when
"compacted=true".
This would be more confusing and unpredictable if the
default
value
of
options will change according to other options.
What happens if we have a third mode in the future?
In terms of usage and options, it's very different from
the
original "kafka" connector.
It would be more handy to use and less fallible if
separating
them
into
two
connectors.
In the implementation layer, we can reuse code as much as
possible.
Therefore, I'm still +1 to have a new connector.
The "kafka-compacted" name sounds good to me.
Best,
Jark
On Wed, 21 Oct 2020 at 17:58, Konstantin Knauf <
kna...@apache.org>
wrote:
Hi Kurt, Hi Shengkai,
thanks for answering my questions and the additional
clarifications. I
don't have a strong opinion on whether to extend the
"kafka"
connector
or
to introduce a new connector. So, from my perspective
feel
free
to
go
with
a separate connector. If we do introduce a new
connector I
wouldn't
call it
"ktable" for aforementioned reasons (In addition, we
might
suggest
that
there is also a "kstreams" connector for symmetry
reasons). I
don't
have a
good alternative name, though, maybe "kafka-compacted"
or
"compacted-kafka".
Thanks,
Konstantin
On Wed, Oct 21, 2020 at 4:43 AM Kurt Young <
ykt...@gmail.com
wrote:
Hi all,
I want to describe the discussion process which drove
us
to
have
such
conclusion, this might make some of
the design choices easier to understand and keep
everyone on
the
same
page.
Back to the motivation, what functionality do we want
to
provide
in
the
first place? We got a lot of feedback and
questions from mailing lists that people want to write
Not-Insert-Only
messages into kafka. They might be
intentional or by accident, e.g. wrote an non-windowed
aggregate
query
or
non-windowed left outer join. And
some users from KSQL world also asked about why Flink
didn't
leverage
the
Key concept of every kafka topic
and make kafka as a dynamic changing keyed table.
To work with kafka better, we were thinking to extend
the
functionality
of
the current kafka connector by letting it
accept updates and deletions. But due to the limitation
of
kafka,
the
update has to be "update by key", aka a table
with primary key.
This introduces a couple of conflicts with current
kafka
table's
options:
1. key.fields: as said above, we need the kafka table
to
have
the
primary
key constraint. And users can also configure
key.fields freely, this might cause friction. (Sure we
can
do
some
sanity
check on this but it also creates friction.)
2. sink.partitioner: to make the semantics right, we
need to
make
sure
all
the updates on the same key are written to
the same kafka partition, such we should force to use a
hash
by
key
partition inside such table. Again, this has conflicts
and creates friction with current user options.
The above things are solvable, though not perfect or
most
user
friendly.
Let's take a look at the reading side. The keyed kafka
table
contains
two
kinds of messages: upsert or deletion. What upsert
means is "If the key doesn't exist yet, it's an insert
record.
Otherwise
it's an update record". For the sake of correctness or
simplicity, the Flink SQL engine also needs such
information.
If
we
interpret all messages to "update record", some queries
or
operators may not work properly. It's weird to see an
update
record
but
you
haven't seen the insert record before.
So what Flink should do is after reading out the
records
from
such
table,
it needs to create a state to record which messages
have
been seen and then generate the correct row type
correspondingly.
This
kind
of couples the state and the data of the message
queue, and it also creates conflicts with current kafka
connector.
Think about if users suspend a running job (which
contains
some
reading
state now), and then change the start offset of the
reader.
By changing the reading offset, it actually change the
whole
story
of
"which records should be insert messages and which
records
should be update messages). And it will also make Flink
to
deal
with
another weird situation that it might receive a
deletion
on a non existing message.
We were unsatisfied with all the frictions and
conflicts
it
will
create
if
we enable the "upsert & deletion" support to the
current
kafka
connector. And later we begin to realize that we
shouldn't
treat
it
as
a
normal message queue, but should treat it as a changing
keyed
table. We should be able to always get the whole data
of
such
table
(by
disabling the start offset option) and we can also read
the
changelog out of such table. It's like a HBase table
with
binlog
support
but doesn't have random access capability (which can be
fulfilled
by Flink's state).
So our intention was instead of telling and persuading
users
what
kind
of
options they should or should not use by extending
current kafka connector when enable upsert support, we
are
actually
create
a whole new and different connector that has total
different abstractions in SQL layer, and should be
treated
totally
different with current kafka connector.
Hope this can clarify some of the concerns.
Best,
Kurt
On Tue, Oct 20, 2020 at 5:20 PM Shengkai Fang <
fskm...@gmail.com
wrote:
Hi devs,
As many people are still confused about the difference
option
behaviours
between the Kafka connector and KTable connector, Jark
and
I
list
the
differences in the doc[1].
Best,
Shengkai
[1]
https://docs.google.com/document/d/13oAWAwQez0lZLsyfV21BfTEze1fc2cz4AZKiNOyBNPk/edit
Shengkai Fang <fskm...@gmail.com> 于2020年10月20日周二
下午12:05写道：
Hi Konstantin,
Thanks for your reply.
It uses the "kafka" connector and does not specify a
primary
key.
The dimensional table `users` is a ktable connector
and we
can
specify
the
pk on the KTable.
Will it possible to use a "ktable" as a dimensional
table
in
FLIP-132
Yes. We can specify the watermark on the KTable and
it
can
be
used
as a
dimension table in temporal join.
Introduce a new connector vs introduce a new
property
The main reason behind is that the KTable connector
almost
has
no
common
options with the Kafka connector. The options that
can
be
reused
by
KTable
connectors are 'topic',
'properties.bootstrap.servers'
and
'value.fields-include' . We can't set cdc format for
'key.format'
and
'value.format' in KTable connector now, which is
available
in
Kafka
connector. Considering the difference between the
options
we
can
use,
it's
more suitable to introduce an another connector
rather
than
a
property.
We are also fine to use "compacted-kafka" as the name
of
the
new
connector. What do you think?
Best,
Shengkai
Konstantin Knauf <kna...@apache.org> 于2020年10月19日周一
下午10:15写道：
Hi Shengkai,
Thank you for driving this effort. I believe this a
very
important
feature
for many users who use Kafka and Flink SQL
together. A
few
questions
and
thoughts:
* Is your example "Use KTable as a
reference/dimension
table"
correct?
It
uses the "kafka" connector and does not specify a
primary
key.
* Will it be possible to use a "ktable" table
directly
as a
dimensional
table in temporal join (*based on event time*)
(FLIP-132)?
This
is
not
completely clear to me from the FLIP.
* I'd personally prefer not to introduce a new
connector
and
instead
to
extend the Kafka connector. We could add an
additional
property
"compacted"
= "true"|"false". If it is set to "true", we can add
additional
validation
logic (e.g. "scan.startup.mode" can not be set,
primary
key
required,
etc.). If we stick to a separate connector I'd not
call
it
"ktable",
but
rather "compacted-kafka" or similar. KTable seems to
carry
more
implicit
meaning than we want to imply here.
* I agree that this is not a bounded source. If we
want
to
support a
bounded mode, this is an orthogonal concern that
also
applies
to
other
unbounded sources.
Best,
Konstantin
On Mon, Oct 19, 2020 at 3:26 PM Jark Wu <
imj...@gmail.com>
wrote:
Hi Danny,
First of all, we didn't introduce any concepts from
KSQL
(e.g.
Stream
vs
Table notion).
This new connector will produce a changelog stream,
so
it's
still
a
dynamic
table and doesn't conflict with Flink core
concepts.
The "ktable" is just a connector name, we can also
call
it
"compacted-kafka" or something else.
Calling it "ktable" is just because KSQL users can
migrate
to
Flink
SQL
easily.
Regarding to why introducing a new connector vs a
new
property
in
existing
kafka connector:
I think the main reason is that we want to have a
clear
separation
for
such
two use cases, because they are very different.
We also listed reasons in the FLIP, including:
1) It's hard to explain what's the behavior when
users
specify
the
start
offset from a middle position (e.g. how to process
non
exist
delete
events).
It's dangerous if users do that. So we don't
provide
the
offset
option
in the new connector at the moment.
2) It's a different perspective/abstraction on the
same
kafka
topic
(append
vs. upsert). It would be easier to understand if we
can
separate
them
instead of mixing them in one connector. The
new
connector
requires
hash sink partitioner, primary key declared,
regular
format.
If we mix them in one connector, it might be
confusing
how
to
use
the
options correctly.
3) The semantic of the KTable connector is just the
same
as
KTable
in
Kafka
Stream. So it's very handy for Kafka Stream and
KSQL
users.
We have seen several questions in the
mailing
list
asking
how
to
model
a KTable and how to join a KTable in Flink SQL.
Best,
Jark
On Mon, 19 Oct 2020 at 19:53, Jark Wu <
imj...@gmail.com
wrote:
Hi Jingsong,
As the FLIP describes, "KTable connector produces
a
changelog
stream,
where each data record represents an update or
delete
event.".
Therefore, a ktable source is an unbounded stream
source.
Selecting
a
ktable source is similar to selecting a kafka
source
with
debezium-json
format
that it never ends and the results are
continuously
updated.
It's possible to have a bounded ktable source in
the
future,
for
example,
add an option 'bounded=true' or 'end-offset=xxx'.
In this way, the ktable will produce a bounded
changelog
stream.
So I think this can be a compatible feature in the
future.
I don't think we should associate with ksql
related
concepts.
Actually,
we
didn't introduce any concepts from KSQL (e.g.
Stream vs
Table
notion).
The "ktable" is just a connector name, we can also
call
it
"compacted-kafka" or something else.
Calling it "ktable" is just because KSQL users can
migrate
to
Flink
SQL
easily.
Regarding the "value.fields-include", this is an
option
introduced
in
FLIP-107 for Kafka connector.
I think we should keep the same behavior with the
Kafka
connector.
I'm
not
sure what's the default behavior of KSQL.
But I guess it also stores the keys in value from
this
example
docs
(see
the "users_original" table) [1].
Best,
Jark
[1]:
https://docs.confluent.io/current/ksqldb/tutorials/basics-local.html#create-a-stream-and-table
On Mon, 19 Oct 2020 at 18:17, Danny Chan <
yuzhao....@gmail.com>
wrote:
The concept seems conflicts with the Flink
abstraction
“dynamic
table”,
in Flink we see both “stream” and “table” as a
dynamic
table,
I think we should make clear first how to express
stream
and
table
specific features on one “dynamic table”,
it is more natural for KSQL because KSQL takes
stream
and
table
as
different abstractions for representing
collections.
In
KSQL,
only
table is
mutable and can have a primary key.
Does this connector belongs to the “table” scope
or
“stream”
scope
?
Some of the concepts (such as the primary key on
stream)
should
be
suitable for all the connectors, not just Kafka,
Shouldn’t
this
be
an
extension of existing Kafka connector instead of
a
totally
new
connector ?
What about the other connectors ?
Because this touches the core abstraction of
Flink, we
better
have
a
top-down overall design, following the KSQL
directly
is
not
the
answer.
P.S. For the source
Shouldn’t this be an extension of existing Kafka
connector
instead
of
a
totally new connector ?
How could we achieve that (e.g. set up the
parallelism
correctly) ?
Best,
Danny Chan
在 2020年10月19日 +0800 PM5:17，Jingsong Li <
jingsongl...@gmail.com
，写道：
Thanks Shengkai for your proposal.
+1 for this feature.
Future Work: Support bounded KTable source
I don't think it should be a future work, I
think
it
is
one
of
the
important concepts of this FLIP. We need to
understand
it
now.
Intuitively, a ktable in my opinion is a bounded
table
rather
than
a
stream, so select should produce a bounded table
by
default.
I think we can list Kafka related knowledge,
because
the
word
`ktable`
is
easy to associate with ksql related concepts.
(If
possible,
it's
better
to
unify with it)
What do you think?
value.fields-include
What about the default behavior of KSQL?
Best,
Jingsong
On Mon, Oct 19, 2020 at 4:33 PM Shengkai Fang <
fskm...@gmail.com
wrote:
Hi, devs.
Jark and I want to start a new FLIP to
introduce
the
KTable
connector. The
KTable is a shortcut of "Kafka Table", it also
has
the
same
semantics
with
the KTable notion in Kafka Stream.
FLIP-149:
https://cwiki.apache.org/confluence/display/FLINK/FLIP-149%3A+Introduce+the+KTable+Connector
Currently many users have expressed their needs
for
the
upsert
Kafka
by
mail lists and issues. The KTable connector has
several
benefits
for
users:
1. Users are able to interpret a compacted
Kafka
Topic
as
an
upsert
stream
in Apache Flink. And also be able to write a
changelog
stream
to
Kafka
(into a compacted topic).
2. As a part of the real time pipeline, store
join
or
aggregate
result (may
contain updates) into a Kafka topic for further
calculation;
3. The semantic of the KTable connector is just
the
same
as
KTable
in
Kafka
Stream. So it's very handy for Kafka Stream and
KSQL
users.
We
have
seen
several questions in the mailing list asking
how
to
model a
KTable
and how
to join a KTable in Flink SQL.
We hope it can expand the usage of the Flink
with
Kafka.
I'm looking forward to your feedback.
Best,
Shengkai
--
Best, Jingsong Lee
--
Konstantin Knauf
https://twitter.com/snntrable
https://github.com/knaufk
--
Konstantin Knauf
https://twitter.com/snntrable
https://github.com/knaufk
--
Seth Wiesman | Solutions Architect
+1 314 387 1463
<https://www.ververica.com/>
Follow us @VervericaData
--
Join Flink Forward <https://flink-forward.org/> - The Apache
Flink
Conference
Stream Processing | Event Driven | Real Time
--
Best, Jingsong Lee
--
Best, Jingsong Lee
