Hi Piotr Thanks for your feedback.
> - stateless operators, could completely ignore the issue and process the records normally, as they are doing right now > - stateful operators, should either: > - if the business doesn't require ordering, they could process the records immediately > - or buffer the records internally, like currently windowed/temporal operators are doing. Non windowed joins/aggregations could also work in a similar manner, like pre-aggregate data per each "epoch" (as demarcated by timestamp barriers). > - sinks implementation would have to match what external system support: > - if the external system requires ordered writes (something like Kafka topic?), the sinks would have to buffer the writes until a "timestamp barrier" arrives > - some sinks might support writing the data simultaneously to different "epochs". For example writing files bucketed by each epoch. Each bucket/epoch could be committed independently It sounds good to me and I totally agree with the proposal. We need to give users more choices to meet different business needs and storage support. I have updated the key points in the FLINK section[1] > Ok, I get it now. Indeed the terminology is confusing. Maybe we shouldn't say that the timestamp barrier has been committed, but that all records for given "epoch" have been processed/written, but not yet committed, so they can still be rolled-back? Nice! According to your suggestion, I have updated the FLIP for "epoch" as: 1. It is PROCESSED when records are written to a table 2. It is WRITTEN when the records are in a snapshot 3. It is PRECOMMIT when all tables are PROCESSED but not WRITTEN 4. It is COMMIT when all tables are WRITTEN Records not WRITTEN in a table will be rolled back due to job failure. > Why do we need to do that? Only to disallow this? To forbid writing from two jobs into a single table? If so, can we not push this responsibility down to the connector? Like sink/source operator coordinators should negotiate with respective external systems if the given read/write is allowed? So if there is a need for such meta service, Flink doesn't need to know about it? As I mentioned, MetaService will do some atomic operations to check and disallow some operations when jobs are submitted concurrently. But I'm sorry that I may not have explained the relationship between it and sink/source clearly. Generally speaking, the interactive between Flink and MetaService is as: 1. When the Client submits a flink job (streaming&batch), it interacts with MetaService through Catalog in CatalogManager, including getting the table version, registering the source/link table relationship for ETL. 2. When the flink job is running, JobManager collects data processing progress (Timestamp Barrier and Checkpoint) from source/link subtasks and reports them to MetaService. We can implement the above functions in a MetaService node. Of course, it can also be based on an atomic system (such as Zookeeper), with Client and JobManager doing their own work. Of course, source and sink also need some special work, such as reading timestamp barrier, collecting timestamp barrier, writing timestamp barrier, etc. But source/sink subtasks will not interact with MetaService directly. [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-276%3A+Data+Consistency+of+Streaming+and+Batch+ETL+in+Flink+and+Table+Store#FLIP276:DataConsistencyofStreamingandBatchETLinFlinkandTableStore-GlobalTimestampBarrierMechanism Best, Shammon On Tue, Feb 7, 2023 at 1:26 AM Piotr Nowojski <pnowoj...@apache.org> wrote: > Hi, > > Thanks for the answers. > > >> Are you proposing that all of the inputs to stateful operators would > have to be sorted? > >> > > Records in stream don't need to be sorted, but it should be managed by > `Timestamp Barrier`, which means > > 1. Records belonging to a specific `Timestamp Barrier` are disordered. > > 2. Computations in different timestamp barriers are ordered. For the > above > > example, each stateful subtask can start computation for T2 only after it > > finishes computation for T1. Subtasks are independent of each other. > > Wouldn't that add significant latency to processing the records? You would > basically introduce a batch processing concept in Flink? > > Have you considered some alternative solutions? Like for example letting > each operator/function/sink to take care of the data disorder? For example: > - stateless operators, could completely ignore the issue and process the > records normally, as they are doing right now > - stateful operators, should either: > - if the business doesn't require ordering, they could process the > records immediately > - or buffer the records internally, like currently windowed/temporal > operators are doing. Non windowed joins/aggregations could also work in a > similar manner, like pre-aggregate data per each "epoch" (as demarcated by > timestamp barriers). > - sinks implementation would have to match what external system support: > - if the external system requires ordered writes (something like Kafka > topic?), the sinks would have to buffer the writes until a "timestamp > barrier" arrives > - some sinks might support writing the data simultaneously to different > "epochs". For example writing files bucketed by each epoch. Each > bucket/epoch could be committed independently > > This way, latency would be behaving very much like it currently does in > Flink. For example if we have a following streaming SQL: > > INSERT INTO alerts_with_user SELECT * FROM alerts a, users u WHERE > a.user_id = u.id > > If there is some lag in the users table, alerts would be still generated. > Downstream applications could process and react to newly generated > `alerts_with_user`, while at the same time, we could have a consistent view > across those three tables (users, alerts, alerts_with_user) if needed. > > > I call the data of the timetamp barrier "committed" if the data > > is written to a table according to the barrier without a snapshot, and > the > > data may be "rolled back" due to job failure. (sorry that the "committed" > > here may not be appropriate) > > Ok, I get it now. Indeed the terminology is confusing. Maybe we shouldn't > say that the timestamp barrier has been committed, but that all records for > given "epoch" have been processed/written, but not yet committed, so they > can still be rolled-back? > > > For example, when multiple jobs start at the same time and register > themselves in `MetaService`, > > it needs to serially check whether they write to the same table > > Why do we need to do that? Only to disallow this? To forbid writing from > two jobs into a single table? If so, can we not push this responsibility > down to the connector? Like sink/source operator coordinators should > negotiate with respective external systems if the given read/write is > allowed? So if there is a need for such meta service, Flink doesn't need to > know about it? > > Best, > Piotrek > > pon., 6 lut 2023 o 10:44 Shammon FY <zjur...@gmail.com> napisał(a): > > > Hi Piotr, > > > > Thanks for your feedback. In general, I think `Timesamp Barrier` is a > > special `Watermark` that all sources send watermarks with the same > > timestamp as `Timestamp Barrier` and aggregation operators will align > data > > by it. For example, all source subtasks are assigned two unified > watermarks > > T1 and T2, T1 < T2. All records with timestamp <= T1 will be aligned by > T1, > > and records with timestamp (T1, T2] will be aligned by T2. > > > > > Are you proposing that all of the inputs to stateful operators would > have > > to be sorted? > > > > Records in stream don't need to be sorted, but it should be managed by > > `Timestamp Barrier`, which means > > 1. Records belonging to a specific `Timestamp Barrier` are disordered. > > 2. Computations in different timestamp barriers are ordered. For the > above > > example, each stateful subtask can start computation for T2 only after it > > finishes computation for T1. Subtasks are independent of each other. > > > > > Can you explain why do you need those 3 states? Why can committed > records > > be rolled back? > > > > Here I try to define the states of data in tables according to Timestamp > > Barrier and Snapshot, and I found that the 3 states are incomplete. For > > example, there is timestamp barrier T associated with checkpoint P, and > > sink operator will create snapshot S for P in tables. The data states in > > tables are as follows > > 1. Sink finishes writing data of timestamp barrier T to a table, but > > snapshot P is not created in the table and T is not finished in all > tables. > > 2. Sink finishes writing data of timestamp barrier T to a table, creates > > snapshot P according to checkpoint C, but the T1 is not finished in all > > tables. > > 3. Timestamp barrier T is finished in all tables, but snapshot P is not > > created in all tables. > > 4. Timestamp barrier T is finished in all tables, and snapshot P is > created > > in all tables too. > > > > Currently users can only get data from snapshots in Table Store and other > > storages such as Iceberg. Users can get different "versioned" data from > > tables according to their data freshness and consistency requirements. > > I think we should support getting data with a timestamp barrier even > before > > the sink operator finishes creating the snapshot in the future. In this > > situation, I call the data of the timetamp barrier "committed" if the > data > > is written to a table according to the barrier without a snapshot, and > the > > data may be "rolled back" due to job failure. (sorry that the "committed" > > here may not be appropriate) > > > > > I'm not sure if I follow. Generally speaking, why do we need > MetaService > > at all? Why can we only support writes to and reads from TableStore, and > > not any source/sink that implements some specific interface? > > > > It's a good point. I added a `MetaService` node in FLIP mainly to perform > > some atomic operations. For example, when multiple jobs start at the same > > time and register themselves in `MetaService`, it needs to serially check > > whether they write to the same table. If we do not use an > > independent `MetaService Node`, we may need to introduce some other > "atomic > > dependency" such as ZooKeeper. But removing `MetaService Node` can make > the > > system more flexible, I think it's also valuable. Maybe we can carefully > > design MetaService API and support different deployment modes in the next > > FLIP? WDYT? > > > > > > Best, > > Shammon > > > > > > On Fri, Feb 3, 2023 at 10:43 PM Piotr Nowojski <pnowoj...@apache.org> > > wrote: > > > > > Hi Shammon, > > > > > > Thanks for pushing the topic further. I'm not sure how this new > proposal > > is > > > supposed to be working? How should timestamp barrier interplay with > event > > > time and watermarks? Or is timestamp barrier supposed to completely > > replace > > > watermarks? > > > > > > > stateful and temporal operators should align them (records) according > > to > > > their timestamp field. > > > > > > Are you proposing that all of the inputs to stateful operators would > have > > > to be sorted? > > > > > > > There're three states in a table for specific transaction : > PreCommit, > > > Commit and Snapshot > > > > > > Can you explain why do you need those 3 states? Why can committed > records > > > be rolled back? > > > > > > >> 10. Have you considered proposing a general consistency mechanism > > > instead > > > >> of restricting it to TableStore+ETL graphs? For example, it seems to > > me > > > to > > > >> be possible and valuable to define instead the contract that > > > sources/sinks > > > >> need to implement in order to participate in globally consistent > > > snapshots. > > > > > > > > A general consistency mechanism is cool! In my mind, the overall > > > > `consistency system` consists of three components: Streaming & Batch > > ETL, > > > > Streaming & Batch Storage and MetaService. MetaService is decoupled > > from > > > > Storage Layer, but it stores consistency information in persistent > > > storage. > > > > It can be started as an independent node or a component in a large > > Flink > > > > cluster. In the FLIP we use TableStore as the Storage Layer. As you > > > > mentioned, we plan to implement specific source and sink on the > > > TableStore > > > > in the first phase, and may consider other storage in the future > > > > > > I'm not sure if I follow. Generally speaking, why do we need > MetaService > > at > > > all? Why can we only support writes to and reads from TableStore, and > not > > > any source/sink that implements some specific interface? > > > > > > Best, > > > Piotrek > > > > > > niedz., 29 sty 2023 o 12:11 Shammon FY <zjur...@gmail.com> napisał(a): > > > > > > > Hi @Vicky > > > > > > > > Thank you for your suggestions about consistency and they're very > nice > > to > > > > me! > > > > > > > > I have updated the examples and consistency types[1] in FLIP. In > > > general, I > > > > regard the Timestamp Barrier processing as a transaction and divide > the > > > > data consistency supported in FLIP into three types > > > > > > > > 1. Read Uncommitted: Read data from tables even when a transaction is > > not > > > > committed. > > > > 2. Read Committed: Read data from tables according to the committed > > > > transaction. > > > > 3. Repeatable Read: Read data from tables according to the committed > > > > transaction in snapshots. > > > > > > > > You can get more information from the updated FLIP. Looking forward > to > > > your > > > > feedback, THX > > > > > > > > > > > > [1] > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-276%3A+Data+Consistency+of+Streaming+and+Batch+ETL+in+Flink+and+Table+Store#FLIP276:DataConsistencyofStreamingandBatchETLinFlinkandTableStore-DataConsistencyType > > > > > > > > Best, > > > > Shammon > > > > > > > > > > > > On Sat, Jan 28, 2023 at 4:42 AM Vasiliki Papavasileiou > > > > <vpapavasile...@confluent.io.invalid> wrote: > > > > > > > > > Hi Shammon, > > > > > > > > > > > > > > > Thank you for opening this FLIP which is very interesting and such > an > > > > > important feature to add to the Flink ecosystem. I have a couple of > > > > > suggestions/questions: > > > > > > > > > > > > > > > > > > > > - > > > > > > > > > > Consistency is a very broad term with different meanings. There > > are > > > > many > > > > > variations between the two extremes of weak and strong > consistency > > > > that > > > > > tradeoff latency for consistency. https://jepsen.io/consistency > > It > > > > > would > > > > > be great if we could devise an approach that allows the user to > > > choose > > > > > which consistency level they want to use for a query. > > > > > > > > > > > > > > > Example: In your figure where you have a DAG, assume a user queries > > > only > > > > > Table1 for a specific key. Then, a failure happens and the table > > > restores > > > > > from a checkpoint. The user issues the same query, looking up the > > same > > > > key. > > > > > What value does she see? With monotonic-reads, the system > guarantees > > > that > > > > > she will only see the same or newer values but not older, hence > will > > > not > > > > > experience time-travel. This is a very useful property for a system > > to > > > > have > > > > > albeit it is at the weaker-end of consistency guarantees. But it > is a > > > > good > > > > > stepping stone. > > > > > > > > > > > > > > > Another example, assume the user queries Table1 for key K1 and gets > > the > > > > > value V11. Then, she queries Table2 that is derived from Table1 for > > the > > > > > same key, K1, that returns value V21. What is the relationship > > between > > > > V21 > > > > > and V11? Is V21 derived from V11 or can it be an older value V1 > (the > > > > > previous value of K1)? What if value V21 is not yet in table > Table2? > > > What > > > > > should she see when she queries Table1? Should she see the key V11 > or > > > > not? > > > > > Should the requirement be that a record is not visible in any of > the > > > > tables > > > > > in a DAG unless it is available in all of them? > > > > > > > > > > > > > > > > > > > > - > > > > > > > > > > It would we good to have a set of examples with consistency > > > anomalies > > > > > that can happen (like the examples above) and what consistency > > > levels > > > > we > > > > > want the system to offer to prevent them. > > > > > Moreover, for each such example, it would be good to have a > > > > description > > > > > of how the approach (Timestamp Barriers) will work in practice > to > > > > > prevent > > > > > such anomalies. > > > > > > > > > > > > > > > Thank you, > > > > > Vicky > > > > > > > > > > > > > > > On Fri, Jan 27, 2023 at 4:46 PM John Roesler <vvcep...@apache.org> > > > > wrote: > > > > > > > > > > > Hello Shammon and all, > > > > > > > > > > > > Thanks for this FLIP! I've been working toward this kind of > global > > > > > > consistency across large scale data infrastructure for a long > time, > > > and > > > > > > it's fantastic to see a high-profile effort like this come into > > play. > > > > > > > > > > > > I have been lurking in the discussion for a while and delaying my > > > > > response > > > > > > while I collected my thoughts. However, I've realized at some > > point, > > > > > > delaying more is not as useful as just asking a few questions, so > > I'm > > > > > sorry > > > > > > if some of this seems beside the point. I'll number these to not > > > > collide > > > > > > with prior discussion points: > > > > > > > > > > > > 10. Have you considered proposing a general consistency mechanism > > > > instead > > > > > > of restricting it to TableStore+ETL graphs? For example, it seems > > to > > > me > > > > > to > > > > > > be possible and valuable to define instead the contract that > > > > > sources/sinks > > > > > > need to implement in order to participate in globally consistent > > > > > snapshots. > > > > > > > > > > > > 11. It seems like this design is assuming that the "ETL Topology" > > > under > > > > > > the envelope of the consistency model is a well-ordered set of > > jobs, > > > > but > > > > > I > > > > > > suspect this is not the case for many organizations. It may be > > > > > > aspirational, but I think the gold-standard here would be to > > provide > > > an > > > > > > entire organization with a consistency model spanning a loosely > > > coupled > > > > > > ecosystem of jobs and data flows spanning teams and systems that > > are > > > > > > organizationally far apart. > > > > > > > > > > > > I realize that may be kind of abstract. Here's some examples of > > > what's > > > > on > > > > > > my mind here: > > > > > > > > > > > > 11a. Engineering may operate one Flink cluster, and some other > org, > > > > like > > > > > > Finance may operate another. In most cases, those are separate > > > domains > > > > > that > > > > > > don't typically get mixed together in jobs, but some people, like > > the > > > > > CEO, > > > > > > would still benefit from being able to make a consistent query > that > > > > spans > > > > > > arbitrary contexts within the business. How well can a feature > like > > > > this > > > > > > transcend a single Flink infrastructure? Does it make sense to > > > > consider a > > > > > > model in which snapshots from different domains can be > composable? > > > > > > > > > > > > 11b. Some groups may have a relatively stable set of long-running > > > jobs, > > > > > > while others (like data science, skunkworks, etc) may adopt a > more > > > > > > experimental, iterative approach with lots of jobs entering and > > > exiting > > > > > the > > > > > > ecosystem over time. It's still valuable to have them participate > > in > > > > the > > > > > > consistency model, but it seems like the consistency system will > > have > > > > to > > > > > > deal with more chaos than I see in the design. For example, how > can > > > > this > > > > > > feature tolerate things like zombie jobs (which are registered in > > the > > > > > > system, but fail to check in for a long time, and then come back > > > > later). > > > > > > > > > > > > 12. I didn't see any statements about patterns like cycles in the > > ETL > > > > > > Topology. I'm aware that there are fundamental constraints on how > > > well > > > > > > cyclic topologies can be supported by a distributed snapshot > > > algorithm. > > > > > > However, there are a range of approaches/compromises that we can > > > apply > > > > to > > > > > > cyclic topologies. At the very least, we can state that we will > > > detect > > > > > > cycles and produce a warning, etc. > > > > > > > > > > > > 13. I'm not sure how heavily you're waiting the query syntax part > > of > > > > the > > > > > > proposal, so please feel free to defer this point. It looked to > me > > > like > > > > > the > > > > > > proposal assumes people want to query either the latest > consistent > > > > > snapshot > > > > > > or the latest inconsistent state. However, it seems like there's > a > > > > > > significant opportunity to maintain a manifest of historical > > > snapshots > > > > > and > > > > > > allow people to query as of old points in time. That can be > > valuable > > > > for > > > > > > individuals answering data questions, building products, and > > > crucially > > > > > > supporting auditability use cases. To that latter point, it seems > > > nice > > > > to > > > > > > provide not only a mechanism to query arbitrary snapshots, but > also > > > to > > > > > > define a TTL/GC model that allows users to keep hourly snapshots > > for > > > N > > > > > > hours, daily snapshots for N days, weekly snapshots for N weeks, > > and > > > > the > > > > > > same for monthly, quarterly, and yearly snapshots. > > > > > > > > > > > > Ok, that's all I have for now :) I'd also like to understand some > > > > > > lower-level details, but I wanted to get these high-level > questions > > > off > > > > > my > > > > > > chest. > > > > > > > > > > > > Thanks again for the FLIP! > > > > > > -John > > > > > > > > > > > > On 2023/01/13 11:43:28 Shammon FY wrote: > > > > > > > Hi Piotr, > > > > > > > > > > > > > > I discussed with @jinsong lee about `Timestamp Barrier` and > > > `Aligned > > > > > > > Checkpoint` for data consistency in FLIP, we think there are > many > > > > > defects > > > > > > > indeed in using `Aligned Checkpoint` to support data > consistency > > as > > > > you > > > > > > > mentioned. > > > > > > > > > > > > > > According to our historical discussion, I think we have reached > > an > > > > > > > agreement on an important point: we finally need `Timestamp > > Barrier > > > > > > > Mechanism` to support data consistency. But according to our > > > > (@jinsong > > > > > > lee > > > > > > > and I) opinions, the total design and implementation based on > > > > > 'Timestamp > > > > > > > Barrier' will be too complex, and it's also too big in one > FLIP. > > > > > > > > > > > > > > So we‘d like to use FLIP-276[1] as an overview design of data > > > > > consistency > > > > > > > in Flink Streaming and Batch ETL based on `Timestamp Barrier`. > > > > @jinsong > > > > > > and > > > > > > > I hope that we can reach an agreement on the overall design in > > > > > FLINK-276 > > > > > > > first, and then on the basic of FLIP-276 we can create other > > FLIPs > > > > with > > > > > > > detailed design according to modules and drive them. Finally, > we > > > can > > > > > > > support data consistency based on Timestamp in Flink. > > > > > > > > > > > > > > I have updated FLIP-276, deleted the Checkpoint section, and > > added > > > > the > > > > > > > overall design of `Timestamp Barrier`. Here I briefly describe > > the > > > > > > modules > > > > > > > of `Timestamp Barrier` as follows > > > > > > > 1. Generation: JobManager must coordinate all source subtasks > and > > > > > > generate > > > > > > > a unified timestamp barrier from System Time or Event Time for > > them > > > > > > > 2. Checkpoint: Store <checkpoint, timestamp barrier> when the > > > > timestamp > > > > > > > barrier is generated, so that the job can recover the same > > > timestamp > > > > > > > barrier for the uncompleted checkpoint. > > > > > > > 3. Replay data: Store <timestamp barrier, offset> for source > when > > > it > > > > > > > broadcasts timestamp barrier, so that the source can replay the > > > same > > > > > data > > > > > > > according to the same timestamp barrier. > > > > > > > 4. Align data: Align data for stateful operator(aggregation, > join > > > and > > > > > > etc.) > > > > > > > and temporal operator(window) > > > > > > > 5. Computation: Operator computation for a specific timestamp > > > barrier > > > > > > based > > > > > > > on the results of a previous timestamp barrier. > > > > > > > 6. Output: Operator outputs or commits results when it collects > > all > > > > the > > > > > > > timestamp barriers, including operators with data buffer or > async > > > > > > > operations. > > > > > > > > > > > > > > I also list the main work in Flink and Table Store in FLIP-276. > > > > Please > > > > > > help > > > > > > > to review the FLIP when you're free and feel free to give any > > > > comments. > > > > > > > > > > > > > > Looking forward for your feedback, THX > > > > > > > > > > > > > > [1] > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-276%3A+Data+Consistency+of+Streaming+and+Batch+ETL+in+Flink+and+Table+Store > > > > > > > > > > > > > > Best, > > > > > > > Shammon > > > > > > > > > > > > > > > > > > > > > On Tue, Dec 20, 2022 at 10:01 AM Shammon FY <zjur...@gmail.com > > > > > > wrote: > > > > > > > > > > > > > > > Hi Piotr, > > > > > > > > > > > > > > > > Thanks for your syncing. I will update the FLIP later and > keep > > > this > > > > > > > > discussion open. Looking forward to your feedback, thanks > > > > > > > > > > > > > > > > > > > > > > > > Best, > > > > > > > > Shammon > > > > > > > > > > > > > > > > > > > > > > > > On Mon, Dec 19, 2022 at 10:45 PM Piotr Nowojski < > > > > > pnowoj...@apache.org> > > > > > > > > wrote: > > > > > > > > > > > > > > > >> Hi Shammon, > > > > > > > >> > > > > > > > >> I've tried to sync with Timo, David Moravek and Dawid > > Wysakowicz > > > > > about > > > > > > > >> this > > > > > > > >> subject. We have only briefly chatted and exchanged some > > > > > > thoughts/ideas, > > > > > > > >> but unfortunately we were not able to finish the discussions > > > > before > > > > > > the > > > > > > > >> holiday season/vacations. Can we get back to this topic in > > > > January? > > > > > > > >> > > > > > > > >> Best, > > > > > > > >> Piotrek > > > > > > > >> > > > > > > > >> pt., 16 gru 2022 o 10:53 Shammon FY <zjur...@gmail.com> > > > > napisał(a): > > > > > > > >> > > > > > > > >> > Hi Piotr, > > > > > > > >> > > > > > > > > >> > I found there may be several points in our discussion, it > > will > > > > > cause > > > > > > > >> > misunderstanding between us when we focus on different > one. > > I > > > > list > > > > > > each > > > > > > > >> > point in our discussion as follows > > > > > > > >> > > > > > > > > >> > > Point 1: Is "Aligned Checkpoint" the only mechanism to > > > > guarantee > > > > > > data > > > > > > > >> > consistency in the current Flink implementation, and > > > "Watermark" > > > > > and > > > > > > > >> > "Aligned Checkpoint cannot do that? > > > > > > > >> > My answer is "Yes", the "Aligned Checkpoint" is the only > one > > > due > > > > > to > > > > > > its > > > > > > > >> > "Align Data" ability, we can do it in the first stage. > > > > > > > >> > > > > > > > > >> > > Point2: Can the combination of "Checkpoint Barrier" and > > > > > > "Watermark" > > > > > > > >> > support the complete consistency semantics based on > > > "Timestamp" > > > > in > > > > > > the > > > > > > > >> > current Flink implementation? > > > > > > > >> > My answer is "No", we need a new "Timestamp Barrier" > > mechanism > > > > to > > > > > do > > > > > > > >> that > > > > > > > >> > which may be upgraded from current "Watermark" or a new > > > > mechanism, > > > > > > we > > > > > > > >> can > > > > > > > >> > do it in the next second or third stage. > > > > > > > >> > > > > > > > > >> > > Point3: Are the "Checkpoint" and the new "Timestamp > > Barrier" > > > > > > > >> completely > > > > > > > >> > independent? The "Checkpoint" whatever "Aligned" or > > > "Unaligned" > > > > or > > > > > > "Task > > > > > > > >> > Local" supports the "Exactly-Once" between ETLs, and the > > > > > "Timestamp > > > > > > > >> > Barrier" mechanism guarantees data consistency between > > tables > > > > > > according > > > > > > > >> to > > > > > > > >> > timestamp for queries. > > > > > > > >> > My answer is "Yes", I totally agree with you. Let > > "Checkpoint" > > > > be > > > > > > > >> > responsible for fault tolerance and "Timestamp Barrier" > for > > > > > > consistency > > > > > > > >> > independently. > > > > > > > >> > > > > > > > > >> > @Piotr, What do you think? If I am missing or > > misunderstanding > > > > > > anything, > > > > > > > >> > please correct me, thanks > > > > > > > >> > > > > > > > > >> > Best, > > > > > > > >> > Shammon > > > > > > > >> > > > > > > > > >> > On Fri, Dec 16, 2022 at 4:17 PM Piotr Nowojski < > > > > > > pnowoj...@apache.org> > > > > > > > >> > wrote: > > > > > > > >> > > > > > > > > >> > > Hi Shammon, > > > > > > > >> > > > > > > > > > >> > > > I don't think we can combine watermarks and checkpoint > > > > > barriers > > > > > > > >> > together > > > > > > > >> > > to > > > > > > > >> > > > guarantee data consistency. There will be a "Timestamp > > > > > Barrier" > > > > > > in > > > > > > > >> our > > > > > > > >> > > > system to "commit data", "single etl failover", "low > > > latency > > > > > > between > > > > > > > >> > > ETLs" > > > > > > > >> > > > and "strong data consistency with completed semantics" > > in > > > > the > > > > > > end. > > > > > > > >> > > > > > > > > > >> > > Why do you think so? I've described to you above an > > > > alternative > > > > > > where > > > > > > > >> we > > > > > > > >> > > could be using watermarks for data consistency, > regardless > > > of > > > > > what > > > > > > > >> > > checkpointing/fault tolerance mechanism Flink would be > > > using. > > > > > Can > > > > > > you > > > > > > > >> > > explain what's wrong with that approach? Let me rephrase > > it: > > > > > > > >> > > > > > > > > > >> > > 1. There is an independent mechanism that provides > > > > exactly-once > > > > > > > >> > guarantees, > > > > > > > >> > > committing records/watermarks/events and taking care of > > the > > > > > > failover. > > > > > > > >> It > > > > > > > >> > > might be aligned, unaligned or task local checkpointing > - > > > this > > > > > > doesn't > > > > > > > >> > > matter. Let's just assume we have such a mechanism. > > > > > > > >> > > 2. There is a watermarking mechanism (it can be some > kind > > of > > > > > > system > > > > > > > >> > > versioning re-using watermarks code path if a user > didn't > > > > > > configure > > > > > > > >> > > watermarks), that takes care of the data consistency. > > > > > > > >> > > > > > > > > > >> > > Because watermarks from 2. are also subject to the > > > > exactly-once > > > > > > > >> > guarantees > > > > > > > >> > > from the 1., once they are committed downstream systems > > > (Flink > > > > > > jobs or > > > > > > > >> > > other 3rd party systems) could just easily work with the > > > > > committed > > > > > > > >> > > watermarks to provide consistent view/snapshot of the > > > tables. > > > > > Any > > > > > > > >> > > downstream system could always check what are the > > committed > > > > > > > >> watermarks, > > > > > > > >> > > select the watermark value (for example min across all > > used > > > > > > tables), > > > > > > > >> and > > > > > > > >> > > ask every table: please give me all of the data up until > > the > > > > > > selected > > > > > > > >> > > watermark. Or give me all tables in the version for the > > > > selected > > > > > > > >> > watermark. > > > > > > > >> > > > > > > > > > >> > > Am I missing something? To me it seems like this way we > > can > > > > > fully > > > > > > > >> > decouple > > > > > > > >> > > the fault tolerance mechanism from the subject of the > data > > > > > > > >> consistency. > > > > > > > >> > > > > > > > > > >> > > Best, > > > > > > > >> > > Piotrek > > > > > > > >> > > > > > > > > > >> > > czw., 15 gru 2022 o 13:01 Shammon FY <zjur...@gmail.com > > > > > > > > napisał(a): > > > > > > > >> > > > > > > > > > >> > > > Hi Piotr, > > > > > > > >> > > > > > > > > > > >> > > > It's kind of amazing about the image, it's a simple > > > example > > > > > and > > > > > > I > > > > > > > >> have > > > > > > > >> > to > > > > > > > >> > > > put it in a document > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > https://bytedance.feishu.cn/docx/FC6zdq0eqoWxHXxli71cOxe9nEe?from=from_copylink > > > > > > > >> > > > :) > > > > > > > >> > > > > > > > > > > >> > > > > Does it have to be combining watermarks and > checkpoint > > > > > > barriers > > > > > > > >> > > together? > > > > > > > >> > > > > > > > > > > >> > > > It's an interesting question. As we discussed above, > > what > > > we > > > > > > need > > > > > > > >> from > > > > > > > >> > > > "Checkpoint" is the "Align Data Ability", and from > > > > "Watermark" > > > > > > is > > > > > > > >> the > > > > > > > >> > > > "Consistency Semantics", > > > > > > > >> > > > > > > > > > > >> > > > 1) Only "Align Data" can reach data consistency when > > > > > performing > > > > > > > >> queries > > > > > > > >> > > on > > > > > > > >> > > > upstream and downstream tables. I gave an example of > > > "Global > > > > > > Count > > > > > > > >> > > Tables" > > > > > > > >> > > > in our previous discussion. We need a "Align Event" in > > the > > > > > > streaming > > > > > > > >> > > > processing, it's the most basic. > > > > > > > >> > > > > > > > > > > >> > > > 2) Only "Timestamp" can provide complete consistency > > > > > semantics. > > > > > > You > > > > > > > >> > gave > > > > > > > >> > > > some good examples about "Window" and ect operators. > > > > > > > >> > > > > > > > > > > >> > > > I don't think we can combine watermarks and checkpoint > > > > > barriers > > > > > > > >> > together > > > > > > > >> > > to > > > > > > > >> > > > guarantee data consistency. There will be a "Timestamp > > > > > Barrier" > > > > > > in > > > > > > > >> our > > > > > > > >> > > > system to "commit data", "single etl failover", "low > > > latency > > > > > > between > > > > > > > >> > > ETLs" > > > > > > > >> > > > and "strong data consistency with completed semantics" > > in > > > > the > > > > > > end. > > > > > > > >> > > > > > > > > > > >> > > > At the beginning I think we can do the simplest thing > > > first: > > > > > > > >> guarantee > > > > > > > >> > > the > > > > > > > >> > > > basic data consistency with a "Barrier Mechanism". In > > the > > > > > > current > > > > > > > >> Flink > > > > > > > >> > > > there's "Aligned Checkpoint" only, that's why we > choose > > > > > > > >> "Checkpoint" in > > > > > > > >> > > our > > > > > > > >> > > > FLIP. > > > > > > > >> > > > > > > > > > > >> > > > > I don't see an actual connection in the the > > > implementation > > > > > > steps > > > > > > > >> > > between > > > > > > > >> > > > the checkpoint barriers approach and the > watermark-like > > > > > approach > > > > > > > >> > > > > > > > > > > >> > > > As I mentioned above, we choose "Checkpoint" to > > guarantee > > > > the > > > > > > basic > > > > > > > >> > data > > > > > > > >> > > > consistency. But as we discussed, the most ideal > > solution > > > is > > > > > > > >> "Timestamp > > > > > > > >> > > > Barrier". After the first stage is completed based on > > the > > > > > > > >> "Checkpoint", > > > > > > > >> > > we > > > > > > > >> > > > need to evolve it to our ideal solution "Timestamp > > > Barrier" > > > > > > > >> > > (watermark-like > > > > > > > >> > > > approach) in the next second or third stage. This does > > not > > > > > mean > > > > > > > >> > upgrading > > > > > > > >> > > > "Checkpoint Mechanism" in Flink. It means that after > we > > > > > > implement a > > > > > > > >> new > > > > > > > >> > > > "Timestamp Barrier" or upgrade "Watermark" to support > > it, > > > we > > > > > can > > > > > > > >> use it > > > > > > > >> > > > instead of the current "Checkpoint Mechanism" directly > > in > > > > our > > > > > > > >> > > "MetaService" > > > > > > > >> > > > and "Table Store". > > > > > > > >> > > > > > > > > > > >> > > > In the discussion between @David and me, I summarized > > the > > > > work > > > > > > of > > > > > > > >> > > upgrading > > > > > > > >> > > > "Watermark" to support "Timestamp Barrier". It looks > > like > > > a > > > > > big > > > > > > job > > > > > > > >> and > > > > > > > >> > > you > > > > > > > >> > > > can find the details in our discussion. I think we > don't > > > > need > > > > > > to do > > > > > > > >> > that > > > > > > > >> > > in > > > > > > > >> > > > our first stage. > > > > > > > >> > > > > > > > > > > >> > > > Also in that discussion (my reply to @David) too, I > > > briefly > > > > > > > >> summarized > > > > > > > >> > > the > > > > > > > >> > > > work that needs to be done to use the new mechanism > > > > (Timestamp > > > > > > > >> Barrier) > > > > > > > >> > > > after we implement the basic function on "Checkpoint". > > It > > > > > seems > > > > > > that > > > > > > > >> > the > > > > > > > >> > > > work is not too big on my side, and it is feasible on > > the > > > > > whole. > > > > > > > >> > > > > > > > > > > >> > > > Based on the above points, I think we can support > basic > > > data > > > > > > > >> > consistency > > > > > > > >> > > on > > > > > > > >> > > > "Checkpoint" in the first stage which is described in > > > FLIP, > > > > > and > > > > > > > >> > continue > > > > > > > >> > > to > > > > > > > >> > > > evolve it to "Timestamp Barrier" to support low > latency > > > > > between > > > > > > ETLs > > > > > > > >> > and > > > > > > > >> > > > completed semantics in the second or third stage > later. > > > > What > > > > > > do you > > > > > > > >> > > think? > > > > > > > >> > > > > > > > > > > >> > > > Best, > > > > > > > >> > > > Shammon > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > > >> > > > On Thu, Dec 15, 2022 at 4:21 PM Piotr Nowojski < > > > > > > > >> pnowoj...@apache.org> > > > > > > > >> > > > wrote: > > > > > > > >> > > > > > > > > > > >> > > > > Hi Shammon, > > > > > > > >> > > > > > > > > > > > >> > > > > > The following is a simple example. Data is > > transferred > > > > > > between > > > > > > > >> > ETL1, > > > > > > > >> > > > ETL2 > > > > > > > >> > > > > and ETL3 in Intermediate Table by Timestamp. > > > > > > > >> > > > > > [image: simple_example.jpg] > > > > > > > >> > > > > > > > > > > > >> > > > > This time it's your image that doesn't want to load > :) > > > > > > > >> > > > > > > > > > > > >> > > > > > Timestamp Barrier > > > > > > > >> > > > > > > > > > > > >> > > > > Does it have to be combining watermarks and > checkpoint > > > > > > barriers > > > > > > > >> > > together? > > > > > > > >> > > > > Can we not achieve the same result with two > > independent > > > > > > processes > > > > > > > >> > > > > checkpointing (regardless if this is a global > > > > > > aligned/unaligned > > > > > > > >> > > > checkpoint, > > > > > > > >> > > > > or a task local checkpoint) plus watermarking? > > > > Checkpointing > > > > > > would > > > > > > > >> > > > provide > > > > > > > >> > > > > exactly-once guarantees, and actually committing the > > > > > results, > > > > > > and > > > > > > > >> it > > > > > > > >> > > > would > > > > > > > >> > > > > be actually committing the last emitted watermark? > > From > > > > the > > > > > > > >> > perspective > > > > > > > >> > > > of > > > > > > > >> > > > > the sink/table, it shouldn't really matter how the > > > > > > exactly-once is > > > > > > > >> > > > > achieved, and whether the job has performed an > > unaligned > > > > > > > >> checkpoint > > > > > > > >> > or > > > > > > > >> > > > > something completely different. It seems to me that > > the > > > > > > sink/table > > > > > > > >> > > > > could/should be able to understand/work with only > the > > > > basic > > > > > > > >> > > information: > > > > > > > >> > > > > here are records and watermarks (with at that point > of > > > > time > > > > > > > >> already > > > > > > > >> > > fixed > > > > > > > >> > > > > order), they are committed and will never change. > > > > > > > >> > > > > > > > > > > > >> > > > > > However, from the perspective of implementation > > > > > complexity, > > > > > > I > > > > > > > >> > > > personally > > > > > > > >> > > > > think using Checkpoint in the first phase makes > sense, > > > > what > > > > > > do you > > > > > > > >> > > think? > > > > > > > >> > > > > > > > > > > > >> > > > > Maybe I'm missing something, but I don't see an > actual > > > > > > connection > > > > > > > >> in > > > > > > > >> > > the > > > > > > > >> > > > > implementation steps between the checkpoint barriers > > > > > approach > > > > > > and > > > > > > > >> the > > > > > > > >> > > > > watermark-like approach. They seem to me (from the > > > > > > perspective of > > > > > > > >> > Flink > > > > > > > >> > > > > runtime at least) like two completely different > > > > mechanisms. > > > > > > Not > > > > > > > >> one > > > > > > > >> > > > leading > > > > > > > >> > > > > to the other. > > > > > > > >> > > > > > > > > > > > >> > > > > Best, > > > > > > > >> > > > > Piotrek > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > śr., 14 gru 2022 o 15:19 Shammon FY < > > zjur...@gmail.com> > > > > > > > >> napisał(a): > > > > > > > >> > > > > > > > > > > > >> > > > > > Hi Piotr, > > > > > > > >> > > > > > > > > > > > > >> > > > > > Thanks for your valuable input which makes me > > consider > > > > the > > > > > > core > > > > > > > >> > point > > > > > > > >> > > > of > > > > > > > >> > > > > > data consistency in deep. I'd like to define the > > data > > > > > > > >> consistency > > > > > > > >> > on > > > > > > > >> > > > the > > > > > > > >> > > > > > whole streaming & batch processing as follows and > I > > > hope > > > > > > that we > > > > > > > >> > can > > > > > > > >> > > > have > > > > > > > >> > > > > > an agreement on it: > > > > > > > >> > > > > > > > > > > > > >> > > > > > BOutput = Fn(BInput), BInput is a bounded input > > which > > > is > > > > > > > >> splitted > > > > > > > >> > > from > > > > > > > >> > > > > > unbounded streaming, Fn is the computation of a > node > > > or > > > > > ETL, > > > > > > > >> > BOutput > > > > > > > >> > > is > > > > > > > >> > > > > the > > > > > > > >> > > > > > bounded output of BInput. All the data in BInput > and > > > > > > BOutput are > > > > > > > >> > > > > unordered, > > > > > > > >> > > > > > and BInput and BOutput are data consistent. > > > > > > > >> > > > > > > > > > > > > >> > > > > > The key points above include 1) the segment > > semantics > > > of > > > > > > > >> BInput; 2) > > > > > > > >> > > the > > > > > > > >> > > > > > computation semantics of Fn > > > > > > > >> > > > > > > > > > > > > >> > > > > > 1. The segment semantics of BInput > > > > > > > >> > > > > > a) Transactionality of data. It is necessary to > > ensure > > > > the > > > > > > > >> semantic > > > > > > > >> > > > > > transaction of the bounded data set when it is > > > splitted > > > > > > from the > > > > > > > >> > > > > unbounded > > > > > > > >> > > > > > streaming. For example, we cannot split multiple > > > records > > > > > in > > > > > > one > > > > > > > >> > > > > transaction > > > > > > > >> > > > > > to different bounded data sets. > > > > > > > >> > > > > > b) Timeliness of data. Some data is related with > > time, > > > > > such > > > > > > as > > > > > > > >> > > boundary > > > > > > > >> > > > > > data for a window. It is necessary to consider > > whether > > > > the > > > > > > > >> bounded > > > > > > > >> > > data > > > > > > > >> > > > > set > > > > > > > >> > > > > > needs to include a watermark which can trigger the > > > > window > > > > > > > >> result. > > > > > > > >> > > > > > c) Constraints of data. The Timestamp Barrier > should > > > > > perform > > > > > > > >> some > > > > > > > >> > > > > specific > > > > > > > >> > > > > > operations after computation in operators, for > > > example, > > > > > > force > > > > > > > >> flush > > > > > > > >> > > > data. > > > > > > > >> > > > > > > > > > > > > >> > > > > > Checkpoint Barrier misses all the semantics above, > > and > > > > we > > > > > > should > > > > > > > >> > > > support > > > > > > > >> > > > > > user to define Timestamp for data on Event Time or > > > > System > > > > > > Time > > > > > > > >> > > > according > > > > > > > >> > > > > to > > > > > > > >> > > > > > the job and computation later. > > > > > > > >> > > > > > > > > > > > > >> > > > > > 2. The computation semantics of Fn > > > > > > > >> > > > > > a) Deterministic computation > > > > > > > >> > > > > > Most computations are deterministic such as map, > > > filter, > > > > > > count, > > > > > > > >> sum > > > > > > > >> > > and > > > > > > > >> > > > > > ect. They generate the same unordered result from > > the > > > > same > > > > > > > >> > unordered > > > > > > > >> > > > > input > > > > > > > >> > > > > > every time, and we can easily define data > > consistency > > > on > > > > > the > > > > > > > >> input > > > > > > > >> > > and > > > > > > > >> > > > > > output for them. > > > > > > > >> > > > > > > > > > > > > >> > > > > > b) Non-deterministic computation > > > > > > > >> > > > > > Some computations are non-deterministic. They will > > > > produce > > > > > > > >> > different > > > > > > > >> > > > > > results from the same input every time. I try to > > > divide > > > > > them > > > > > > > >> into > > > > > > > >> > the > > > > > > > >> > > > > > following types: > > > > > > > >> > > > > > 1) Non-deterministic computation semantics, such > as > > > rank > > > > > > > >> operator. > > > > > > > >> > > When > > > > > > > >> > > > > it > > > > > > > >> > > > > > computes multiple times (for example, failover), > the > > > > first > > > > > > or > > > > > > > >> last > > > > > > > >> > > > output > > > > > > > >> > > > > > results can both be the final result which will > > cause > > > > > > different > > > > > > > >> > > > failover > > > > > > > >> > > > > > handlers for downstream jobs. I will expand it > > later. > > > > > > > >> > > > > > 2) Non-deterministic computation optimization, > such > > as > > > > > async > > > > > > > >> io. It > > > > > > > >> > > is > > > > > > > >> > > > > > necessary to sync these operations when the > barrier > > of > > > > > input > > > > > > > >> > arrives. > > > > > > > >> > > > > > 3) Deviation caused by data segmentat and > > computation > > > > > > semantics, > > > > > > > >> > such > > > > > > > >> > > > as > > > > > > > >> > > > > > Window. This requires that the users should > > customize > > > > the > > > > > > data > > > > > > > >> > > > > segmentation > > > > > > > >> > > > > > according to their needs correctly. > > > > > > > >> > > > > > > > > > > > > >> > > > > > Checkpoint Barrier matches a) and Timestamp > Barrier > > > can > > > > > > match > > > > > > > >> all > > > > > > > >> > a) > > > > > > > >> > > > and > > > > > > > >> > > > > > b). > > > > > > > >> > > > > > > > > > > > > >> > > > > > We define data consistency of BInput and BOutput > > based > > > > all > > > > > > > >> above. > > > > > > > >> > The > > > > > > > >> > > > > > BOutput of upstream ETL will be the BInput of the > > next > > > > > ETL, > > > > > > and > > > > > > > >> > > > multiple > > > > > > > >> > > > > > ETL jobs form a complex "ETL Topology". > > > > > > > >> > > > > > > > > > > > > >> > > > > > Based on the above definitions, I'd like to give a > > > > general > > > > > > > >> proposal > > > > > > > >> > > > with > > > > > > > >> > > > > > "Timetamp Barrier" in my mind, it's not very > > detailed > > > > and > > > > > > please > > > > > > > >> > help > > > > > > > >> > > > to > > > > > > > >> > > > > > review it and feel free to comment @David, @Piotr > > > > > > > >> > > > > > > > > > > > > >> > > > > > 1. Data segment with Timestamp > > > > > > > >> > > > > > a) Users can define the Timestamp Barrier with > > System > > > > > Time, > > > > > > > >> Event > > > > > > > >> > > Time. > > > > > > > >> > > > > > b) Source nodes generate the same Timestamp > Barrier > > > > after > > > > > > > >> reading > > > > > > > >> > > data > > > > > > > >> > > > > > from RootTable > > > > > > > >> > > > > > c) There is a same Timetamp data in each record > > > > according > > > > > to > > > > > > > >> > > Timestamp > > > > > > > >> > > > > > Barrier, such as (a, T), (b, T), (c, T), (T, > > barrier) > > > > > > > >> > > > > > > > > > > > > >> > > > > > 2. Computation with Timestamp > > > > > > > >> > > > > > a) Records are unordered with the same Timestamp. > > > > > Stateless > > > > > > > >> > operators > > > > > > > >> > > > > such > > > > > > > >> > > > > > as map/flatmap/filter can process data without > > > aligning > > > > > > > >> Timestamp > > > > > > > >> > > > > Barrier, > > > > > > > >> > > > > > which is different from Checkpoint Barrier. > > > > > > > >> > > > > > b) Records between Timestamp are ordered. Stateful > > > > > operators > > > > > > > >> must > > > > > > > >> > > align > > > > > > > >> > > > > > data and compute by each Timestamp, then compute > by > > > > > Timetamp > > > > > > > >> > > sequence. > > > > > > > >> > > > > > c) Stateful operators will output results of > > specific > > > > > > Timestamp > > > > > > > >> > after > > > > > > > >> > > > > > computation. > > > > > > > >> > > > > > d) Sink operator "commit records" with specific > > > > Timestamp > > > > > > and > > > > > > > >> > report > > > > > > > >> > > > the > > > > > > > >> > > > > > status to JobManager > > > > > > > >> > > > > > > > > > > > > >> > > > > > 3. Read data with Timestamp > > > > > > > >> > > > > > a) Downstream ETL reads data according to > Timestamp > > > > after > > > > > > > >> upstream > > > > > > > >> > > ETL > > > > > > > >> > > > > > "commit" it. > > > > > > > >> > > > > > b) Stateful operators interact with state when > > > computing > > > > > > data of > > > > > > > >> > > > > > Timestamp, but they won't trigger checkpoint for > > every > > > > > > > >> Timestamp. > > > > > > > >> > > > > Therefore > > > > > > > >> > > > > > source ETL job can generate Timestamp every few > > > seconds > > > > or > > > > > > even > > > > > > > >> > > > hundreds > > > > > > > >> > > > > of > > > > > > > >> > > > > > milliseconds > > > > > > > >> > > > > > c) Based on Timestamp the delay between ETL jobs > > will > > > be > > > > > > very > > > > > > > >> > small, > > > > > > > >> > > > and > > > > > > > >> > > > > > in the best case the E2E latency maybe only tens > of > > > > > seconds. > > > > > > > >> > > > > > > > > > > > > >> > > > > > 4. Failover and Recovery > > > > > > > >> > > > > > ETL jobs are cascaded through the Intermediate > > Table. > > > > > After > > > > > > a > > > > > > > >> > single > > > > > > > >> > > > ETL > > > > > > > >> > > > > > job fails, it needs to replay the input data and > > > > recompute > > > > > > the > > > > > > > >> > > results. > > > > > > > >> > > > > As > > > > > > > >> > > > > > you mentioned, whether the cascaded ETL jobs are > > > > restarted > > > > > > > >> depends > > > > > > > >> > on > > > > > > > >> > > > the > > > > > > > >> > > > > > determinacy of the intermediate data between them. > > > > > > > >> > > > > > a) An ETL job will rollback and reread data from > > > > upstream > > > > > > ETL by > > > > > > > >> > > > specific > > > > > > > >> > > > > > Timestamp according to the Checkpoint. > > > > > > > >> > > > > > b) According to the management of Checkpoint and > > > > > Timestamp, > > > > > > ETL > > > > > > > >> can > > > > > > > >> > > > > replay > > > > > > > >> > > > > > all Timestamp and data after failover, which means > > > > BInput > > > > > > is the > > > > > > > >> > same > > > > > > > >> > > > > > before and after failover. > > > > > > > >> > > > > > > > > > > > > >> > > > > > c) For deterministic Fn, it generates the same > > BOutput > > > > > from > > > > > > the > > > > > > > >> > same > > > > > > > >> > > > > BInput > > > > > > > >> > > > > > 1) If there's no data of the specific Timestamp in > > the > > > > > sink > > > > > > > >> table, > > > > > > > >> > > ETL > > > > > > > >> > > > > > just "commit" it as normal. > > > > > > > >> > > > > > 2) If the Timestamp data exists in the sink table, > > ETL > > > > can > > > > > > just > > > > > > > >> > > discard > > > > > > > >> > > > > > the new data. > > > > > > > >> > > > > > > > > > > > > >> > > > > > d) For non-deterministic Fn, it generates > different > > > > > BOutput > > > > > > from > > > > > > > >> > the > > > > > > > >> > > > same > > > > > > > >> > > > > > BInput before and after failover. For example, > > > BOutput1 > > > > > > before > > > > > > > >> > > failover > > > > > > > >> > > > > and > > > > > > > >> > > > > > BOutput2 after failover. The state in ETL is > > > consistent > > > > > with > > > > > > > >> > > BOutput2. > > > > > > > >> > > > > > There are two cases according to users' > requirements > > > > > > > >> > > > > > 1) Users can accept BOutput1 as the final output > and > > > > > > downstream > > > > > > > >> > ETLs > > > > > > > >> > > > > don't > > > > > > > >> > > > > > need to restart. Sink in ETL can discard BOutput2 > > > > directly > > > > > > if > > > > > > > >> the > > > > > > > >> > > > > Timestamp > > > > > > > >> > > > > > exists in the sink table. > > > > > > > >> > > > > > 2) Users only accept BOutput2 as the final output, > > > then > > > > > all > > > > > > the > > > > > > > >> > > > > downstream > > > > > > > >> > > > > > ETLs and Intermediate Table should rollback to > > > specific > > > > > > > >> Timestamp, > > > > > > > >> > > the > > > > > > > >> > > > > > downstream ETLs should be restarted too. > > > > > > > >> > > > > > > > > > > > > >> > > > > > The following is a simple example. Data is > > transferred > > > > > > between > > > > > > > >> > ETL1, > > > > > > > >> > > > ETL2 > > > > > > > >> > > > > > and ETL3 in Intermediate Table by Timestamp. > > > > > > > >> > > > > > [image: simple_example.jpg] > > > > > > > >> > > > > > > > > > > > > >> > > > > > Besides Timestamp, there's a big challenge in > > > > Intermediate > > > > > > > >> Table. > > > > > > > >> > It > > > > > > > >> > > > > > should support a highly implemented "commit > > Timestamp > > > > > > snapshot" > > > > > > > >> > with > > > > > > > >> > > > high > > > > > > > >> > > > > > throughput, which requires the Table Store to > > enhance > > > > > > streaming > > > > > > > >> > > > > > capabilities like pulsar or kafka. > > > > > > > >> > > > > > > > > > > > > >> > > > > > In this FLIP, we plan to implement the proposal > with > > > > > > Checkpoint, > > > > > > > >> > the > > > > > > > >> > > > > above > > > > > > > >> > > > > > Timestamp can be replaced by Checkpoint. Of > course, > > > > > > Checkpoint > > > > > > > >> has > > > > > > > >> > > some > > > > > > > >> > > > > > problems. I think we have reached some consensus > in > > > the > > > > > > > >> discussion > > > > > > > >> > > > about > > > > > > > >> > > > > > the Checkpoint problems, including data segment > > > > semantics, > > > > > > flush > > > > > > > >> > data > > > > > > > >> > > > of > > > > > > > >> > > > > > some operators, and the increase of E2E delay. > > > However, > > > > > > from the > > > > > > > >> > > > > > perspective of implementation complexity, I > > personally > > > > > think > > > > > > > >> using > > > > > > > >> > > > > > Checkpoint in the first phase makes sense, what do > > you > > > > > > think? > > > > > > > >> > > > > > > > > > > > > >> > > > > > Finally, I think I misunderstood the "Rolling > > > > Checkpoint" > > > > > > and > > > > > > > >> "All > > > > > > > >> > at > > > > > > > >> > > > > once > > > > > > > >> > > > > > Checkpoint" in my last explanation which you and > > > @David > > > > > > > >> mentioned. > > > > > > > >> > I > > > > > > > >> > > > > > thought their differences were mainly to select > > > > different > > > > > > table > > > > > > > >> > > > versions > > > > > > > >> > > > > > for queries. According to your reply, I think it > is > > > > > whether > > > > > > > >> there > > > > > > > >> > are > > > > > > > >> > > > > > multiple "rolling checkpoints" in each ETL job, > > right? > > > > If > > > > > I > > > > > > > >> > > understand > > > > > > > >> > > > > > correctly, the "Rolling Checkpoint" is a good > idea, > > > and > > > > we > > > > > > can > > > > > > > >> > > > guarantee > > > > > > > >> > > > > > "Strong Data Consistency" between multiple tables > in > > > > > > MetaService > > > > > > > >> > for > > > > > > > >> > > > > > queries. Thanks. > > > > > > > >> > > > > > > > > > > > > >> > > > > > Best, > > > > > > > >> > > > > > Shammon > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > On Tue, Dec 13, 2022 at 9:36 PM Piotr Nowojski < > > > > > > > >> > pnowoj...@apache.org > > > > > > > >> > > > > > > > > > > >> > > > > > wrote: > > > > > > > >> > > > > > > > > > > > > >> > > > > >> Hi Shammon, > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> Thanks for the explanations, I think I understand > > the > > > > > > problem > > > > > > > >> > better > > > > > > > >> > > > > now. > > > > > > > >> > > > > >> I have a couple of follow up questions, but > first: > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> >> 3. I'm pretty sure there are counter examples, > > > where > > > > > > your > > > > > > > >> > > proposed > > > > > > > >> > > > > >> mechanism of using checkpoints (even aligned!) > will > > > > > produce > > > > > > > >> > > > > >> inconsistent data from the perspective of the > event > > > > time. > > > > > > > >> > > > > >> >> a) For example what if one of your "ETL" > jobs, > > > has > > > > > the > > > > > > > >> > following > > > > > > > >> > > > > DAG: > > > > > > > >> > > > > >> >> > > > > > > > >> > > > > >> >> Even if you use aligned checkpoints for > > > committing > > > > > the > > > > > > > >> data to > > > > > > > >> > > the > > > > > > > >> > > > > >> sink table, the watermarks of "Window1" and > > "Window2" > > > > are > > > > > > > >> > completely > > > > > > > >> > > > > >> independent. The sink table might easily have > data > > > from > > > > > the > > > > > > > >> > > > Src1/Window1 > > > > > > > >> > > > > >> from the event time T1 and Src2/Window2 from > later > > > > event > > > > > > time > > > > > > > >> T2. > > > > > > > >> > > > > >> >> b) I think the same applies if you have two > > > > > completely > > > > > > > >> > > > > >> independent ETL jobs writing either to the same > > sink > > > > > > table, or > > > > > > > >> two > > > > > > > >> > > to > > > > > > > >> > > > > >> different sink tables (that are both later used > in > > > the > > > > > same > > > > > > > >> > > downstream > > > > > > > >> > > > > job). > > > > > > > >> > > > > >> > > > > > > > > >> > > > > >> > Thank you for your feedback. I cannot see the > DAG > > > in > > > > > 3.a > > > > > > in > > > > > > > >> your > > > > > > > >> > > > > reply, > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> I've attached the image directly. I hope you can > > see > > > it > > > > > > now. > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> Basically what I meant is that if you have a > > topology > > > > > like > > > > > > > >> (from > > > > > > > >> > the > > > > > > > >> > > > > >> attached image): > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> window1 = src1.keyBy(...).window(...) > > > > > > > >> > > > > >> window2 = src2.keyBy(...).window(...) > > > > > > > >> > > > > >> window1.join(window2, ...).addSink(sink) > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> or with even simpler (note no keyBy between `src` > > and > > > > > > > >> `process`): > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> > > > > > src.process(some_function_that_buffers_data)..addSink(sink) > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> you will have the same problem. Generally > speaking > > if > > > > > > there is > > > > > > > >> an > > > > > > > >> > > > > >> operator buffering some data, and if the data are > > not > > > > > > flushed > > > > > > > >> on > > > > > > > >> > > every > > > > > > > >> > > > > >> checkpoint (any windowed or temporal operator, > > > > > > > >> AsyncWaitOperator, > > > > > > > >> > > CEP, > > > > > > > >> > > > > >> ...), you can design a graph that will produce > > > > > > "inconsistent" > > > > > > > >> data > > > > > > > >> > > as > > > > > > > >> > > > > part > > > > > > > >> > > > > >> of a checkpoint. > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> Apart from that a couple of other > questions/issues. > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> > 1) Global Checkpoint Commit: a) "rolling > fashion" > > > or > > > > b) > > > > > > > >> > altogether > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> Do we need to support the "altogether" one? > Rolling > > > > > > > >> checkpoint, as > > > > > > > >> > > > it's > > > > > > > >> > > > > >> more independent, I could see it scale much > better, > > > and > > > > > > avoid a > > > > > > > >> > lot > > > > > > > >> > > of > > > > > > > >> > > > > >> problems that I mentioned before. > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> > 1) Checkpoint VS Watermark > > > > > > > >> > > > > >> > > > > > > > > >> > > > > >> > 1. Stateful Computation is aligned according to > > > > > Timestamp > > > > > > > >> > Barrier > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> Indeed the biggest obstacle I see here, is that > we > > > > would > > > > > > indeed > > > > > > > >> > most > > > > > > > >> > > > > >> likely have: > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> > b) Similar to the window operator, align data > in > > > > memory > > > > > > > >> > according > > > > > > > >> > > to > > > > > > > >> > > > > >> Timestamp. > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> for every operator. > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> > 4. Failover supports Timestamp fine-grained > data > > > > > recovery > > > > > > > >> > > > > >> > > > > > > > > >> > > > > >> > As we mentioned in the FLIP, each ETL is a > > complex > > > > > single > > > > > > > >> node. > > > > > > > >> > A > > > > > > > >> > > > > single > > > > > > > >> > > > > >> > ETL job failover should not cause the failure > of > > > the > > > > > > entire > > > > > > > >> "ETL > > > > > > > >> > > > > >> Topology". > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> I don't understand this point. Regardless if we > are > > > > using > > > > > > > >> > > > > >> rolling checkpoints, all at once checkpoints or > > > > > > watermarks, I > > > > > > > >> see > > > > > > > >> > > the > > > > > > > >> > > > > same > > > > > > > >> > > > > >> problems with non determinism, if we want to > > preserve > > > > the > > > > > > > >> > > requirement > > > > > > > >> > > > to > > > > > > > >> > > > > >> not fail over the whole topology at once. > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> Both Watermarks and "rolling checkpoint" I think > > have > > > > the > > > > > > same > > > > > > > >> > > issue, > > > > > > > >> > > > > >> that either require deterministic logic, or > global > > > > > > failover, or > > > > > > > >> > > > > downstream > > > > > > > >> > > > > >> jobs can only work on the already committed by > the > > > > > upstream > > > > > > > >> > records. > > > > > > > >> > > > But > > > > > > > >> > > > > >> working with only "committed records" would > either > > > > brake > > > > > > > >> > consistency > > > > > > > >> > > > > >> between different jobs, or would cause huge delay > > in > > > > > > > >> checkpointing > > > > > > > >> > > and > > > > > > > >> > > > > e2e > > > > > > > >> > > > > >> latency, as: > > > > > > > >> > > > > >> 1. upstream job has to produce some data, > > downstream > > > > can > > > > > > not > > > > > > > >> > process > > > > > > > >> > > > it, > > > > > > > >> > > > > >> downstream can not process this data yet > > > > > > > >> > > > > >> 2. checkpoint 42 is triggered on the upstream job > > > > > > > >> > > > > >> 3. checkpoint 42 is completed on the upstream > job, > > > data > > > > > > > >> processed > > > > > > > >> > > > since > > > > > > > >> > > > > >> last checkpoint has been committed > > > > > > > >> > > > > >> 4. upstream job can continue producing more data > > > > > > > >> > > > > >> 5. only now downstream can start processing the > > data > > > > > > produced > > > > > > > >> in > > > > > > > >> > 1., > > > > > > > >> > > > but > > > > > > > >> > > > > >> it can not read the not-yet-committed data from > 4. > > > > > > > >> > > > > >> 6. once downstream finishes processing data from > > 1., > > > it > > > > > can > > > > > > > >> > trigger > > > > > > > >> > > > > >> checkpoint 42 > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> The "all at once checkpoint", I can see only > > working > > > > with > > > > > > > >> global > > > > > > > >> > > > > failover > > > > > > > >> > > > > >> of everything. > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> This is assuming exactly-once mode. at-least-once > > > would > > > > > be > > > > > > much > > > > > > > >> > > > easier. > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> Best, > > > > > > > >> > > > > >> Piotrek > > > > > > > >> > > > > >> > > > > > > > >> > > > > >> wt., 13 gru 2022 o 08:57 Shammon FY < > > > zjur...@gmail.com > > > > > > > > > > > > >> > napisał(a): > > > > > > > >> > > > > >> > > > > > > > >> > > > > >>> Hi David, > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> Thanks for the comments from you and @Piotr. I'd > > > like > > > > to > > > > > > > >> explain > > > > > > > >> > > the > > > > > > > >> > > > > >>> details about the FLIP first. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> 1) Global Checkpoint Commit: a) "rolling > fashion" > > or > > > > b) > > > > > > > >> > altogether > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> This mainly depends on the needs of users. Users > > can > > > > > > decide > > > > > > > >> the > > > > > > > >> > > data > > > > > > > >> > > > > >>> version of tables in their queries according to > > > > > different > > > > > > > >> > > > requirements > > > > > > > >> > > > > >>> for > > > > > > > >> > > > > >>> data consistency and freshness. Since we manage > > > > multiple > > > > > > > >> versions > > > > > > > >> > > for > > > > > > > >> > > > > >>> each > > > > > > > >> > > > > >>> table, this will not bring too much complexity > to > > > the > > > > > > system. > > > > > > > >> We > > > > > > > >> > > only > > > > > > > >> > > > > >>> need > > > > > > > >> > > > > >>> to support different strategies when calculating > > > table > > > > > > > >> versions > > > > > > > >> > for > > > > > > > >> > > > > >>> query. > > > > > > > >> > > > > >>> So we give this decision to users, who can use > > > > > > > >> "consistency.type" > > > > > > > >> > > to > > > > > > > >> > > > > set > > > > > > > >> > > > > >>> different consistency in "Catalog". We can > > continue > > > to > > > > > > refine > > > > > > > >> > this > > > > > > > >> > > > > later. > > > > > > > >> > > > > >>> For example, dynamic parameters support > different > > > > > > consistency > > > > > > > >> > > > > >>> requirements > > > > > > > >> > > > > >>> for each query > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> 2) MetaService module > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> Many Flink streaming jobs use application mode, > > and > > > > they > > > > > > are > > > > > > > >> > > > > independent > > > > > > > >> > > > > >>> of > > > > > > > >> > > > > >>> each other. So we currently assume that > > MetaService > > > is > > > > > an > > > > > > > >> > > independent > > > > > > > >> > > > > >>> node. > > > > > > > >> > > > > >>> In the first phase, it will be started in > > > standalone, > > > > > and > > > > > > HA > > > > > > > >> will > > > > > > > >> > > be > > > > > > > >> > > > > >>> supported later. This node will reuse many Flink > > > > > modules, > > > > > > > >> > including > > > > > > > >> > > > > REST, > > > > > > > >> > > > > >>> Gateway-RpcServer, etc. We hope that the core > > > > functions > > > > > of > > > > > > > >> > > > MetaService > > > > > > > >> > > > > >>> can > > > > > > > >> > > > > >>> be developed as a component. When Flink > > subsequently > > > > > uses > > > > > > a > > > > > > > >> large > > > > > > > >> > > > > session > > > > > > > >> > > > > >>> cluster to support various computations, it can > be > > > > > > integrated > > > > > > > >> > into > > > > > > > >> > > > the > > > > > > > >> > > > > >>> "ResourceManager" as a plug-in component. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> Besides above, I'd like to describe the > Checkpoint > > > and > > > > > > > >> Watermark > > > > > > > >> > > > > >>> mechanisms > > > > > > > >> > > > > >>> in detail as follows. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> 1) Checkpoint VS Watermark > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> As you mentioned, I think it's very correct that > > > what > > > > we > > > > > > want > > > > > > > >> in > > > > > > > >> > > the > > > > > > > >> > > > > >>> Checkpoint is to align streaming computation and > > > data > > > > > > > >> according > > > > > > > >> > to > > > > > > > >> > > > > >>> certain > > > > > > > >> > > > > >>> semantics. Timestamp is a very ideal solution. > To > > > > > achieve > > > > > > this > > > > > > > >> > > goal, > > > > > > > >> > > > we > > > > > > > >> > > > > >>> can > > > > > > > >> > > > > >>> think of the following functions that need to be > > > > > > supported in > > > > > > > >> the > > > > > > > >> > > > > >>> Watermark > > > > > > > >> > > > > >>> mechanism: > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> 1. Stateful Computation is aligned according to > > > > > Timestamp > > > > > > > >> Barrier > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> As the "three tables example" we discussed > above, > > we > > > > > need > > > > > > to > > > > > > > >> > align > > > > > > > >> > > > the > > > > > > > >> > > > > >>> stateful operator computation according to the > > > barrier > > > > > to > > > > > > > >> ensure > > > > > > > >> > > the > > > > > > > >> > > > > >>> consistency of the result data. In order to > align > > > the > > > > > > > >> > computation, > > > > > > > >> > > > > there > > > > > > > >> > > > > >>> are two ways in my mind > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> a) Similar to the Aligned Checkpoint Barrier. > > > > Timestamp > > > > > > > >> Barrier > > > > > > > >> > > > aligns > > > > > > > >> > > > > >>> data > > > > > > > >> > > > > >>> according to the channel, which will lead to > > > > > backpressure > > > > > > just > > > > > > > >> > like > > > > > > > >> > > > the > > > > > > > >> > > > > >>> aligned checkpoint. It seems not a good idea. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> b) Similar to the window operator, align data in > > > > memory > > > > > > > >> according > > > > > > > >> > > to > > > > > > > >> > > > > >>> Timestamp. Two steps need to be supported here: > > > first, > > > > > > data is > > > > > > > >> > > > aligned > > > > > > > >> > > > > by > > > > > > > >> > > > > >>> timestamp for state operators; secondly, > Timestamp > > > is > > > > > > strictly > > > > > > > >> > > > > >>> sequential, > > > > > > > >> > > > > >>> global aggregation operators need to perform > > > > aggregation > > > > > > in > > > > > > > >> > > timestamp > > > > > > > >> > > > > >>> order > > > > > > > >> > > > > >>> and output the final results. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> 2. Coordinate multiple source nodes to assign > > > unified > > > > > > > >> Timestamp > > > > > > > >> > > > > Barriers > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> Since the stateful operator needs to be aligned > > > > > according > > > > > > to > > > > > > > >> the > > > > > > > >> > > > > >>> Timestamp > > > > > > > >> > > > > >>> Barrier, source subtasks of multiple jobs should > > > > > generate > > > > > > the > > > > > > > >> > same > > > > > > > >> > > > > >>> Timestamp Barrier. ETL jobs consuming RootTable > > > should > > > > > > > >> interact > > > > > > > >> > > with > > > > > > > >> > > > > >>> "MetaService" to generate the same Timestamp T1, > > T2, > > > > T3 > > > > > > ... > > > > > > > >> and > > > > > > > >> > so > > > > > > > >> > > > on. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> 3. JobManager needs to manage the completed > > > Timestamp > > > > > > Barrier > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> When the Timestamp Barrier of the ETL job has > been > > > > > > completed, > > > > > > > >> it > > > > > > > >> > > > means > > > > > > > >> > > > > >>> that > > > > > > > >> > > > > >>> the data of the specified Timestamp can be > queried > > > by > > > > > > users. > > > > > > > >> > > > JobManager > > > > > > > >> > > > > >>> needs to summarize its Timestamp processing and > > > report > > > > > the > > > > > > > >> > > completed > > > > > > > >> > > > > >>> Timestamp and data snapshots to the MetaServer. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> 4. Failover supports Timestamp fine-grained data > > > > > recovery > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> As we mentioned in the FLIP, each ETL is a > complex > > > > > single > > > > > > > >> node. A > > > > > > > >> > > > > single > > > > > > > >> > > > > >>> ETL job failover should not cause the failure of > > the > > > > > > entire > > > > > > > >> "ETL > > > > > > > >> > > > > >>> Topology". > > > > > > > >> > > > > >>> This requires that the result data of Timestamp > > > > > generated > > > > > > by > > > > > > > >> > > upstream > > > > > > > >> > > > > ETL > > > > > > > >> > > > > >>> should be deterministic. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> a) The determinacy of Timestamp, that is, before > > and > > > > > > after ETL > > > > > > > >> > job > > > > > > > >> > > > > >>> failover, the same Timestamp sequence must be > > > > generated. > > > > > > Each > > > > > > > >> > > > > Checkpoint > > > > > > > >> > > > > >>> needs to record the included Timestamp list, > > > > especially > > > > > > the > > > > > > > >> > source > > > > > > > >> > > > node > > > > > > > >> > > > > >>> of > > > > > > > >> > > > > >>> the RootTable. After Failover, it needs to > > > regenerate > > > > > > > >> Timestamp > > > > > > > >> > > > > according > > > > > > > >> > > > > >>> to the Timestamp list. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> b) The determinacy of Timestamp data, that is, > the > > > > same > > > > > > > >> Timestamp > > > > > > > >> > > > needs > > > > > > > >> > > > > >>> to > > > > > > > >> > > > > >>> replay the same data before and after Failover, > > and > > > > > > generate > > > > > > > >> the > > > > > > > >> > > same > > > > > > > >> > > > > >>> results in Sink Table. Each Timestamp must save > > > start > > > > > and > > > > > > end > > > > > > > >> > > offsets > > > > > > > >> > > > > (or > > > > > > > >> > > > > >>> snapshot id) of RootTable. After failover, the > > > source > > > > > > nodes > > > > > > > >> need > > > > > > > >> > to > > > > > > > >> > > > > >>> replay > > > > > > > >> > > > > >>> the data according to the offset to ensure that > > the > > > > data > > > > > > of > > > > > > > >> each > > > > > > > >> > > > > >>> Timestamp > > > > > > > >> > > > > >>> is consistent before and after Failover. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> For the specific requirements and complexity, > > please > > > > > help > > > > > > to > > > > > > > >> > review > > > > > > > >> > > > > when > > > > > > > >> > > > > >>> you are free @David @Piotr, thanks :) > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> 2) Evolution from Checkpoint to Timestamp > > Mechanism > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> You give a very important question in your reply > > > > which I > > > > > > > >> missed > > > > > > > >> > > > before: > > > > > > > >> > > > > >>> if > > > > > > > >> > > > > >>> Aligned Checkpoint is used in the first stage, > how > > > > > > complex is > > > > > > > >> the > > > > > > > >> > > > > >>> evolution > > > > > > > >> > > > > >>> from Checkpoint to Timestamp later? I made a > > general > > > > > > > >> comparison > > > > > > > >> > > here, > > > > > > > >> > > > > >>> which > > > > > > > >> > > > > >>> may not be very detailed. There are three roles > in > > > the > > > > > > whole > > > > > > > >> > > system: > > > > > > > >> > > > > >>> MetaService, Flink ETL Job and Table Store. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> a) MetaService > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> It manages the data consistency among multiple > ETL > > > > jobs, > > > > > > > >> > including > > > > > > > >> > > > > >>> coordinating the Barrier for the Source ETL > nodes, > > > > > > setting the > > > > > > > >> > > > starting > > > > > > > >> > > > > >>> Barrier for ETL job startup, and calculating the > > > Table > > > > > > version > > > > > > > >> > for > > > > > > > >> > > > > >>> queries > > > > > > > >> > > > > >>> according to different strategies. It has little > > to > > > do > > > > > > with > > > > > > > >> > > > Checkpoint > > > > > > > >> > > > > in > > > > > > > >> > > > > >>> fact, we can pay attention to it when designing > > the > > > > API > > > > > > and > > > > > > > >> > > > > implementing > > > > > > > >> > > > > >>> the functions. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> b) Flink ETL Job > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> At present, the workload is relatively small and > > we > > > > need > > > > > > to > > > > > > > >> > trigger > > > > > > > >> > > > > >>> checkpoints in CheckpointCoordinator manually by > > > > > > > >> SplitEnumerator. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> c) Table Store > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> Table Store mainly provides the ability to write > > and > > > > > read > > > > > > > >> data. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> c.1) Write data. At present, Table Store > generates > > > > > > snapshots > > > > > > > >> > > > according > > > > > > > >> > > > > to > > > > > > > >> > > > > >>> two phases in Flink. When using Checkpoint as > > > > > consistency > > > > > > > >> > > management, > > > > > > > >> > > > > we > > > > > > > >> > > > > >>> need to write checkpoint information to > snapshots. > > > > After > > > > > > using > > > > > > > >> > > > > Timestamp > > > > > > > >> > > > > >>> Barrier, the snapshot in Table Store may be > > > > disassembled > > > > > > more > > > > > > > >> > > finely, > > > > > > > >> > > > > and > > > > > > > >> > > > > >>> we need to write Timestamp information to the > data > > > > > file. A > > > > > > > >> > > > > "checkpointed > > > > > > > >> > > > > >>> snapshot" may contain multiple "Timestamp > > > snapshots". > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> c.2) Read data. The SplitEnumerator that reads > > data > > > > from > > > > > > the > > > > > > > >> > Table > > > > > > > >> > > > > Store > > > > > > > >> > > > > >>> will manage multiple splits according to the > > version > > > > > > number. > > > > > > > >> > After > > > > > > > >> > > > the > > > > > > > >> > > > > >>> specified splits are completed, it sends a > Barrier > > > > > > command to > > > > > > > >> > > > trigger a > > > > > > > >> > > > > >>> checkpoint in the ETL job. The source node will > > > > > broadcast > > > > > > the > > > > > > > >> > > > > checkpoint > > > > > > > >> > > > > >>> barrier downstream after receiving it. When > using > > > > > > Timestamp > > > > > > > >> > > Barrier, > > > > > > > >> > > > > the > > > > > > > >> > > > > >>> overall process is similar, but the > > SplitEnumerator > > > > does > > > > > > not > > > > > > > >> need > > > > > > > >> > > to > > > > > > > >> > > > > >>> trigger a checkpoint to the Flink ETL, and the > > > Source > > > > > node > > > > > > > >> needs > > > > > > > >> > to > > > > > > > >> > > > > >>> support > > > > > > > >> > > > > >>> broadcasting Timestamp Barrier to the downstream > > at > > > > that > > > > > > time. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> From the above overall, the evolution complexity > > > from > > > > > > > >> Checkpoint > > > > > > > >> > to > > > > > > > >> > > > > >>> Timestamp seems controllable, but the specific > > > > > > implementation > > > > > > > >> > needs > > > > > > > >> > > > > >>> careful > > > > > > > >> > > > > >>> design, and the concept and features of > Checkpoint > > > > > should > > > > > > not > > > > > > > >> be > > > > > > > >> > > > > >>> introduced > > > > > > > >> > > > > >>> too much into relevant interfaces and functions. > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> What do you think of it? Looking forward to your > > > > > feedback, > > > > > > > >> thanks > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> Best, > > > > > > > >> > > > > >>> Shammon > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> On Mon, Dec 12, 2022 at 11:46 PM David Morávek < > > > > > > > >> d...@apache.org> > > > > > > > >> > > > > wrote: > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >>> > Hi Shammon, > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > I'm starting to see what you're trying to > > achieve, > > > > and > > > > > > it's > > > > > > > >> > > really > > > > > > > >> > > > > >>> > exciting. I share Piotr's concerns about e2e > > > latency > > > > > and > > > > > > > >> > > disability > > > > > > > >> > > > > to > > > > > > > >> > > > > >>> use > > > > > > > >> > > > > >>> > unaligned checkpoints. > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > I have a couple of questions that are not > clear > > to > > > > me > > > > > > from > > > > > > > >> > going > > > > > > > >> > > > over > > > > > > > >> > > > > >>> the > > > > > > > >> > > > > >>> > FLIP: > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > 1) Global Checkpoint Commit > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > Are you planning on committing the checkpoints > > in > > > > a) a > > > > > > > >> "rolling > > > > > > > >> > > > > >>> fashion" - > > > > > > > >> > > > > >>> > one pipeline after another, or b) altogether - > > > once > > > > > the > > > > > > data > > > > > > > >> > have > > > > > > > >> > > > > been > > > > > > > >> > > > > >>> > processed by all pipelines? > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > Option a) would be eventually consistent (for > > > batch > > > > > > queries, > > > > > > > >> > > you'd > > > > > > > >> > > > > >>> need to > > > > > > > >> > > > > >>> > use the last checkpoint produced by the most > > > > > downstream > > > > > > > >> table), > > > > > > > >> > > > > >>> whereas b) > > > > > > > >> > > > > >>> > would be strongly consistent at the cost of > > > > increasing > > > > > > the > > > > > > > >> e2e > > > > > > > >> > > > > latency > > > > > > > >> > > > > >>> even > > > > > > > >> > > > > >>> > more. > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > I feel that option a) is what this should be > > > headed > > > > > for. > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > 2) MetaService > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > Should this be a new general Flink component > or > > > one > > > > > > > >> specific to > > > > > > > >> > > the > > > > > > > >> > > > > >>> Flink > > > > > > > >> > > > > >>> > Table Store? > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > 3) Follow-ups > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > From the above discussion, there is a > consensus > > > > that, > > > > > > in the > > > > > > > >> > > ideal > > > > > > > >> > > > > >>> case, > > > > > > > >> > > > > >>> > watermarks would be a way to go, but there is > > some > > > > > > > >> underlying > > > > > > > >> > > > > mechanism > > > > > > > >> > > > > >>> > missing. It would be great to discuss this > > option > > > in > > > > > > more > > > > > > > >> > detail > > > > > > > >> > > to > > > > > > > >> > > > > >>> compare > > > > > > > >> > > > > >>> > the solutions in terms of implementation cost, > > > maybe > > > > > it > > > > > > > >> could > > > > > > > >> > not > > > > > > > >> > > > be > > > > > > > >> > > > > as > > > > > > > >> > > > > >>> > complex. > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > All in all, I don't feel that checkpoints are > > > > suitable > > > > > > for > > > > > > > >> > > > providing > > > > > > > >> > > > > >>> > consistent table versioning between multiple > > > > > pipelines. > > > > > > The > > > > > > > >> > main > > > > > > > >> > > > > >>> reason is > > > > > > > >> > > > > >>> > that they are designed to be a fault tolerance > > > > > > mechanism. > > > > > > > >> > > Somewhere > > > > > > > >> > > > > >>> between > > > > > > > >> > > > > >>> > the lines, you've already noted that the > > primitive > > > > > > you're > > > > > > > >> > looking > > > > > > > >> > > > for > > > > > > > >> > > > > >>> is > > > > > > > >> > > > > >>> > cross-pipeline barrier alignment, which is the > > > > > > mechanism a > > > > > > > >> > subset > > > > > > > >> > > > of > > > > > > > >> > > > > >>> > currently supported checkpointing > > implementations > > > > > > happen to > > > > > > > >> be > > > > > > > >> > > > using. > > > > > > > >> > > > > >>> Is > > > > > > > >> > > > > >>> > that correct? > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > My biggest concern is that tying this with a > > > > > > "side-effect" > > > > > > > >> of > > > > > > > >> > the > > > > > > > >> > > > > >>> > checkpointing mechanism could block us from > > > evolving > > > > > it > > > > > > > >> > further. > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > Best, > > > > > > > >> > > > > >>> > D. > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > On Mon, Dec 12, 2022 at 6:11 AM Shammon FY < > > > > > > > >> zjur...@gmail.com> > > > > > > > >> > > > > wrote: > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > > Hi Piotr, > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > Thank you for your feedback. I cannot see > the > > > DAG > > > > in > > > > > > 3.a > > > > > > > >> in > > > > > > > >> > > your > > > > > > > >> > > > > >>> reply, > > > > > > > >> > > > > >>> > but > > > > > > > >> > > > > >>> > > I'd like to answer some questions first. > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > Your understanding is very correct. We want > to > > > > align > > > > > > the > > > > > > > >> data > > > > > > > >> > > > > >>> versions of > > > > > > > >> > > > > >>> > > all intermediate tables through checkpoint > > > > mechanism > > > > > > in > > > > > > > >> > Flink. > > > > > > > >> > > > I'm > > > > > > > >> > > > > >>> sorry > > > > > > > >> > > > > >>> > > that I have omitted some default constraints > > in > > > > > FLIP, > > > > > > > >> > including > > > > > > > >> > > > > only > > > > > > > >> > > > > >>> > > supporting aligned checkpoints; one table > can > > > only > > > > > be > > > > > > > >> written > > > > > > > >> > > by > > > > > > > >> > > > > one > > > > > > > >> > > > > >>> ETL > > > > > > > >> > > > > >>> > > job. I will add these later. > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > Why can't the watermark mechanism achieve > the > > > data > > > > > > > >> > consistency > > > > > > > >> > > we > > > > > > > >> > > > > >>> wanted? > > > > > > > >> > > > > >>> > > For example, there are 3 tables, Table1 is > > word > > > > > table, > > > > > > > >> Table2 > > > > > > > >> > > is > > > > > > > >> > > > > >>> > word->cnt > > > > > > > >> > > > > >>> > > table and Table3 is cnt1->cnt2 table. > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > 1. ETL1 from Table1 to Table2: INSERT INTO > > > Table2 > > > > > > SELECT > > > > > > > >> > word, > > > > > > > >> > > > > >>> count(*) > > > > > > > >> > > > > >>> > > FROM Table1 GROUP BY word > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > 2. ETL2 from Table2 to Table3: INSERT INTO > > > Table3 > > > > > > SELECT > > > > > > > >> cnt, > > > > > > > >> > > > > >>> count(*) > > > > > > > >> > > > > >>> > FROM > > > > > > > >> > > > > >>> > > Table2 GROUP BY cnt > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > ETL1 has 2 subtasks to read multiple buckets > > > from > > > > > > Table1, > > > > > > > >> > where > > > > > > > >> > > > > >>> subtask1 > > > > > > > >> > > > > >>> > > reads streaming data as [a, b, c, a, d, a, > b, > > > c, d > > > > > > ...] > > > > > > > >> and > > > > > > > >> > > > > subtask2 > > > > > > > >> > > > > >>> > reads > > > > > > > >> > > > > >>> > > streaming data as [a, c, d, q, a, v, c, d > > ...]. > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > 1. Unbounded streaming data is divided into > > > > multiple > > > > > > sets > > > > > > > >> > > > according > > > > > > > >> > > > > >>> to > > > > > > > >> > > > > >>> > some > > > > > > > >> > > > > >>> > > semantic requirements. The most extreme may > be > > > one > > > > > > set for > > > > > > > >> > each > > > > > > > >> > > > > data. > > > > > > > >> > > > > >>> > > Assume that the sets of subtask1 and > subtask2 > > > > > > separated by > > > > > > > >> > the > > > > > > > >> > > > same > > > > > > > >> > > > > >>> > > semantics are [a, b, c, a, d] and [a, c, d, > > q], > > > > > > > >> respectively. > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > 2. After the above two sets are computed by > > > ETL1, > > > > > the > > > > > > > >> result > > > > > > > >> > > data > > > > > > > >> > > > > >>> > generated > > > > > > > >> > > > > >>> > > in Table 2 is [(a, 3), (b, 1), (c, 1), (d, > 2), > > > (q, > > > > > > 1)]. > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > 3. The result data generated in Table 3 > after > > > the > > > > > > data in > > > > > > > >> > > Table 2 > > > > > > > >> > > > > is > > > > > > > >> > > > > >>> > > computed by ETL2 is [(1, 3), (2, 1), (3, 1)] > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > We want to align the data of Table1, Table2 > > and > > > > > > Table3 and > > > > > > > >> > > manage > > > > > > > >> > > > > the > > > > > > > >> > > > > >>> > data > > > > > > > >> > > > > >>> > > versions. When users execute OLAP/Batch > > queries > > > > join > > > > > > on > > > > > > > >> these > > > > > > > >> > > > > >>> tables, the > > > > > > > >> > > > > >>> > > following consistency data can be found > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > 1. Table1: [a, b, c, a, d] and [a, c, d, q] > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > 2. Table2: [a, 3], [b, 1], [c, 1], [d, 2], > [q, > > > 1] > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > 3. Table3: [1, 3], [2, 1], [3, 1] > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > Users can perform query: SELECT t1.word, > > t2.cnt, > > > > > > t3.cnt2 > > > > > > > >> from > > > > > > > >> > > > > Table1 > > > > > > > >> > > > > >>> t1 > > > > > > > >> > > > > >>> > > JOIN Table2 t2 JOIN Table3 t3 on > > t1.word=t2.word > > > > and > > > > > > > >> > > > > t2.cnt=t3.cnt1; > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > In the view of users, the data is consistent > > on > > > a > > > > > > unified > > > > > > > >> > > > "version" > > > > > > > >> > > > > >>> > between > > > > > > > >> > > > > >>> > > Table1, Table2 and Table3. > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > In the current Flink implementation, the > > aligned > > > > > > > >> checkpoint > > > > > > > >> > can > > > > > > > >> > > > > >>> achieve > > > > > > > >> > > > > >>> > the > > > > > > > >> > > > > >>> > > above capabilities (let's ignore the > > > segmentation > > > > > > > >> semantics > > > > > > > >> > of > > > > > > > >> > > > > >>> checkpoint > > > > > > > >> > > > > >>> > > first). Because the Checkpoint Barrier will > > > align > > > > > the > > > > > > data > > > > > > > >> > when > > > > > > > >> > > > > >>> > performing > > > > > > > >> > > > > >>> > > the global Count aggregation, we can > associate > > > the > > > > > > > >> snapshot > > > > > > > >> > > with > > > > > > > >> > > > > the > > > > > > > >> > > > > >>> > > checkpoint in the Table Store, query the > > > specified > > > > > > > >> snapshot > > > > > > > >> > of > > > > > > > >> > > > > >>> > > Table1/Table2/Table3 through the checkpoint, > > and > > > > > > achieve > > > > > > > >> the > > > > > > > >> > > > > >>> consistency > > > > > > > >> > > > > >>> > > requirements of the above unified "version". > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > Current watermark mechanism in Flink cannot > > > > achieve > > > > > > the > > > > > > > >> above > > > > > > > >> > > > > >>> > consistency. > > > > > > > >> > > > > >>> > > For example, we use watermark to divide data > > > into > > > > > > multiple > > > > > > > >> > sets > > > > > > > >> > > > in > > > > > > > >> > > > > >>> > subtask1 > > > > > > > >> > > > > >>> > > and subtask2 as followed > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > 1. subtask1：[(a, T1), (b, T1), (c, T1), (a, > > T1), > > > > (d, > > > > > > T1)], > > > > > > > >> > T1, > > > > > > > >> > > > [(a, > > > > > > > >> > > > > >>> T2), > > > > > > > >> > > > > >>> > > (b, T2), (c, T2), (d, T2)], T2 > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > 2. subtask2: [(a, T1), (c, T1), (d, T1), (q, > > > T1)], > > > > > T1, > > > > > > > >> .... > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > As Flink watermark does not have barriers > and > > > > cannot > > > > > > align > > > > > > > >> > > data, > > > > > > > >> > > > > ETL1 > > > > > > > >> > > > > >>> > Count > > > > > > > >> > > > > >>> > > operator may compute the data of subtask1 > > first: > > > > > [(a, > > > > > > T1), > > > > > > > >> > (b, > > > > > > > >> > > > T1), > > > > > > > >> > > > > >>> (c, > > > > > > > >> > > > > >>> > > T1), (a, T1), (d, T1)], T1, [(a, T2), (b, > > T2)], > > > > then > > > > > > > >> compute > > > > > > > >> > > the > > > > > > > >> > > > > >>> data of > > > > > > > >> > > > > >>> > > subtask2: [(a, T1), (c, T1), (d, T1), (q, > > T1)], > > > > T1, > > > > > > which > > > > > > > >> is > > > > > > > >> > > not > > > > > > > >> > > > > >>> possible > > > > > > > >> > > > > >>> > > in aligned checkpoint. > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > In this order, the result output to Table2 > > after > > > > the > > > > > > Count > > > > > > > >> > > > > >>> aggregation > > > > > > > >> > > > > >>> > will > > > > > > > >> > > > > >>> > > be: (a, 1, T1), (b, 1, T1), (c, 1, T1), (a, > 2, > > > > T1), > > > > > > (a, 3, > > > > > > > >> > T2), > > > > > > > >> > > > (b, > > > > > > > >> > > > > >>> 2, > > > > > > > >> > > > > >>> > T2), > > > > > > > >> > > > > >>> > > (a, 4, T1), (c, 2, T1), (d, 1, T1), (q, 1, > > T1), > > > > > which > > > > > > can > > > > > > > >> be > > > > > > > >> > > > > >>> simplified > > > > > > > >> > > > > >>> > as: > > > > > > > >> > > > > >>> > > [(b, 1, T1), (a, 3, T2), (b, 2, T2), (a, 4, > > T1), > > > > (c, > > > > > > 2, > > > > > > > >> T1), > > > > > > > >> > > (d, > > > > > > > >> > > > 1, > > > > > > > >> > > > > >>> T1), > > > > > > > >> > > > > >>> > > (q, 1, T1)] > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > There's no (a, 3, T1), we have been unable > to > > > > query > > > > > > > >> > consistent > > > > > > > >> > > > data > > > > > > > >> > > > > >>> > results > > > > > > > >> > > > > >>> > > on Table1 and Table2 according to T1. Table > 3 > > > has > > > > > the > > > > > > same > > > > > > > >> > > > problem. > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > In addition to using Checkpoint Barrier, the > > > other > > > > > > > >> > > implementation > > > > > > > >> > > > > >>> > > supporting watermark above is to convert > Count > > > > > > aggregation > > > > > > > >> > into > > > > > > > >> > > > > >>> Window > > > > > > > >> > > > > >>> > > Count. After the global Count is converted > > into > > > > > window > > > > > > > >> > > operator, > > > > > > > >> > > > it > > > > > > > >> > > > > >>> needs > > > > > > > >> > > > > >>> > > to support cross window data computation. > > > Similar > > > > to > > > > > > the > > > > > > > >> data > > > > > > > >> > > > > >>> > relationship > > > > > > > >> > > > > >>> > > between the previous and the current > > Checkpoint, > > > > it > > > > > is > > > > > > > >> > > equivalent > > > > > > > >> > > > > to > > > > > > > >> > > > > >>> > > introducing the Watermark Barrier, which > > > requires > > > > > > > >> adjustments > > > > > > > >> > > to > > > > > > > >> > > > > the > > > > > > > >> > > > > >>> > > current Flink Watermark mechanism. > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > Besides the above global aggregation, there > > are > > > > > window > > > > > > > >> > > operators > > > > > > > >> > > > in > > > > > > > >> > > > > >>> > Flink. > > > > > > > >> > > > > >>> > > I don't know if my understanding is > correct(I > > > > cannot > > > > > > see > > > > > > > >> the > > > > > > > >> > > DAG > > > > > > > >> > > > in > > > > > > > >> > > > > >>> your > > > > > > > >> > > > > >>> > > example), please correct me if it's wrong. I > > > think > > > > > you > > > > > > > >> raise > > > > > > > >> > a > > > > > > > >> > > > very > > > > > > > >> > > > > >>> > > important and interesting question: how to > > > define > > > > > data > > > > > > > >> > > > consistency > > > > > > > >> > > > > in > > > > > > > >> > > > > >>> > > different window computations which will > > > generate > > > > > > > >> different > > > > > > > >> > > > > >>> timestamps of > > > > > > > >> > > > > >>> > > the same data. This situation also occurs > when > > > > using > > > > > > event > > > > > > > >> > time > > > > > > > >> > > > to > > > > > > > >> > > > > >>> align > > > > > > > >> > > > > >>> > > data. At present, what I can think of is to > > > store > > > > > > these > > > > > > > >> > > > information > > > > > > > >> > > > > >>> in > > > > > > > >> > > > > >>> > > Table Store, users can perform filter or > join > > on > > > > > data > > > > > > with > > > > > > > >> > > them. > > > > > > > >> > > > > This > > > > > > > >> > > > > >>> > FLIP > > > > > > > >> > > > > >>> > > is our first phase, and the specific > > > > implementation > > > > > of > > > > > > > >> this > > > > > > > >> > > will > > > > > > > >> > > > be > > > > > > > >> > > > > >>> > > designed and considered in the next phase > and > > > > FLIP. > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > Although the Checkpoint Barrier can achieve > > the > > > > most > > > > > > basic > > > > > > > >> > > > > >>> consistency, > > > > > > > >> > > > > >>> > as > > > > > > > >> > > > > >>> > > you mentioned, using the Checkpoint > mechanism > > > will > > > > > > cause > > > > > > > >> many > > > > > > > >> > > > > >>> problems, > > > > > > > >> > > > > >>> > > including the increase of checkpoint time > for > > > > > multiple > > > > > > > >> > cascade > > > > > > > >> > > > > jobs, > > > > > > > >> > > > > >>> the > > > > > > > >> > > > > >>> > > increase of E2E data freshness time (several > > > > minutes > > > > > > or > > > > > > > >> even > > > > > > > >> > > > dozens > > > > > > > >> > > > > >>> of > > > > > > > >> > > > > >>> > > minutes), and the increase of the overall > > system > > > > > > > >> complexity. > > > > > > > >> > At > > > > > > > >> > > > the > > > > > > > >> > > > > >>> same > > > > > > > >> > > > > >>> > > time, the semantics of Checkpoint data > > > > segmentation > > > > > is > > > > > > > >> > unclear. > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > The current FLIP is the first phase of our > > whole > > > > > > proposal, > > > > > > > >> > and > > > > > > > >> > > > you > > > > > > > >> > > > > >>> can > > > > > > > >> > > > > >>> > find > > > > > > > >> > > > > >>> > > the follow-up plan in our future worker. In > > the > > > > > first > > > > > > > >> stage, > > > > > > > >> > we > > > > > > > >> > > > do > > > > > > > >> > > > > >>> not > > > > > > > >> > > > > >>> > want > > > > > > > >> > > > > >>> > > to modify the Flink mechanism. We'd like to > > > > realize > > > > > > basic > > > > > > > >> > > system > > > > > > > >> > > > > >>> > functions > > > > > > > >> > > > > >>> > > based on existing mechanisms in Flink, > > including > > > > the > > > > > > > >> > > relationship > > > > > > > >> > > > > >>> > > management of ETL and tables, and the basic > > data > > > > > > > >> consistency, > > > > > > > >> > > so > > > > > > > >> > > > we > > > > > > > >> > > > > >>> > choose > > > > > > > >> > > > > >>> > > Global Checkpoint in our FLIP. > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > We agree with you very much that event time > is > > > > more > > > > > > > >> suitable > > > > > > > >> > > for > > > > > > > >> > > > > data > > > > > > > >> > > > > >>> > > consistency management. We'd like consider > > this > > > > > > matter in > > > > > > > >> the > > > > > > > >> > > > > second > > > > > > > >> > > > > >>> or > > > > > > > >> > > > > >>> > > third stage after the current FLIP. We hope > to > > > > > > improve the > > > > > > > >> > > > > watermark > > > > > > > >> > > > > >>> > > mechanism in Flink to support barriers. As > you > > > > > > mentioned > > > > > > > >> in > > > > > > > >> > > your > > > > > > > >> > > > > >>> reply, > > > > > > > >> > > > > >>> > we > > > > > > > >> > > > > >>> > > can achieve data consistency based on > > timestamp, > > > > > while > > > > > > > >> > > > maintaining > > > > > > > >> > > > > >>> E2E > > > > > > > >> > > > > >>> > data > > > > > > > >> > > > > >>> > > freshness of seconds or even milliseconds > for > > > 10+ > > > > > > cascaded > > > > > > > >> > > jobs. > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > What do you think? Thanks > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > Best, > > > > > > > >> > > > > >>> > > Shammon > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > On Fri, Dec 9, 2022 at 6:13 PM Piotr > Nowojski > > < > > > > > > > >> > > > > pnowoj...@apache.org> > > > > > > > >> > > > > >>> > > wrote: > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > > Hi Shammon, > > > > > > > >> > > > > >>> > > > > > > > > > > >> > > > > >>> > > > Do I understand it correctly, that you > > > > effectively > > > > > > want > > > > > > > >> to > > > > > > > >> > > > expand > > > > > > > >> > > > > >>> the > > > > > > > >> > > > > >>> > > > checkpoint alignment mechanism across many > > > > > different > > > > > > > >> jobs > > > > > > > >> > and > > > > > > > >> > > > > hand > > > > > > > >> > > > > >>> over > > > > > > > >> > > > > >>> > > > checkpoint barriers from upstream to > > > downstream > > > > > jobs > > > > > > > >> using > > > > > > > >> > > the > > > > > > > >> > > > > >>> > > intermediate > > > > > > > >> > > > > >>> > > > tables? > > > > > > > >> > > > > >>> > > > > > > > > > > >> > > > > >>> > > > Re the watermarks for the "Rejected > > > > > Alternatives". I > > > > > > > >> don't > > > > > > > >> > > > > >>> understand > > > > > > > >> > > > > >>> > why > > > > > > > >> > > > > >>> > > > this has been rejected. Could you > elaborate > > on > > > > > this > > > > > > > >> point? > > > > > > > >> > > Here > > > > > > > >> > > > > >>> are a > > > > > > > >> > > > > >>> > > > couple of my thoughts on this matter, but > > > please > > > > > > > >> correct me > > > > > > > >> > > if > > > > > > > >> > > > > I'm > > > > > > > >> > > > > >>> > wrong, > > > > > > > >> > > > > >>> > > > as I haven't dived deeper into this topic. > > > > > > > >> > > > > >>> > > > > > > > > > > >> > > > > >>> > > > > As shown above, there are 2 watermarks > T1 > > > and > > > > > T2, > > > > > > T1 < > > > > > > > >> > T2. > > > > > > > >> > > > > >>> > > > > The StreamTask reads data in order: > > > > > > > >> > > > > >>> > > > V11，V12，V21，T1(channel1)，V13，T1(channel2). > > > > > > > >> > > > > >>> > > > > At this time, StreamTask will confirm > that > > > > > > watermark > > > > > > > >> T1 > > > > > > > >> > is > > > > > > > >> > > > > >>> completed, > > > > > > > >> > > > > >>> > > > but the data beyond > > > > > > > >> > > > > >>> > > > > T1 has been processed(V13) and the > results > > > are > > > > > > > >> written to > > > > > > > >> > > the > > > > > > > >> > > > > >>> sink > > > > > > > >> > > > > >>> > > > table. > > > > > > > >> > > > > >>> > > > > > > > > > > >> > > > > >>> > > > 1. I see the same "problem" with unaligned > > > > > > checkpoints > > > > > > > >> in > > > > > > > >> > > your > > > > > > > >> > > > > >>> current > > > > > > > >> > > > > >>> > > > proposal. > > > > > > > >> > > > > >>> > > > 2. I don't understand why this is a > problem? > > > > Just > > > > > > store > > > > > > > >> in > > > > > > > >> > > the > > > > > > > >> > > > > >>> "sink > > > > > > > >> > > > > >>> > > > table" what's the watermark (T1), and > > > downstream > > > > > > jobs > > > > > > > >> > should > > > > > > > >> > > > > >>> process > > > > > > > >> > > > > >>> > the > > > > > > > >> > > > > >>> > > > data with that "watermark" anyway. Record > > > "V13" > > > > > > should > > > > > > > >> be > > > > > > > >> > > > treated > > > > > > > >> > > > > >>> as > > > > > > > >> > > > > >>> > > > "early" data. Downstream jobs if: > > > > > > > >> > > > > >>> > > > a) they are streaming jobs, for example > > they > > > > > should > > > > > > > >> > > aggregate > > > > > > > >> > > > it > > > > > > > >> > > > > >>> in > > > > > > > >> > > > > >>> > > > windowed/temporal state, but they > shouldn't > > > > > produce > > > > > > the > > > > > > > >> > > result > > > > > > > >> > > > > that > > > > > > > >> > > > > >>> > > > contains it, as the watermark T2 was not > yet > > > > > > processed. > > > > > > > >> Or > > > > > > > >> > > they > > > > > > > >> > > > > >>> would > > > > > > > >> > > > > >>> > > just > > > > > > > >> > > > > >>> > > > pass that record as "early" data. > > > > > > > >> > > > > >>> > > > b) they are batch jobs, it looks to me > like > > > > batch > > > > > > jobs > > > > > > > >> > > > shouldn't > > > > > > > >> > > > > >>> take > > > > > > > >> > > > > >>> > > > "all available data", but only consider > "all > > > the > > > > > > data > > > > > > > >> until > > > > > > > >> > > > some > > > > > > > >> > > > > >>> > > > watermark", for example the latest > > available: > > > T1 > > > > > > > >> > > > > >>> > > > > > > > > > > >> > > > > >>> > > > 3. I'm pretty sure there are counter > > examples, > > > > > where > > > > > > > >> your > > > > > > > >> > > > > proposed > > > > > > > >> > > > > >>> > > > mechanism of using checkpoints (even > > aligned!) > > > > > will > > > > > > > >> produce > > > > > > > >> > > > > >>> > > > inconsistent data from the perspective of > > the > > > > > event > > > > > > > >> time. > > > > > > > >> > > > > >>> > > > a) For example what if one of your "ETL" > > > jobs, > > > > > > has the > > > > > > > >> > > > > following > > > > > > > >> > > > > >>> DAG: > > > > > > > >> > > > > >>> > > > [image: flip276.jpg] > > > > > > > >> > > > > >>> > > > Even if you use aligned checkpoints for > > > > > > committing the > > > > > > > >> > data > > > > > > > >> > > > to > > > > > > > >> > > > > >>> the > > > > > > > >> > > > > >>> > sink > > > > > > > >> > > > > >>> > > > table, the watermarks of "Window1" and > > > "Window2" > > > > > are > > > > > > > >> > > completely > > > > > > > >> > > > > >>> > > > independent. The sink table might easily > > have > > > > data > > > > > > from > > > > > > > >> the > > > > > > > >> > > > > >>> > Src1/Window1 > > > > > > > >> > > > > >>> > > > from the event time T1 and Src2/Window2 > from > > > > later > > > > > > event > > > > > > > >> > time > > > > > > > >> > > > T2. > > > > > > > >> > > > > >>> > > > b) I think the same applies if you have > > two > > > > > > completely > > > > > > > >> > > > > >>> independent > > > > > > > >> > > > > >>> > ETL > > > > > > > >> > > > > >>> > > > jobs writing either to the same sink > table, > > or > > > > two > > > > > > to > > > > > > > >> > > different > > > > > > > >> > > > > >>> sink > > > > > > > >> > > > > >>> > > tables > > > > > > > >> > > > > >>> > > > (that are both later used in the same > > > downstream > > > > > > job). > > > > > > > >> > > > > >>> > > > > > > > > > > >> > > > > >>> > > > 4a) I'm not sure if I like the idea of > > > > > centralising > > > > > > the > > > > > > > >> > whole > > > > > > > >> > > > > >>> system in > > > > > > > >> > > > > >>> > > > this way. If you have 10 jobs, the > > likelihood > > > of > > > > > the > > > > > > > >> > > checkpoint > > > > > > > >> > > > > >>> failure > > > > > > > >> > > > > >>> > > > will be 10 times higher, and/or the > duration > > > of > > > > > the > > > > > > > >> > > checkpoint > > > > > > > >> > > > > can > > > > > > > >> > > > > >>> be > > > > > > > >> > > > > >>> > > much > > > > > > > >> > > > > >>> > > > much longer (especially under > backpressure). > > > And > > > > > > this is > > > > > > > >> > > > actually > > > > > > > >> > > > > >>> > > already a > > > > > > > >> > > > > >>> > > > limitation of Apache Flink (global > > checkpoints > > > > are > > > > > > more > > > > > > > >> > prone > > > > > > > >> > > > to > > > > > > > >> > > > > >>> fail > > > > > > > >> > > > > >>> > the > > > > > > > >> > > > > >>> > > > larger the scale), so I would be anxious > > about > > > > > > making it > > > > > > > >> > > > > >>> potentially > > > > > > > >> > > > > >>> > > even a > > > > > > > >> > > > > >>> > > > larger issue. > > > > > > > >> > > > > >>> > > > 4b) I'm also worried about increased > > > complexity > > > > of > > > > > > the > > > > > > > >> > system > > > > > > > >> > > > > after > > > > > > > >> > > > > >>> > > adding > > > > > > > >> > > > > >>> > > > the global checkpoint, and additional > > > (single?) > > > > > > point of > > > > > > > >> > > > failure. > > > > > > > >> > > > > >>> > > > 5. Such a design would also not work if we > > > ever > > > > > > wanted > > > > > > > >> to > > > > > > > >> > > have > > > > > > > >> > > > > task > > > > > > > >> > > > > >>> > local > > > > > > > >> > > > > >>> > > > checkpoints. > > > > > > > >> > > > > >>> > > > > > > > > > > >> > > > > >>> > > > All in all, it seems to me like actually > the > > > > > > watermarks > > > > > > > >> and > > > > > > > >> > > > even > > > > > > > >> > > > > >>> time > > > > > > > >> > > > > >>> > are > > > > > > > >> > > > > >>> > > > the better concept in this context that > > should > > > > > have > > > > > > been > > > > > > > >> > used > > > > > > > >> > > > for > > > > > > > >> > > > > >>> > > > synchronising and data consistency across > > the > > > > > whole > > > > > > > >> system. > > > > > > > >> > > > > >>> > > > > > > > > > > >> > > > > >>> > > > Best, > > > > > > > >> > > > > >>> > > > Piotrek > > > > > > > >> > > > > >>> > > > > > > > > > > >> > > > > >>> > > > czw., 1 gru 2022 o 11:50 Shammon FY < > > > > > > zjur...@gmail.com> > > > > > > > >> > > > > >>> napisał(a): > > > > > > > >> > > > > >>> > > > > > > > > > > >> > > > > >>> > > >> Hi @Martijn > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> Thanks for your comments, and I'd like to > > > reply > > > > > to > > > > > > them > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> 1. It sounds good to me, I'll update the > > > > content > > > > > > > >> structure > > > > > > > >> > > in > > > > > > > >> > > > > FLIP > > > > > > > >> > > > > >>> > later > > > > > > > >> > > > > >>> > > >> and give the problems first. > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> 2. "Each ETL job creates snapshots with > > > > > checkpoint > > > > > > > >> info on > > > > > > > >> > > > sink > > > > > > > >> > > > > >>> tables > > > > > > > >> > > > > >>> > > in > > > > > > > >> > > > > >>> > > >> Table Store" -> That reads like you're > > > > proposing > > > > > > that > > > > > > > >> > > > snapshots > > > > > > > >> > > > > >>> need > > > > > > > >> > > > > >>> > to > > > > > > > >> > > > > >>> > > >> be > > > > > > > >> > > > > >>> > > >> written to Table Store? > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> Yes. To support the data consistency in > the > > > > FLIP, > > > > > > we > > > > > > > >> need > > > > > > > >> > to > > > > > > > >> > > > get > > > > > > > >> > > > > >>> > through > > > > > > > >> > > > > >>> > > >> checkpoints in Flink and snapshots in > > store, > > > > this > > > > > > > >> > requires a > > > > > > > >> > > > > close > > > > > > > >> > > > > >>> > > >> combination of Flink and store > > > implementation. > > > > In > > > > > > the > > > > > > > >> > first > > > > > > > >> > > > > stage > > > > > > > >> > > > > >>> we > > > > > > > >> > > > > >>> > > plan > > > > > > > >> > > > > >>> > > >> to implement it based on Flink and Table > > > Store > > > > > > only, > > > > > > > >> > > snapshots > > > > > > > >> > > > > >>> written > > > > > > > >> > > > > >>> > > to > > > > > > > >> > > > > >>> > > >> external storage don't support > consistency. > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> 3. If you introduce a MetaService, it > > becomes > > > > the > > > > > > > >> single > > > > > > > >> > > point > > > > > > > >> > > > > of > > > > > > > >> > > > > >>> > > failure > > > > > > > >> > > > > >>> > > >> because it coordinates everything. But I > > > can't > > > > > find > > > > > > > >> > anything > > > > > > > >> > > > in > > > > > > > >> > > > > >>> the > > > > > > > >> > > > > >>> > FLIP > > > > > > > >> > > > > >>> > > >> on > > > > > > > >> > > > > >>> > > >> making the MetaService high available or > > how > > > to > > > > > > deal > > > > > > > >> with > > > > > > > >> > > > > >>> failovers > > > > > > > >> > > > > >>> > > there. > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> I think you raise a very important > problem > > > and > > > > I > > > > > > > >> missed it > > > > > > > >> > > in > > > > > > > >> > > > > >>> FLIP. > > > > > > > >> > > > > >>> > The > > > > > > > >> > > > > >>> > > >> MetaService is a single point and should > > > > support > > > > > > > >> failover, > > > > > > > >> > > we > > > > > > > >> > > > > >>> will do > > > > > > > >> > > > > >>> > it > > > > > > > >> > > > > >>> > > >> in > > > > > > > >> > > > > >>> > > >> future in the first stage we only support > > > > > > standalone > > > > > > > >> mode, > > > > > > > >> > > THX > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> 4. The FLIP states under Rejected > > > Alternatives > > > > > > > >> "Currently > > > > > > > >> > > > > >>> watermark in > > > > > > > >> > > > > >>> > > >> Flink cannot align data." which is not > > true, > > > > > given > > > > > > that > > > > > > > >> > > there > > > > > > > >> > > > is > > > > > > > >> > > > > >>> > > FLIP-182 > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-182%3A+Support+watermark+alignment+of+FLIP-27+Sources > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> Watermark alignment in FLIP-182 is > > different > > > > from > > > > > > > >> > > requirements > > > > > > > >> > > > > >>> > > "watermark > > > > > > > >> > > > > >>> > > >> align data" in our FLIP. FLIP-182 aims to > > fix > > > > > > watermark > > > > > > > >> > > > > >>> generation in > > > > > > > >> > > > > >>> > > >> different sources for "slight imbalance > or > > > data > > > > > > skew", > > > > > > > >> > which > > > > > > > >> > > > > >>> means in > > > > > > > >> > > > > >>> > > some > > > > > > > >> > > > > >>> > > >> cases the source must generate watermark > > even > > > > if > > > > > > they > > > > > > > >> > should > > > > > > > >> > > > > not. > > > > > > > >> > > > > >>> When > > > > > > > >> > > > > >>> > > the > > > > > > > >> > > > > >>> > > >> operator collects watermarks, the data > > > > processing > > > > > > is as > > > > > > > >> > > > > described > > > > > > > >> > > > > >>> in > > > > > > > >> > > > > >>> > our > > > > > > > >> > > > > >>> > > >> FLIP, and the data cannot be aligned > > through > > > > the > > > > > > > >> barrier > > > > > > > >> > > like > > > > > > > >> > > > > >>> > > Checkpoint. > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> 5. Given the MetaService role, it feels > > like > > > > this > > > > > > is > > > > > > > >> > > > > introducing a > > > > > > > >> > > > > >>> > tight > > > > > > > >> > > > > >>> > > >> dependency between Flink and the Table > > Store. > > > > How > > > > > > > >> > pluggable > > > > > > > >> > > is > > > > > > > >> > > > > >>> this > > > > > > > >> > > > > >>> > > >> solution, given the changes that need to > be > > > > made > > > > > to > > > > > > > >> Flink > > > > > > > >> > in > > > > > > > >> > > > > >>> order to > > > > > > > >> > > > > >>> > > >> support this? > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> This is a good question, and I will try > to > > > > expand > > > > > > it. > > > > > > > >> Most > > > > > > > >> > > of > > > > > > > >> > > > > the > > > > > > > >> > > > > >>> work > > > > > > > >> > > > > >>> > > >> will > > > > > > > >> > > > > >>> > > >> be completed in the Table Store, such as > > the > > > > new > > > > > > > >> > > > SplitEnumerator > > > > > > > >> > > > > >>> and > > > > > > > >> > > > > >>> > > >> Source > > > > > > > >> > > > > >>> > > >> implementation. The changes in Flink are > as > > > > > > followed: > > > > > > > >> > > > > >>> > > >> 1) Flink job should put its job id in > > context > > > > > when > > > > > > > >> > creating > > > > > > > >> > > > > >>> > source/sink > > > > > > > >> > > > > >>> > > to > > > > > > > >> > > > > >>> > > >> help MetaService to create relationship > > > between > > > > > > source > > > > > > > >> and > > > > > > > >> > > > sink > > > > > > > >> > > > > >>> > tables, > > > > > > > >> > > > > >>> > > >> it's tiny > > > > > > > >> > > > > >>> > > >> 2) Notify a listener when job is > terminated > > > in > > > > > > Flink, > > > > > > > >> and > > > > > > > >> > > the > > > > > > > >> > > > > >>> listener > > > > > > > >> > > > > >>> > > >> implementation in Table Store will send > > > "delete > > > > > > event" > > > > > > > >> to > > > > > > > >> > > > > >>> MetaService. > > > > > > > >> > > > > >>> > > >> 3) The changes are related to Flink > > > Checkpoint > > > > > > includes > > > > > > > >> > > > > >>> > > >> a) Support triggering checkpoint with > > > > > checkpoint > > > > > > id > > > > > > > >> by > > > > > > > >> > > > > >>> > SplitEnumerator > > > > > > > >> > > > > >>> > > >> b) Create the SplitEnumerator in Table > > > Store > > > > > > with a > > > > > > > >> > > strategy > > > > > > > >> > > > > to > > > > > > > >> > > > > >>> > > perform > > > > > > > >> > > > > >>> > > >> the specific checkpoint when all > > > > > > "SplitEnumerator"s in > > > > > > > >> the > > > > > > > >> > > job > > > > > > > >> > > > > >>> manager > > > > > > > >> > > > > >>> > > >> trigger it. > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> Best, > > > > > > > >> > > > > >>> > > >> Shammon > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> On Thu, Dec 1, 2022 at 3:43 PM Martijn > > > Visser < > > > > > > > >> > > > > >>> > martijnvis...@apache.org > > > > > > > >> > > > > >>> > > > > > > > > > > >> > > > > >>> > > >> wrote: > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > >> > Hi all, > > > > > > > >> > > > > >>> > > >> > > > > > > > > >> > > > > >>> > > >> > A couple of first comments on this: > > > > > > > >> > > > > >>> > > >> > 1. I'm missing the problem statement in > > the > > > > > > overall > > > > > > > >> > > > > >>> introduction. It > > > > > > > >> > > > > >>> > > >> > immediately goes into proposal mode, I > > > would > > > > > > like to > > > > > > > >> > first > > > > > > > >> > > > > read > > > > > > > >> > > > > >>> what > > > > > > > >> > > > > >>> > > is > > > > > > > >> > > > > >>> > > >> the > > > > > > > >> > > > > >>> > > >> > actual problem, before diving into > > > solutions. > > > > > > > >> > > > > >>> > > >> > 2. "Each ETL job creates snapshots with > > > > > > checkpoint > > > > > > > >> info > > > > > > > >> > on > > > > > > > >> > > > > sink > > > > > > > >> > > > > >>> > tables > > > > > > > >> > > > > >>> > > >> in > > > > > > > >> > > > > >>> > > >> > Table Store" -> That reads like you're > > > > > proposing > > > > > > > >> that > > > > > > > >> > > > > snapshots > > > > > > > >> > > > > >>> > need > > > > > > > >> > > > > >>> > > >> to be > > > > > > > >> > > > > >>> > > >> > written to Table Store? > > > > > > > >> > > > > >>> > > >> > 3. If you introduce a MetaService, it > > > becomes > > > > > the > > > > > > > >> single > > > > > > > >> > > > point > > > > > > > >> > > > > >>> of > > > > > > > >> > > > > >>> > > >> failure > > > > > > > >> > > > > >>> > > >> > because it coordinates everything. But > I > > > > can't > > > > > > find > > > > > > > >> > > anything > > > > > > > >> > > > > in > > > > > > > >> > > > > >>> the > > > > > > > >> > > > > >>> > > >> FLIP on > > > > > > > >> > > > > >>> > > >> > making the MetaService high available > or > > > how > > > > to > > > > > > deal > > > > > > > >> > with > > > > > > > >> > > > > >>> failovers > > > > > > > >> > > > > >>> > > >> there. > > > > > > > >> > > > > >>> > > >> > 4. The FLIP states under Rejected > > > > Alternatives > > > > > > > >> > "Currently > > > > > > > >> > > > > >>> watermark > > > > > > > >> > > > > >>> > in > > > > > > > >> > > > > >>> > > >> > Flink cannot align data." which is not > > > true, > > > > > > given > > > > > > > >> that > > > > > > > >> > > > there > > > > > > > >> > > > > is > > > > > > > >> > > > > >>> > > >> FLIP-182 > > > > > > > >> > > > > >>> > > >> > > > > > > > > >> > > > > >>> > > >> > > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-182%3A+Support+watermark+alignment+of+FLIP-27+Sources > > > > > > > >> > > > > >>> > > >> > > > > > > > > >> > > > > >>> > > >> > 5. Given the MetaService role, it feels > > > like > > > > > > this is > > > > > > > >> > > > > >>> introducing a > > > > > > > >> > > > > >>> > > tight > > > > > > > >> > > > > >>> > > >> > dependency between Flink and the Table > > > Store. > > > > > How > > > > > > > >> > > pluggable > > > > > > > >> > > > is > > > > > > > >> > > > > >>> this > > > > > > > >> > > > > >>> > > >> > solution, given the changes that need > to > > be > > > > > made > > > > > > to > > > > > > > >> > Flink > > > > > > > >> > > in > > > > > > > >> > > > > >>> order > > > > > > > >> > > > > >>> > to > > > > > > > >> > > > > >>> > > >> > support this? > > > > > > > >> > > > > >>> > > >> > > > > > > > > >> > > > > >>> > > >> > Best regards, > > > > > > > >> > > > > >>> > > >> > > > > > > > > >> > > > > >>> > > >> > Martijn > > > > > > > >> > > > > >>> > > >> > > > > > > > > >> > > > > >>> > > >> > > > > > > > > >> > > > > >>> > > >> > On Thu, Dec 1, 2022 at 4:49 AM Shammon > > FY < > > > > > > > >> > > > zjur...@gmail.com> > > > > > > > >> > > > > >>> > wrote: > > > > > > > >> > > > > >>> > > >> > > > > > > > > >> > > > > >>> > > >> > > Hi devs: > > > > > > > >> > > > > >>> > > >> > > > > > > > > > >> > > > > >>> > > >> > > I'd like to start a discussion about > > > > > FLIP-276: > > > > > > Data > > > > > > > >> > > > > >>> Consistency of > > > > > > > >> > > > > >>> > > >> > > Streaming and Batch ETL in Flink and > > > Table > > > > > > > >> Store[1]. > > > > > > > >> > In > > > > > > > >> > > > the > > > > > > > >> > > > > >>> whole > > > > > > > >> > > > > >>> > > data > > > > > > > >> > > > > >>> > > >> > > stream processing, there are > > consistency > > > > > > problems > > > > > > > >> such > > > > > > > >> > > as > > > > > > > >> > > > > how > > > > > > > >> > > > > >>> to > > > > > > > >> > > > > >>> > > >> manage > > > > > > > >> > > > > >>> > > >> > the > > > > > > > >> > > > > >>> > > >> > > dependencies of multiple jobs and > > tables, > > > > how > > > > > > to > > > > > > > >> > define > > > > > > > >> > > > and > > > > > > > >> > > > > >>> handle > > > > > > > >> > > > > >>> > > E2E > > > > > > > >> > > > > >>> > > >> > > delays, and how to ensure the data > > > > > consistency > > > > > > of > > > > > > > >> > > queries > > > > > > > >> > > > on > > > > > > > >> > > > > >>> > flowing > > > > > > > >> > > > > >>> > > >> > data? > > > > > > > >> > > > > >>> > > >> > > This FLIP aims to support data > > > consistency > > > > > and > > > > > > > >> answer > > > > > > > >> > > > these > > > > > > > >> > > > > >>> > > questions. > > > > > > > >> > > > > >>> > > >> > > > > > > > > > >> > > > > >>> > > >> > > I'v discussed the details of this > FLIP > > > with > > > > > > > >> @Jingsong > > > > > > > >> > > Lee > > > > > > > >> > > > > and > > > > > > > >> > > > > >>> > > >> @libenchao > > > > > > > >> > > > > >>> > > >> > > offline several times. We hope to > > support > > > > > data > > > > > > > >> > > consistency > > > > > > > >> > > > > of > > > > > > > >> > > > > >>> > > queries > > > > > > > >> > > > > >>> > > >> on > > > > > > > >> > > > > >>> > > >> > > tables, managing relationships > between > > > > Flink > > > > > > jobs > > > > > > > >> and > > > > > > > >> > > > tables > > > > > > > >> > > > > >>> and > > > > > > > >> > > > > >>> > > >> revising > > > > > > > >> > > > > >>> > > >> > > tables on streaming in Flink and > Table > > > > Store > > > > > to > > > > > > > >> > improve > > > > > > > >> > > > the > > > > > > > >> > > > > >>> whole > > > > > > > >> > > > > >>> > > data > > > > > > > >> > > > > >>> > > >> > > stream processing. > > > > > > > >> > > > > >>> > > >> > > > > > > > > > >> > > > > >>> > > >> > > Looking forward to your feedback. > > > > > > > >> > > > > >>> > > >> > > > > > > > > > >> > > > > >>> > > >> > > [1] > > > > > > > >> > > > > >>> > > >> > > > > > > > > > >> > > > > >>> > > >> > > > > > > > > > >> > > > > >>> > > >> > > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-276%3A+Data+Consistency+of+Streaming+and+Batch+ETL+in+Flink+and+Table+Store > > > > > > > >> > > > > >>> > > >> > > > > > > > > > >> > > > > >>> > > >> > > > > > > > > > >> > > > > >>> > > >> > > Best, > > > > > > > >> > > > > >>> > > >> > > Shammon > > > > > > > >> > > > > >>> > > >> > > > > > > > > > >> > > > > >>> > > >> > > > > > > > > >> > > > > >>> > > >> > > > > > > > >> > > > > >>> > > > > > > > > > > >> > > > > >>> > > > > > > > > > >> > > > > >>> > > > > > > > > >> > > > > >>> > > > > > > > >> > > > > >> > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >
