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 >> >
