Re: [VOTE] FLIP-374: Adding a separate configuration for specifying Java Options of the SQL Gateway
+1(binding), good job! Best, Shammon FY On Wed, Oct 11, 2023 at 10:18 AM Benchao Li wrote: > +1 (binding) > > Rui Fan <1996fan...@gmail.com> 于2023年10月11日周三 10:17写道： > > > > +1(binding) > > > > Best, > > Rui > > > > On Wed, Oct 11, 2023 at 10:07 AM Yangze Guo wrote: > > > > > Hi everyone, > > > > > > I'd like to start the vote of FLIP-374 [1]. This FLIP is discussed in > > > the thread [2]. > > > > > > The vote will be open for at least 72 hours. Unless there is an > > > objection, I'll try to close it by October 16, 2023 if we have > > > received sufficient votes. > > > > > > [1] > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-374%3A+Adding+a+separate+configuration+for+specifying+Java+Options+of+the+SQL+Gateway > > > [2] https://lists.apache.org/thread/g4vl8mgnwgl7vjyvjy6zrc8w54b2lthv > > > > > > Best, > > > Yangze Guo > > > > > > > -- > > Best, > Benchao Li >
Re: [VOTE] FLIP-367: Support Setting Parallelism for Table/SQL Sources
+1 (binding) On Mon, Oct 9, 2023 at 11:12 AM Benchao Li wrote: > +1 (binding) > > Zhanghao Chen 于2023年10月9日周一 10:20写道： > > > > Hi All, > > > > Thanks for all the feedback on FLIP-367: Support Setting Parallelism for > Table/SQL Sources [1][2]. > > > > I'd like to start a vote for FLIP-367. The vote will be open until Oct > 12th 12:00 PM GMT) unless there is an objection or insufficient votes. > > > > [1] > https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=263429150 > > [2] https://lists.apache.org/thread/gtpswl42jzv0c9o3clwqskpllnw8rh87 > > > > Best, > > Zhanghao Chen > > > > -- > > Best, > Benchao Li >
[RESULT][VOTE] FLIP-314: Support Customized Job Lineage Listener
Hi devs, Thanks everyone for your review and the votes! I am happy to announce that FLIP-314: Support Customized Job Lineage Listener [1] has been accepted. There are 5 binding votes and 4 non-binding vote [2]: - Jing Ge(binding) - Feng Jin - Leonard Xu(binding) - Yangze Guo(binding) - Yun Tang(binding) - 曹帝胄 - Chen Zhanghao - Rui Fan(binding) - Yuepeng Pan There is no disapproving vote. Best, Shammon FY [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-314%3A+Support+Customized+Job+Lineage+Listener [2] https://lists.apache.org/thread/dxdqjc0dd8rf1vbdg755zo1n2zj1tj8d
Re: [DISCUSS] FLIP-370 : Support Balanced Tasks Scheduling
Thanks Rui, I check the codes and you're right. As you described above, the entire process is actually two independent steps from slot to TM and task to slot. Currenlty we use option `cluster.evenly-spread-out-slots` for both of them. Can we provide different options for the two steps, such as ANY/SLOTS for RM and ANY/TASKS for slot pool? I want this because we would like to add a new slot to TM strategy such as SLOTS_NUM in the future for OLAP to improve the performance for olap jobs, which will use TASKS strategy for task to slot. cc Guoyangze Best, Shammon FY On Fri, Oct 6, 2023 at 6:19 PM xiangyu feng wrote: > Thanks Yuepeng and Rui for driving this Discussion. > > Internally when we try to use Flink 1.17.1 in production, we are also > suffering from the unbalanced task distribution problem for jobs with high > qps and complex dag. So +1 for the overall proposal. > > Some questions about the details: > > 1, About the waiting mechanism: Will the waiting mechanism happen only in > the second level 'assigning slots to TM'? IIUC, the first level 'assigning > Tasks to Slots' needs only the asynchronous slot result from slotpool. > > 2, About the slot LoadingWeight: it is reasonable to use the number of > tasks by default in the beginning, but it would be better if this could be > easily extended in future to distinguish between CPU-intensive and > IO-intensive workloads. In some cases, TMs may have IO bottlenecks but > others have CPU bottlenecks. > > Regards, > Xiangyu > > > Yuepeng Pan 于2023年10月5日周四 18:34写道： > > > Hi, Zhu Zhu, > > > > Thanks for your feedback! > > > > > I think we can introduce a new config option > > > `taskmanager.load-balance.mode`, > > > which accepts "None"/"Slots"/"Tasks". `cluster.evenly-spread-out-slots` > > > can be superseded by the "Slots" mode and get deprecated. In the future > > > it can support more mode, e.g. "CpuCores", to work better for jobs with > > > fine-grained resources. The proposed config option > > > `slot.request.max-interval` > > > then can be renamed to > > `taskmanager.load-balance.request-stablizing-timeout` > > > to show its relation with the feature. The proposed > > `slot.sharing-strategy` > > > is not needed, because the configured "Tasks" mode will do the work. > > > > The new proposed configuration option sounds good to me. > > > > I have a small question, If we set our configuration value to 'Tasks,' it > > will initiate two processes: balancing the allocation of task quantities > at > > the slot level and balancing the number of tasks across TaskManagers > (TMs). > > Alternatively, if we configure it as 'Slots,' the system will employ the > > LocalPreferred allocation policy (which is the default) when assigning > > tasks to slots, and it will ensure that the number of slots used across > TMs > > is balanced. > > Does this configuration essentially combine a balanced selection > strategy > > across two dimensions into fixed configuration items, right? > > > > I would appreciate it if you could correct me if I've made any errors. > > > > Best, > > Yuepeng. > > >
Re: [DISCUSS] FLIP-370 : Support Balanced Tasks Scheduling
Thanks Yuepeng for initiating this discussion. +1 in general too, in fact we have implemented a similar mechanism internally to ensure a balanced allocation of tasks to slots, it works well. Some comments about the mechanism 1. This mechanism will be only supported in `SlotPool` or both `SlotPool` and `DeclarativeSlotPool`? Currently the two slot pools are used in different schedulers. I think this will also bring value to `DeclarativeSlotPool`, but currently FLIP content seems to be based on `SlotPool`, right? 2. In fine-grained resource management, we can set different resource requirements for different nodes, which means that the resources of each slot are different. What should be done when the slot selected by the round-robin strategy cannot meet the resource requirements? Will this lead to the failure of the balance strategy? 3. Is the assignment of tasks to slots balanced based on region or job level? When multiple TMs fail over, will it cause the balancing strategy to fail or even worse? What is the current processing strategy? For Zhuzhu and Rui: IIUC, the overall balance is divided into two parts: slot to TM and task to slot. 1. Slot to TM is guaranteed by SlotManager in ResourceManager 2. Task to slot is guaranteed by the slot pool in JM These two are completely independent, what are the benefits of unifying these two into one option? Also, do we want to share the same option between SlotPool in JM and SlotManager in RM? This sounds a bit strange. Best, Shammon FY On Thu, Sep 28, 2023 at 12:08 PM Rui Fan <1996fan...@gmail.com> wrote: > Hi Zhu Zhu, > > Thanks for your feedback here! > > You are right, user needs to set 2 options: > - cluster.evenly-spread-out-slots=true > - slot.sharing-strategy=TASK_BALANCED_PREFERRED > > Update it to one option is useful at user side, so > `taskmanager.load-balance.mode` sounds good to me. > I want to check some points and behaviors about this option: > > 1. The default value is None, right? > 2. When it's set to Tasks, how to assign slots to TM? > - Option1: It's just check task number > - Option2: It''s check the slot number first, then check the > task number when the slot number is the same. > > Giving an example to explain what's the difference between them: > > - A session cluster has 2 flink jobs, they are jobA and jobB > - Each TM has 4 slots. > - The task number of one slot of jobA is 3 > - The task number of one slot of jobB is 1 > - We have 2 TaskManagers: > - tm1 runs 3 slots of jobB, so tm1 runs 3 tasks > - tm2 runs 1 slot of jobA, and 1 slot of jobB, so tm2 runs 4 tasks. > > Now, we need to run a new slot, which tm should offer it? > - Option1: If we just check the task number, the tm1 is better. > - Option2: If we check the slot number first, and then check task, the tm2 > is better > > The original FLIP selected option2, that's why we didn't add the > third option. The option2 didn't break the semantics when > `cluster.evenly-spread-out-slots` is true, and it just improve the > behavior without the semantics is changed. > > In the other hands, if we choose option2, when user set > `taskmanager.load-balance.mode` is Tasks. It also can achieve > the goal when it's Slots. > > So I think the `Slots` enum isn't needed if we choose option2. > Of course, If we choose the option1, the enum is needed. > > Looking forward to your feedback, thanks~ > > Best, > Rui > > On Wed, Sep 27, 2023 at 9:11 PM Zhu Zhu wrote: > > > Thanks Yuepeng and Rui for creating this FLIP. > > > > +1 in general > > The idea is straight forward: best-effort gather all the slot requests > > and offered slots to form an overview before assigning slots, trying to > > balance the loads of task managers when assigning slots. > > > > I have one comment regarding the configuration for ease of use: > > > > IIUC, this FLIP uses an existing config 'cluster.evenly-spread-out-slots' > > as the main switch of the new feature. That is, from user perspective, > > with this improvement, the 'cluster.evenly-spread-out-slots' feature not > > only balances the number of slots on task managers, but also balances the > > number of tasks. This is a behavior change anyway. Besides that, it also > > requires users to set 'slot.sharing-strategy' to > 'TASK_BALANCED_PREFERRED' > > to balance the tasks in each slot. > > > > I think we can introduce a new config option > > `taskmanager.load-balance.mode`, > > which accepts "None"/"Slots"/"Tasks". `cluster.evenly-spread-out-slots` > > can be superseded by the "Slots" mode and get deprecated. In the future > > it can support more mode, e.g. "CpuCores", to work better for jobs with > > fine-gr
Re: [VOTE] FLIP-362: Support minimum resource limitation
+1(binding), thanks for the proposal. Best, Shammon FY On Mon, Sep 25, 2023 at 11:20 PM Jing Ge wrote: > +1(binding) Thanks! > > Best regards, > Jing > > On Mon, Sep 25, 2023 at 3:48 AM Chen Zhanghao > wrote: > > > Thanks for driving this. +1 (non-binding) > > > > Best, > > Zhanghao Chen > > > > 发件人: xiangyu feng > > 发送时间: 2023年9月25日 17:38 > > 收件人: dev@flink.apache.org > > 主题: [VOTE] FLIP-362: Support minimum resource limitation > > > > Hi all, > > > > I would like to start the vote for FLIP-362: Support minimum resource > > limitation[1]. > > This FLIP was discussed in this thread [2]. > > > > The vote will be open for at least 72 hours unless there is an objection > or > > insufficient votes. > > > > Regards, > > Xiangyu > > > > [1] > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-362%3A+Support+minimum+resource+limitation > > [2] https://lists.apache.org/thread/m2v9n4yynm97v8swhqj2o5k0sqlb5ym4 > > >
[VOTE] FLIP-314: Support Customized Job Lineage Listener
Hi devs, Thanks for all the feedback on FLIP-314: Support Customized Job Lineage Listener [1] in thread [2]. I would like to start a vote for it. The vote will be opened for at least 72 hours unless there is an objection or insufficient votes. [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-314%3A+Support+Customized+Job+Lineage+Listener [2] https://lists.apache.org/thread/wopprvp3ww243mtw23nj59p57cghh7mc Best, Shammon FY
Re: FW: RE: [DISCUSS] FLIP-314: Support Customized Job Lineage Listener
Hi David, Do you want the detailed topology for Flink job? You can get `JobDetailsInfo` in `RestCusterClient` with the submitted job id, it has `String jsonPlan`. You can parse the json plan to get all steps and relations between them in a Flink job. Hope this can help you, thanks! Best, Shammon FY On Tue, Sep 19, 2023 at 11:46 PM David Radley wrote: > Hi there, > I am looking at the interfaces. If I am reading it correctly,there is one > relationship between the source and sink and this relationship represents > the operational lineage. Lineage is usually represented as asset -> process > - > asset – see for example > https://egeria-project.org/features/lineage-management/overview/#the-lineage-graph > > Maybe I am missing it, but it seems to be that it would be useful to store > the process in the lineage graph. > > It is useful to have the top level lineage as source -> Flink job -> sink. > Where the Flink job is the process, but also to have this asset -> process > -> asset pattern for each of the steps in the job. If this is present, > please could you point me to it, > > Kind regards, David. > > > > > > From: David Radley > Date: Tuesday, 19 September 2023 at 16:11 > To: dev@flink.apache.org > Subject: [EXTERNAL] RE: [DISCUSS] FLIP-314: Support Customized Job Lineage > Listener > Hi, > I notice that there is an experimental lineage integration for Flink with > OpenLineage https://openlineage.io/docs/integrations/flink . I think > this feature would allow for a superior Flink OpenLineage integration, > Kind regards, David. > > From: XTransfer > Date: Tuesday, 19 September 2023 at 15:47 > To: dev@flink.apache.org > Subject: [EXTERNAL] Re: [DISCUSS] FLIP-314: Support Customized Job Lineage > Listener > Thanks Shammon for this proposal. > > That’s helpful for collecting the lineage of Flink tasks. > Looking forward to its implementation. > > Best, > Jiabao > > > > 2023年9月18日 20:56，Leonard Xu 写道： > > > > Thanks Shammon for the informations, the comment makes the lifecycle > clearer. > > +1 > > > > > > Best, > > Leonard > > > > > >> On Sep 18, 2023, at 7:54 PM, Shammon FY wrote: > >> > >> Hi devs, > >> > >> After discussing with @Qingsheng, I fixed a minor issue of the lineage > lifecycle in `StreamExecutionEnvironment`. I have added the comment to > explain that the lineage information in `StreamExecutionEnvironment` will > be consistent with that of transformations. When users clear the existing > transformations, the added lineage information will also be deleted. > >> > >> Please help to review it again, and If there are no more concerns about > FLIP-314[1], I would like to start voting later, thanks. cc @ <>Leonard > >> > >> Best, > >> Shammon FY > >> > >> On Mon, Jul 17, 2023 at 3:43 PM Shammon FY zjur...@gmail.com>> wrote: > >> Hi devs, > >> > >> Thanks for all the valuable feedback. If there are no more concerns > about FLIP-314[1], I would like to start voting later, thanks. > >> > >> > >> [1] > https://cwiki.apache.org/confluence/display/FLINK/FLIP-314%3A+Support+Customized+Job+Lineage+Listener > < > https://cwiki.apache.org/confluence/display/FLINK/FLIP-314%3A+Support+Customized+Job+Lineage+Listener > > > >> > >> Best, > >> Shammon FY > >> > >> > >> On Wed, Jul 12, 2023 at 11:18 AM Shammon FY zjur...@gmail.com>> wrote: > >> Thanks for the valuable feedback, Leonard. > >> > >> I have discussed with Leonard off-line. We have reached some > conclusions about these issues and I have updated the FLIP as follows: > >> > >> 1. Simplify the `LineageEdge` interface by creating an edge from one > source vertex to sink vertex. > >> 2. Remove the `TableColumnSourceLineageVertex` interface and update > `TableColumnLineageEdge` to create an edge from columns in one source to > each sink column. > >> 3. Rename `SupportsLineageVertex` to `LineageVertexProvider` > >> 4. Add method `addLineageEdges(LineageEdge ... edges)` in > `StreamExecutionEnviroment` for datastream job and remove previous methods > in `DataStreamSource` and `DataStreamSink`. > >> > >> Looking forward to your feedback, thanks. > >> > >> Best, > >> Shammon FY > > > > Unless otherwise stated above: > > IBM United Kingdom Limited > Registered in England and Wales with number 741598 > Registered office: PO Box 41, North Harbour, Portsmouth, Hants. PO6 3AU > > Unless otherwise stated above: > > IBM United Kingdom Limited > Registered in England and Wales with number 741598 > Registered office: PO Box 41, North Harbour, Portsmouth, Hants. PO6 3AU >
Re: [Discuss] FLIP-366: Support standard YAML for FLINK configuration
+1 for the proposal, thanks for driving. Bet, Shammon FY On Fri, Sep 22, 2023 at 12:41 PM Yangze Guo wrote: > Thanks for driving this, +1 for the proposal. > > Best, > Yangze Guo > > > On Fri, Sep 22, 2023 at 11:59 AM Lijie Wang > wrote: > > > > Hi Junrui, > > > > +1 for this proposal, thanks for driving. > > > > Best, > > Lijie > > > > ConradJam 于2023年9月22日周五 10:07写道： > > > > > +1 Support for standard YAML format facilitates specification > > > > > > Jing Ge 于2023年9月22日周五 02:23写道： > > > > > > > Hi Junrui, > > > > > > > > +1 for following the standard. Thanks for your effort! > > > > > > > > Best regards, > > > > Jing > > > > > > > > On Thu, Sep 21, 2023 at 5:09 AM Junrui Lee > wrote: > > > > > > > > > Hi Jane, > > > > > > > > > > Thank you for your valuable feedback and suggestions. > > > > > I agree with your point about differentiating between > > > "flink-config.yaml" > > > > > and "flink-conf.yaml" to determine the standard syntax at a glance. > > > > > > > > > > While I understand your suggestion of using > "flink-conf-default.yaml" > > > to > > > > > represent the default YAML file for Flink 1.x, I have been > considering > > > > > the option of using "flink-configuration.yaml" as the file name > for the > > > > > new configuration file. > > > > > This name "flink-configuration.yaml" provides a clear distinction > > > between > > > > > the new and old configuration files based on their names, and it > does > > > not > > > > > introduce any additional semantics. Moreover, this name > > > > > "flink-configuration.yaml" can continue to be used in future > versions > > > > > FLINK-2.0. > > > > > > > > > > WDYT? If we can reach a consensus on this, I will update the FLIP > > > > > documentation > > > > > accordingly. > > > > > > > > > > Best regards, > > > > > Junrui > > > > > > > > > > Jane Chan 于2023年9月20日周三 23:38写道： > > > > > > > > > > > Hi Junrui, > > > > > > > > > > > > Thanks for driving this FLIP. +1 for adoption of the standard > YAML > > > > > syntax. > > > > > > I just have one minor suggestion. It's a little bit challenging > to > > > > > > differentiate between `flink-config.yaml` and `flink-conf.yaml` > to > > > > > > determine which one uses the standard syntax at a glance. How > about > > > > > > using `flink-conf-default.yaml` to represent the default yaml > file > > > for > > > > > > Flink 1.x? > > > > > > > > > > > > Best, > > > > > > Jane > > > > > > > > > > > > On Wed, Sep 20, 2023 at 11:06 AM Junrui Lee > > > > > wrote: > > > > > > > > > > > > > Hi devs, > > > > > > > > > > > > > > I would like to start a discussion about FLIP-366: > > > > > > > Support standard YAML for FLINK configuration[1] > > > > > > > > > > > > > > The current flink-conf.yaml parser in FLINK is not a standard > YAML > > > > > > parser, > > > > > > > which has some shortcomings. > > > > > > > Firstly, it does not support nested structure configuration > items > > > and > > > > > > only > > > > > > > supports key-value pairs, resulting in poor readability. > Secondly, > > > if > > > > > the > > > > > > > value is a collection type, such as a List or Map, users are > > > required > > > > > to > > > > > > > write the value in a FLINK-specific pattern, which is > inconvenient > > > to > > > > > > use. > > > > > > > Additionally, the parser of FLINK has some differences in > syntax > > > > > compared > > > > > > > to the standard YAML parser, such as the syntax for parsing > > > comments > > > > > and > > > > > > > null values. These inconsistencies can cause confusion for > users, > > > as > > > > > seen > > > > > > > in FLINK-15358 and FLINK-32740. > > > > > > > > > > > > > > By supporting standard YAML, these issues can be resolved, and > > > users > > > > > can > > > > > > > create a Flink configuration file using third-party tools and > > > > leverage > > > > > > > some advanced YAML features. Therefore, we propose to support > > > > standard > > > > > > > YAML for FLINK configuration. > > > > > > > > > > > > > > You can find more details in the FLIP-366[1]. Looking forward > to > > > your > > > > > > > feedback. > > > > > > > > > > > > > > [1] > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-366%3A+Support+standard+YAML+for+FLINK+configuration > > > > > > > > > > > > > > Best, > > > > > > > Junrui > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > -- > > > Best > > > > > > ConradJam > > > >
Re: [Discuss] FLIP-362: Support minimum resource limitation
Hi, I agree that `minimum resource limitation` will bring values for flink session clusters, but for `Application Mode`, is it useful for streaming and batch jobs? Is it necessary for us to not support the application mode, rather than relying on the default value 0? Best, Shammon FY On Thu, Sep 21, 2023 at 10:18 AM Yangze Guo wrote: > Thanks for the comments, Jing. > > > Will the minimum resource configuration also take effect for streaming > jobs in application mode? > > Since it is not recommended to configure slotmanager.number-of-slots.max > for streaming jobs, does it make sense to disable it for common streaming > jobs? At least disable the check for avoiding the oscillation? > > Yes. The minimum resource configuration will only disabled in > standalone cluster atm. I agree it make sense to disable it for a pure > streaming job, however: > - By default, the minimum resource is configured to 0. If users do not > proactively set it, either the oscillation check or the minimum > restriction can be considered as disabled. > - The minimum resource is a cluster-level configuration rather than a > job-level configuration. If a user has an application with two batch > jobs preceding the streaming job, they may also require this > configuration to accelerate the execution of batch jobs. > > WDYT? > > Best, > Yangze Guo > > On Thu, Sep 21, 2023 at 4:49 AM Jing Ge > wrote: > > > > Hi Xiangyu, > > > > Thanks for driving it! There is one thing I am not really sure if I > > understand you correctly. > > > > According to the FLIP: "The minimum resource limitation will be > implemented > > in the DefaultResourceAllocationStrategy of FineGrainedSlotManager. > > > > Each time when SlotManager needs to reconcile the cluster resources or > > fulfill job resource requirements, the DefaultResourceAllocationStrategy > > will check if the minimum resource requirement has been fulfilled. If it > is > > not, DefaultResourceAllocationStrategy will request new > PendingTaskManagers > > and FineGrainedSlotManager will allocate new worker resources > accordingly." > > > > "To avoid this oscillation, we need to check the worker number derived > from > > minimum and maximum resource configuration is consistent before starting > > SlotManager." > > > > Will the minimum resource configuration also take effect for streaming > jobs > > in application mode? Since it is not recommended to > > configure slotmanager.number-of-slots.max for streaming jobs, does it > make > > sense to disable it for common streaming jobs? At least disable the check > > for avoiding the oscillation? > > > > Best regards, > > Jing > > > > > > On Tue, Sep 19, 2023 at 4:58 PM Chen Zhanghao > > > wrote: > > > > > Thanks for driving this, Xiangyu. We use Session clusters for quick SQL > > > debugging internally, and found cold-start job submission slow due to > lack > > > of the exact minimum resource reservation feature proposed here. This > > > should improve the experience a lot for running short lived-jobs in > session > > > clusters. > > > > > > Best, > > > Zhanghao Chen > > > > > > 发件人: Yangze Guo > > > 发送时间: 2023年9月19日 13:10 > > > 收件人: xiangyu feng > > > 抄送: dev@flink.apache.org > > > 主题: Re: [Discuss] FLIP-362: Support minimum resource limitation > > > > > > Thanks for driving this @Xiangyu. This is a feature that many users > > > have requested for a long time. +1 for the overall proposal. > > > > > > Best, > > > Yangze Guo > > > > > > On Tue, Sep 19, 2023 at 11:48 AM xiangyu feng > > > wrote: > > > > > > > > Hi Devs, > > > > > > > > I'm opening this thread to discuss FLIP-362: Support minimum resource > > > limitation. The design doc can be found at: > > > > FLIP-362: Support minimum resource limitation > > > > > > > > Currently, the Flink cluster only requests Task Managers (TMs) when > > > there is a resource requirement, and idle TMs are released after a > certain > > > period of time. However, in certain scenarios, such as running short > > > lived-jobs in session cluster and scheduling batch jobs stage by > stage, we > > > need to improve the efficiency of job execution by maintaining a > certain > > > number of available workers in the cluster all the time. > > > > > > > > After discussed with Yangze, we introduced this new feature. The new > > > added public options and proposed changes are described in this FLIP. > > > > > > > > Looking forward to your feedback, thanks. > > > > > > > > Best regards, > > > > Xiangyu > > > > > > > >
Re: [Discuss] CRD for flink sql gateway in the flink k8s operator
Thanks for all the valuable feedback, helm charts / templates sound good to me too. It will bring a lot of convenience to the production deployment of Sql-Gateway, looking forward to this, thanks Best, Shammon FY On Wed, Sep 20, 2023 at 10:01 AM Yangze Guo wrote: > Thanks for the reply, Gyula. Also thanks for the input from Thomas and > Dongwoo. > > Helm charts / templates for creating SQL Gateway deployment / services > sounds good to me. I'll work on it and also update that to the OLAP > quickstart doc. > > Best, > Yangze Guo > > On Tue, Sep 19, 2023 at 11:46 PM Dongwoo Kim > wrote: > > > > A simple Helm chart that covers both scenarios - running gateway > > 1) as a sidecar, 2) as an independent deployment- with solid docs sounds > > good to me. > > It would be nice if we could also include optional features in the chart > > such as k8s service for exposing the sql gateway. > > > > Best regards, > > Dongwoo > > > > 2023년 9월 19일 (화) 오후 10:15, Thomas Weise 님이 작성: > > > > > It is already possible to bring up a SQL Gateway as a sidecar > utilizing the > > > pod templates - I tend to also see this more of a documentation/example > > > issue rather than something that calls for a separate CRD or other > > > dedicated operator support. > > > > > > Thanks, > > > Thomas > > > > > > > > > > > > On Tue, Sep 19, 2023 at 3:41 PM Gyula Fóra > wrote: > > > > > > > Based on this I think we should start with simple Helm charts / > templates > > > > for creating the `FlinkDeployment` together with a separate > Deployment > > > for > > > > the SQL Gateway. > > > > If the gateway itself doesn't integrate well with the operator > managed > > > CRs > > > > (sessionjobs) then I think it's better and simpler to have it > separately. > > > > > > > > These Helm charts should be part of the operator repo / examples with > > > nice > > > > docs. If we see that it's useful and popular we can start thinking of > > > > integrating it into the CRD. > > > > > > > > What do you think? > > > > Gyula > > > > > > > > On Tue, Sep 19, 2023 at 6:09 AM Yangze Guo > wrote: > > > > > > > > > Thanks for the reply, @Gyula. > > > > > > > > > > I would like to first provide more context on OLAP scenarios. In > OLAP > > > > > scenarios, users typically submit multiple short batch jobs that > have > > > > > execution times typically measured in seconds or even sub-seconds. > > > > > Additionally, due to the lightweight nature of these jobs, they > often > > > > > do not require lifecycle management features and can disable high > > > > > availability functionalities such as failover and checkpointing. > > > > > > > > > > Regarding the integration issue, I believe that supporting the > > > > > generation of FlinkSessionJob through a gateway is a "nice to have" > > > > > feature rather than a "must-have." Firstly, it may be overkill to > > > > > create a CRD for such lightweight jobs, and it could potentially > > > > > impact the end-to-end execution time of the OLAP job. Secondly, as > > > > > mentioned earlier, these jobs do not have strong lifecycle > management > > > > > requirements, so having an operator manage them would be a bit > wasted. > > > > > Therefore, atm, we can allow users to directly submit jobs using > JDBC > > > > > or REST API. WDYT? > > > > > > > > > > Best, > > > > > Yangze Guo > > > > > > > > > > On Mon, Sep 18, 2023 at 4:08 PM Gyula Fóra > > > wrote: > > > > > > > > > > > > As I wrote in my previous answer, this could be done as a helm > chart > > > or > > > > > as > > > > > > part of the operator easily. Both would work. > > > > > > My main concern for adding this into the operator is that the SQL > > > > Gateway > > > > > > itself is not properly integrated with the Operator Custom > resources. > > > > > > > > > > > > Gyula > > > > > > > > > > > > On Mon, Sep 18, 2023 at 4:24 AM Shammon FY > > > wrote: > > > > > > > > > > > > > Thanks @Gyula, I wo
Re: [DISCUSS] FLIP-314: Support Customized Job Lineage Listener
Hi devs, After discussing with @Qingsheng, I fixed a minor issue of the lineage lifecycle in `StreamExecutionEnvironment`. I have added the comment to explain that the lineage information in `StreamExecutionEnvironment` will be consistent with that of transformations. When users clear the existing transformations, the added lineage information will also be deleted. Please help to review it again, and If there are no more concerns about FLIP -314[1], I would like to start voting later, thanks. cc @Leonard Best, Shammon FY On Mon, Jul 17, 2023 at 3:43 PM Shammon FY wrote: > Hi devs, > > Thanks for all the valuable feedback. If there are no more concerns about > FLIP-314[1], I would like to start voting later, thanks. > > > [1] > https://cwiki.apache.org/confluence/display/FLINK/FLIP-314%3A+Support+Customized+Job+Lineage+Listener > > Best, > Shammon FY > > > On Wed, Jul 12, 2023 at 11:18 AM Shammon FY wrote: > >> Thanks for the valuable feedback, Leonard. >> >> I have discussed with Leonard off-line. We have reached some conclusions >> about these issues and I have updated the FLIP as follows: >> >> 1. Simplify the `LineageEdge` interface by creating an edge from one >> source vertex to sink vertex. >> 2. Remove the `TableColumnSourceLineageVertex` interface and update >> `TableColumnLineageEdge` to create an edge from columns in one source to >> each sink column. >> 3. Rename `SupportsLineageVertex` to `LineageVertexProvider` >> 4. Add method `addLineageEdges(LineageEdge ... edges)` in >> `StreamExecutionEnviroment` for datastream job and remove previous methods >> in `DataStreamSource` and `DataStreamSink`. >> >> Looking forward to your feedback, thanks. >> >> Best, >> Shammon FY >> >
Re: [DISCUSS] FLIP-360: Merging ExecutionGraphInfoStore and JobResultStore into a single component
Hi Matthias, Thanks for initiating this discussion, and +1 for overall from my side. It's really strange to have two different places to store completed jobs, this also brings about the complexity of Flink. I agree with using a unified instance to store the completed job information. In terms of ability, `ExecutionGraphInfoStore` and `JobResultStore` are different: one is mainly used for information display, and the other is for failover. So after unifying storage, can we use different interfaces to meet the different requirements for jobs? Adding all these methods for different components into one interface such as `CompletedJobStore` may be a little strange. What do you think? Best, Shammon FY On Fri, Sep 8, 2023 at 8:08 PM Gyula Fóra wrote: > Hi Matthias! > > Thank you for the detailed proposal, overall I am in favor of making this > unification to simplify the logic and make the integration for external > components more straightforward. > I will try to read through the proposal more carefully next week and > provide some detailed feedback. > > +1 > > Thanks > Gyula > > On Fri, Sep 8, 2023 at 8:36 AM Matthias Pohl .invalid> > wrote: > > > Just a bit more elaboration on the question that we need to answer here: > Do > > we want to expose the internal ArchivedExecutionGraph data structure > > through JSON? > > > > - The JSON approach allows the user to have (almost) full access to the > > information (that would be otherwise derived from the REST API). > Therefore, > > there's no need to spin up a cluster to access this information. > > Any information that shall be exposed through the REST API needs to be > > well-defined in this JSON structure, though. Large parts of the > > ArchivedExecutionGraph data structure (essentially anything that shall be > > used to populate the REST API) become public domain, though, which puts > > more constraints on this data structure and makes it harder to change it > in > > the future. > > > > - The binary data approach allows us to keep the data structure itself > > internal. We have more control over what we want to expose by providing > > access points in the ClusterClient (e.g. just add a command to extract > the > > external storage path from the file). > > > > - The compromise (i.e. keeping ExecutionGraphInfoStore and JobResultStore > > separate and just expose the checkpoint information next to the JobResult > > in the JobResultStore file) would keep us the closest to the current > state, > > requires the least code changes and the least exposure of internal data > > structures. It would allow any system (like the Kubernetes Operator) to > > extract the checkpoint's external storage path. But we would still be > stuck > > with kind-of redundant components. > > > > From a user's perspective, I feel like the JSON approach is the best one > > because it gives him/her the most freedom to be independent of Flink > > binaries when handling completed jobs. But I see benefits from a Flink > > developer's perspective to not expose the entire data structure but use > the > > ClusterClient as an access point. > > > > The last option is my least favorite one: Moving the ExecutionGraphInfo > out > > of the JobManager seems to be the right thing to do when thinking about > > Flink's vision to become cloud-native. > > > > Just my 2cts on that topic. > > Matthias > > > > On Mon, Sep 4, 2023 at 1:11 PM Matthias Pohl > > wrote: > > > > > Hi everyone, > > > I want to open the discussion on FLIP-360 [1]. The goal of this FLIP is > > to > > > combine the two very similar components ExecutionGraphInfoStore and > > > JobResultStore into a single component. > > > > > > The benefit of this effort would be to expose the metadata of a > > > globally-terminated job even in cases where the JobManager fails > shortly > > > after the job finished. This is relevant for external checkpoint > > management > > > (like it's done in the Kubernetes Operator) which relies on the > > checkpoint > > > information to be available. > > > > > > More generally, it would allow completed jobs to be listed as part of > the > > > Flink cluster even after a JM failover. This would allow users to gain > > more > > > control over finished jobs. > > > > > > The current state of the FLIP doesn't come up with a final conclusion > on > > > the serialization format of the data (JSON vs binary). I want to > > emphasize > > > that there's also a third option which keeps both components separate > and > > > only exposes the additional checkpoint information through the > > > JobResultStore. > > > > > > I'm looking forward to feedback. > > > Best, > > > Matthias > > > > > > PS: I might be less responsive in the next 2-3 weeks but want to > initiate > > > the discussion, anyway. > > > > > > [1] > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-360%3A+Merging+the+ExecutionGraphInfoStore+and+the+JobResultStore+into+a+single+component+CompletedJobStore > > > > > >
Re: [Discuss] CRD for flink sql gateway in the flink k8s operator
Thanks @Gyula, I would like to share our use of sql-gateway with the Flink session cluster and I hope that it could help you to have a clearer understanding of our needs :) As @Yangze mentioned, currently we use flink as an olap platform by the following steps 1. Setup a flink session cluster by flink k8s session with k8s or zk highavailable. 2. Write a Helm chart for Sql-Gateway image and launch multiple gateway instances to submit jobs to the same flink session cluster. As we mentioned in docs[1], we hope that users can easily launch sql-gateway instances in k8s. Does it only need to add a Helm chart for sql-gateway, or should we need to add this feature to the flink operator? Can you help give the conclusion? Thank you very much @Gyula [1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/olap_quickstart/ Best, Shammon FY On Sun, Sep 17, 2023 at 2:02 PM Gyula Fóra wrote: > Hi! > It sounds pretty easy to deploy the gateway automatically with session > cluster deployments from the operator , but there is a major limitation > currently. The SQL gateway itself doesn't really support any operator > integration so jobs submitted through the SQL gateway would not be > manageable by the operator (they won't show up as session jobs). > > Without that, this is a very strange feature. We would make something much > easier for users that is not well supported by the operator in the first > place. The operator is designed to manage clusters and jobs > (FlinkDeployment / FlinkSessionJob). It would be good to understand if we > could make the SQL Gateway create a FlinkSessionJob / Deployment (that > would require application cluster support) and basically submit the job > through the operator. > > Cheers, > Gyula > > On Sun, Sep 17, 2023 at 1:26 AM Yangze Guo wrote: > > > > There would be many different ways of doing this. One gateway per > > session cluster, one gateway shared across different clusters... > > > > Currently, sql gateway cannot be shared across multiple clusters. > > > > > understand the tradeoff and the simplest way of accomplishing this. > > > > I'm not familiar with the Flink operator codebase, it would be > > appreciated if you could elaborate more on the cost of adding this > > feature. I agree that deploying a gateway using the native Kubernetes > > Deployment can be a simple way and straightforward for users. However, > > integrating it into an operator can provide additional benefits and be > > more user-friendly, especially for users who are less familiar with > > Kubernetes. By using an operator, users can benefit from consistent > > version management with the session cluster and upgrade capabilities. > > > > > > Best, > > Yangze Guo > > > > On Fri, Sep 15, 2023 at 5:38 PM Gyula Fóra wrote: > > > > > > There would be many different ways of doing this. One gateway per > session > > > cluster, one gateway shared across different clusters... > > > I would not rush to add anything anywhere until we understand the > > tradeoff > > > and the simplest way of accomplishing this. > > > > > > The operator already supports ingresses for session clusters so we > could > > > have a gateway sitting somewhere else simply using it. > > > > > > Gyula > > > > > > On Fri, Sep 15, 2023 at 10:18 AM Yangze Guo > wrote: > > > > > > > Thanks for bringing this up, Dongwoo. Flink SQL Gateway is also a key > > > > component for OLAP scenarios. > > > > > > > > @Gyula > > > > How about add sql gateway as an optional component to Session Cluster > > > > Deployments. User can specify the resource / instance number and > ports > > > > of the sql gateway. I think that would help a lot for OLAP and batch > > > > user. > > > > > > > > > > > > Best, > > > > Yangze Guo > > > > > > > > On Fri, Sep 15, 2023 at 3:19 PM ConradJam > wrote: > > > > > > > > > > If we start from the crd direction, I think this mode is more like > a > > > > > sidecar of the session cluster, which is submitted to the session > > cluster > > > > > by sending sql commands to the sql gateway. I don't know if my > > statement > > > > is > > > > > accurate. > > > > > > > > > > Xiaolong Wang 于2023年9月15日周五 > > > > 13:27写道： > > > > > > > > > > > Hi, Dongwoo, > > > > > > > > > > > > Since Flink SQL gateway should run u
Re: [Discuss] CRD for flink sql gateway in the flink k8s operator
Hi, Currently `sql-gateway` can be started with the script `sql-gateway.sh` in an existing node, it is more like a simple "standalone" node. I think it's valuable if we can do more work to start it in k8s. For xiaolong: Do you want to start a sql-gateway instance in the jobmanager pod? I think maybe we need a script like `kubernetes-sql-gatewah.sh` to start `sql-gateway` pods with a flink image, what do you think? Best, Shammon FY On Fri, Sep 15, 2023 at 10:02 AM Xiaolong Wang wrote: > Hi, I've experiment this feature on K8S recently, here is some of my trial: > > > 1. Create a new kubernetes-jobmanager.sh script with the following content > > #!/usr/bin/env bash > $FLINK_HOME/bin/sql-gateway.sh start > $FLINK_HOME/bin/kubernetes-jobmanager1.sh kubernetes-session > > 2. Build your own Flink docker image something like this > FROM flink:1.17.1-scala_2.12-java11 > > RUN mv $FLINK_HOME/bin/kubernetes-jobmanager.sh $FLINK_HOME/bin/ > kubernetes-jobmanager1.sh > COPY ./kubernetes-jobmanager.sh $FLINK_HOME/bin/kubernetes-jobmanager.sh > > RUN chmod +x $FLINK_HOME/bin/*.sh > USER flink > > 3. Create a Flink session job with the operator using the above image. > > On Thu, Sep 14, 2023 at 9:49 PM Gyula Fóra wrote: > > > Hi! > > > > I don't completely understand what would be a content of such CRD, could > > you give a minimal example how the Flink SQL Gateway CR yaml would look > > like? > > > > Adding a CRD would mean you need to add some operator/controller logic as > > well. Why not simply use a Deployment / StatefulSet in Kubernetes? > > > > Or a Helm chart if you want to make it more user friendly? > > > > Cheers, > > Gyula > > > > On Thu, Sep 14, 2023 at 12:57 PM Dongwoo Kim > > wrote: > > > > > Hi all, > > > > > > I've been working on setting up a flink SQL gateway in a k8s > environment > > > and it got me thinking — what if we had a CRD for this? > > > > > > So I have quick questions below. > > > 1. Is there ongoing work to create a CRD for the Flink SQL Gateway? > > > 2. If not, would the community be open to considering a CRD for this? > > > > > > I've noticed a growing demand for simplified setup of the flink sql > > gateway > > > in flink's slack channel. > > > Implementing a CRD could make deployments easier and offer better > > > integration with k8s. > > > > > > If this idea is accepted, I'm open to drafting a FLIP for further > > > discussion > > > > > > Thanks for your time and looking forward to your thoughts! > > > > > > Best regards, > > > Dongwoo > > > > > >
Re: [DISCUSS] Flink annotation strategy/consensus
Thanks Jing for starting this discussion. For @Becket > 1. All the public methods / classes / interfaces MUST be annotated with one of the @Experimental / @PublicEvolving / @Public. In practice, all the methods by default inherit the annotation from the containing class, unless annotated otherwise. e.g. an @Internal method in a @Public class. Do we really need to have `@Internal` methods in an `@Public` interface or class? In general, if a class or interface is marked as `@Public `, it is better that their public methods should also be `@Public`, because even if marked as `@Internal`, users are not aware of it when using it, which could be strange. @Jing Besides `@Internal`, I have some cents about `@PublicEvolving` and `@Public`. Currently when we add an interface which will be used by external systems, we often annotate it as `@PublicEvolving`. But when should this interface be marked as `@Public`? I find it is difficult to determine this. Is `@PublicEvolving` really necessary? Should we directly remove `@PublicEvolving` and use `@Public` instead? I think it would be simpler. Best, Shammon FY On Mon, Sep 11, 2023 at 11:05 AM Becket Qin wrote: > Hi Jing, > > Thanks for bringing up the discussion. My two cents: > > 1. All the public methods / classes / interfaces MUST be annotated with one > of the @Experimental / @PublicEvolving / @Public. In practice, all the > methods by default inherit the annotation from the containing class, unless > annotated otherwise. e.g. an @Internal method in a @Public class. > > 2. I agree it would be too verbose to annotate every internal method / > class / interface. Currently we already treat the methods / interfaces / > classes without annotations as effectively @Internal. > > 3. Per our discussion in the other thread, @Deprecated SHOULD coexist with > one of the @Experimental / @PublicEvolving / @Public. In that > case, @Deprecated overrides the other annotation, which means that public > API will not evolve and will be removed according to the deprecation > process. > > 4. The internal methods / classes / interfaces SHOULD NOT be marked as > deprecated. Instead, an immediate refactor should be done to remove the > "deprecated" internal methods / classes / interfaces, and migrate the code > to its successor. Otherwise, technical debts will build up. > > Thanks, > > Jiangjie (Becket) Qin > > > > On Sat, Sep 9, 2023 at 5:29 AM Jing Ge wrote: > > > Hi devs, > > > > While I was joining the flink-avro enhancement and cleanup discussion > > driven by Becket[1], I realized that there are some issues with the > current > > Flink API annotation usage in the source code. > > > > As far as I am concerned, Flink wants to control the access/visibility of > > APIs across modules and for downstreams. Since no OSGI is used(it should > > not be used because of its complexity, IMHO), Flink decided to use a very > > lightweight but manual solution: customized annotation like @Internal, > > @Experimental, @PublicEvolving, > > etc. This is a Flink only concept on top of JDK annotation and is > therefore > > orthogonal to @Deprecated or any other annotations offered by JDK. After > > this concept has been used, APIs without one of these annotations are in > > the kind of gray area which means they have no contract in the context of > > this new concept. Without any given metadata they could be considered > > as @Internal or @Experimental, because changes are allowed to be applied > at > > any time. But there is no clear definition and therefore different people > > will understand it differently. > > > > There are two options to improve it, as far as I could figure out: > > > > option 1: All APIs must have one of those annotations. We should put some > > effort into going through all source code and add missing annotations. > > There were discussions[2] and activities going in this direction. > > option 2: the community comes to a new consensus that APIs without > > annotation equals one of @Internal, @Experimental, or @PublicEvolving. I > > personally will choose @Internal, because it is the safest one. And if > > @Internal is chosen as the default one, it could also be deprecated, > > because no annotation equals @Internal. If it makes sense, I can create a > > FLIP and help the community reach this consensus. > > > > Both options have their own pros and cons. I would choose option 2, since > > we will not end up with a lot of APIs marked as @Internal. > > > > Looking forward to hearing your thoughts. > > > > Best regards > > Jing > > > > > > [1] https://lists.apache.org/thread/7zsv528swbjxo5zk0bxq33hrkvd77d6f > > [2] https://lists.apache.org/thread/zl2rmodsjsdb49tt4hn6wv3gdwo0m31o > > >
Re: [DISCUSS] FLIP-363: Unify the Representation of TaskManager Location in REST API and Web UI
Thanks Zhanghao for initialing this discussion, I have just one comment:
I checked the classes `SubtasksAllAccumulatorsHandler`,
`SubtasksTimesHandler`, `SubtaskCurrentAttemptDetailsHandler`,
`JobVertexTaskManagersHandler` and `JobExceptionsHandler` you mentioned in
`Public Interfaces` and they are not annotated as `Public`. So do you want
to annotate them as `Plublic`? If not, I think you may need to move them
from `Public Interfaces` to `Proposed Changes`.
Best,
Shammon FY
On Sat, Sep 9, 2023 at 12:11 PM Chen Zhanghao
wrote:
> Hi Devs,
>
> I would like to start a discussion on FLIP-363: Unify the Representation
> of TaskManager Location in REST API and Web UI [1].
>
> The TaskManager location of subtasks is important for identifying
> TM-related problems. There are a number of places in REST API and Web UI
> where TaskManager location is returned/displayed.
>
> Problems:
>
> * Only hostname is provided to represent TaskManager location in some
> places (e.g. SubtaskCurrentAttemptDetailsHandler). However, in a
> containerized era, it is common to have multiple TMs on the same host, and
> port info is crucial to distinguish different TMs.
> * Inconsistent naming of the field to represent TaskManager location:
> "host" is used in most places but "location" is also used in
> JobExceptions-related places.
> * Inconsistent semantics of the "host" field: The semantics of the
> host field are inconsistent, sometimes it denotes hostname only while in
> other times it denotes hostname + port (which is also inconsistent with the
> name of "host").
>
> We propose to improve the current situation by:
>
> * Use a field named "location" that represents TaskManager location in
> the form of "${hostname}:${port}" in a consistent manner across REST APIs
> and the front-end.
> * Rename the column name from "Host" to "Location" on the Web UI to
> reflect the change that both hostname and port are displayed.
> * Keep the old "host" fields untouched for compatibility. They can be
> removed in the next major version.
>
> Looking forward to your feedback.
>
> [1] FLIP-363: Unify the Representation of TaskManager Location in REST API
> and Web UI - Apache Flink - Apache Software Foundation<
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-363%3A+Unify+the+Representation+of+TaskManager+Location+in+REST+API+and+Web+UI
> >
>
> Best,
> Zhanghao Chen
>
Re: [DISCUSS] Update Flink Roadmap
Hi Jark, Sounds good and I would love to, thanks! I will involve you and Xingtong on the document after updating. Best, Shammon FY On Mon, Aug 14, 2023 at 10:39 PM Jark Wu wrote: > Hi Shammon, > > Sure, could you help to draft a subsection about this in the google doc? > > Best, > Jark > > 2023年8月14日 20:30，Shammon FY 写道： > > Thanks @Jark for driving the Flink Roadmap. > > As we discussed olap in the thread [1] and according to the suggestions > from @Xingtong Song, could we add a subsection in `Towards Streaming > Warehouses` or `Performance` that the short-lived query in Flink Session > Cluster is one of the future directions for Flink? > > Best, > Shammon FY > > On Mon, Aug 14, 2023 at 8:03 PM Jark Wu wrote: > >> Thank you everyone for helping polish the roadmap [1]. >> >> I think I have addressed all the comments and we have included all ongoing >> parts of Flink. >> Please feel free to take a last look. I'm going to prepare the pull >> request >> if there are no more concerns. >> >> Best, >> Jark >> >> [1]: >> >> https://docs.google.com/document/d/12BDiVKEsY-f7HI3suO_IxwzCmR04QcVqLarXgyJAb7c/edit >> >> On Sun, 13 Aug 2023 at 13:04, Yuan Mei wrote: >> >> > Sorry for taking so long >> > >> > I've added a section about Flink Disaggregated State Management >> Evolution >> > in the attached doc. >> > >> > I found some of the contents might be overlapped with the "large-scale >> > streaming jobs". So that part might need some changes as well. >> > >> > Please let me know what you think. >> > >> > Best >> > Yuan >> > >> > On Mon, Jul 24, 2023 at 12:07 PM Yuan Mei >> wrote: >> > >> > > Sorry have missed this email and respond a bit late. >> > > >> > > I will put a draft for the long-term vision for the state as well as >> > > large-scale state support into the roadmap. >> > > >> > > Best >> > > Yuan >> > > >> > > On Mon, Jul 17, 2023 at 10:34 AM Jark Wu wrote: >> > > >> > >> Hi Jiabao, >> > >> >> > >> Thank you for your suggestions. I have added them to the "Going >> Beyond a >> > >> SQL Stream/Batch Processing Engine" and "Large-Scale State Jobs" >> > sections. >> > >> >> > >> Best, >> > >> Jark >> > >> >> > >> On Thu, 13 Jul 2023 at 16:06, Jiabao Sun > > >> .invalid> >> > >> wrote: >> > >> >> > >> > Thanks Jark and Martijn for driving this. >> > >> > >> > >> > There are two suggestions about the Table API: >> > >> > >> > >> > - Add the JSON type to adapt to the no sql database type. >> > >> > - Remove changelog normalize operator for upsert stream. >> > >> > >> > >> > >> > >> > Best, >> > >> > Jiabao >> > >> > >> > >> > >> > >> > > 2023年7月13日 下午3:49，Jark Wu 写道： >> > >> > > >> > >> > > Hi all, >> > >> > > >> > >> > > Sorry for taking so long back here. >> > >> > > >> > >> > > Martijn and I have drafted the first version of the updated >> roadmap, >> > >> > > including the updated feature radar reflecting the current state >> of >> > >> > > different components. >> > >> > > >> > >> > >> > >> >> > >> https://docs.google.com/document/d/12BDiVKEsY-f7HI3suO_IxwzCmR04QcVqLarXgyJAb7c/edit >> > >> > > >> > >> > > Feel free to leave comments in the thread or the document. >> > >> > > We may miss mentioning something important, so your help in >> > enriching >> > >> > > the content is greatly appreciated. >> > >> > > >> > >> > > Best, >> > >> > > Jark & Martijn >> > >> > > >> > >> > > >> > >> > > On Fri, 2 Jun 2023 at 00:50, Jing Ge > > >> > >> wrote: >> > >> > > >> > >> > >> Hi Jark, >> > >> > >> >> > >> > >> Fair enough. Let's do it like you suggested.
Re: [DISCUSS] How about adding OLAP to Flink Roadmap?
Hi, Thanks for all the feedback. I'm so glad that I can see some consensus we have reached from the feedback. I am trying to summarize our consensus as follows and please correct me if I'm wrong or misunderstanding. 1) Batch is a special case of Streaming, while olap is a special case of batch, so Flink will not lose focus from supporting short-lived olap jobs. 2) As a streaming and batch processing engine, it's valuable for Flink to support olap jobs which will bring big merit to users and worth our efforts to promote and achieve. 3) From the evolution of unified streaming-batch-olap processing engine for Flink, we could add a subsection for Flink olap in the roadmap and continue to evolve. 4) In order to support short live queries in Flink, it is necessary to do some very careful design in terms of architecture and implementation. These designs cannot affect streaming and batch capabilities in Flink while supporting olap. 5) In order to better guide and measure the optimization of Flink olap, we need to add relevant olap benchmarks to the flink project repository such as flink-benchmarks. As I replied in the roadmap thread [1], @Jark Wu and @Xingtong Song could you please help to add the flink olap related subsection to the doc [2] ? Thanks very much! [1] https://lists.apache.org/thread/szdr4ngrfcmo7zko4917393zbqhgw0v5 [2] https://docs.google.com/document/d/12BDiVKEsY-f7HI3suO_IxwzCmR04QcVqLarXgyJAb7c/edit Best, Shammon FY On Mon, Aug 14, 2023 at 4:08 PM Yun Tang wrote: > Thanks to the guys from ByteDance driving this topic, which could be > another big story to extend Flink's ability. > > In general, I think this is a great idea. However, before we move forward, > I think we should first answer the question: which is the target for Flink > OLAP? > > We run Presto/Trino and SparkSQL in the production environment for OLAP > SQL analysis. Since Presto runs faster than SparkSQL in many cases, > especially for ad-hoc queries at medium-sized data, we would run queries on > Presto first or switch to SparkSQL for large-scale queries if necessary. > Presto runs as a service and emphasis on query performance without node > fault tolerance. Moreover, it leverages a pipeline-like data exchange mode > instead of the classic stage blocking exchange mode, which is a bit like > Flink's pipeline mode vs blocking mode. > > Can we say we hope Flink OLAP could target Presto/Trino in medium-sized > data query, and switch to Flink batch SQL for large-scale analysis query? > If so, I also think the naming of Flink OLAP looks a bit strange, as Flink > batch SQL shall also serve for large-scale OLAP analysis. > > Best > Yun Tang > > From: Jing Ge > Sent: Thursday, August 10, 2023 13:52 > To: dev@flink.apache.org > Subject: Re: [DISCUSS] How about adding OLAP to Flink Roadmap? > > Hi Shammon, Hi Xiangyu, > > Thanks for bringing this to our attention. I can see this is a great > proposal born from real business scenarios. +1 for it. > > People have been keen to use one platform to cover all their data > production and consumption requirements. Flink did a great job for the > production, i.e. streaming and batch processing with all excellent > ecosystems. This is the big advantage for Flink to go one step further and > cover the consumption part. It will turn Flink into a unified compute > platform like what the Ray project(the platform behind ChatGPT, if someone > is not aware of it)[1] is doing and secure Flink to be one of the most > interesting open source platforms for the next decade. > > Frankly speaking, it will be a big change. As far as I am concerned, the > following should be considered(just thought about it at the first glance, > there must be more). > > Architecture upgrade - since we will have three capabilities(I wanted to > use "engines", but it might be too early to use the big word), i.e. > streaming, batch, OLAP, it might make sense to upgrade the architecture > while we are building the OLAP in Flink. The unified foundation or > abstraction for distributed computation should be designed and implemented > underneath those capabilities. In the future, new capabilities can leverage > the foundation and could be developed at a very fast pace. > > MPP architecture - Flink session cluster is not the MMP architecture. > Commonly speaking, SNA(shared nothing architecture) is the key that could > implement MPP. Flink has everything to offer SNA. That is the reason why we > can consider building OLAP into or on top of the Flink. And speaking of > MPP, there will be a lot of things to do, e.g. the Retrieval > Architecture[2], multiple level task split, dynamic retry or even split, > etc. I will not expand all those topics at this early stage. > > OLAP queries syntax -
Re: [DISCUSS] Update Flink Roadmap
Thanks @Jark for driving the Flink Roadmap. As we discussed olap in the thread [1] and according to the suggestions from @Xingtong Song, could we add a subsection in `Towards Streaming Warehouses` or `Performance` that the short-lived query in Flink Session Cluster is one of the future directions for Flink? Best, Shammon FY On Mon, Aug 14, 2023 at 8:03 PM Jark Wu wrote: > Thank you everyone for helping polish the roadmap [1]. > > I think I have addressed all the comments and we have included all ongoing > parts of Flink. > Please feel free to take a last look. I'm going to prepare the pull request > if there are no more concerns. > > Best, > Jark > > [1]: > > https://docs.google.com/document/d/12BDiVKEsY-f7HI3suO_IxwzCmR04QcVqLarXgyJAb7c/edit > > On Sun, 13 Aug 2023 at 13:04, Yuan Mei wrote: > > > Sorry for taking so long > > > > I've added a section about Flink Disaggregated State Management Evolution > > in the attached doc. > > > > I found some of the contents might be overlapped with the "large-scale > > streaming jobs". So that part might need some changes as well. > > > > Please let me know what you think. > > > > Best > > Yuan > > > > On Mon, Jul 24, 2023 at 12:07 PM Yuan Mei > wrote: > > > > > Sorry have missed this email and respond a bit late. > > > > > > I will put a draft for the long-term vision for the state as well as > > > large-scale state support into the roadmap. > > > > > > Best > > > Yuan > > > > > > On Mon, Jul 17, 2023 at 10:34 AM Jark Wu wrote: > > > > > >> Hi Jiabao, > > >> > > >> Thank you for your suggestions. I have added them to the "Going > Beyond a > > >> SQL Stream/Batch Processing Engine" and "Large-Scale State Jobs" > > sections. > > >> > > >> Best, > > >> Jark > > >> > > >> On Thu, 13 Jul 2023 at 16:06, Jiabao Sun > >> .invalid> > > >> wrote: > > >> > > >> > Thanks Jark and Martijn for driving this. > > >> > > > >> > There are two suggestions about the Table API: > > >> > > > >> > - Add the JSON type to adapt to the no sql database type. > > >> > - Remove changelog normalize operator for upsert stream. > > >> > > > >> > > > >> > Best, > > >> > Jiabao > > >> > > > >> > > > >> > > 2023年7月13日 下午3:49，Jark Wu 写道： > > >> > > > > >> > > Hi all, > > >> > > > > >> > > Sorry for taking so long back here. > > >> > > > > >> > > Martijn and I have drafted the first version of the updated > roadmap, > > >> > > including the updated feature radar reflecting the current state > of > > >> > > different components. > > >> > > > > >> > > > >> > > > https://docs.google.com/document/d/12BDiVKEsY-f7HI3suO_IxwzCmR04QcVqLarXgyJAb7c/edit > > >> > > > > >> > > Feel free to leave comments in the thread or the document. > > >> > > We may miss mentioning something important, so your help in > > enriching > > >> > > the content is greatly appreciated. > > >> > > > > >> > > Best, > > >> > > Jark & Martijn > > >> > > > > >> > > > > >> > > On Fri, 2 Jun 2023 at 00:50, Jing Ge > > >> wrote: > > >> > > > > >> > >> Hi Jark, > > >> > >> > > >> > >> Fair enough. Let's do it like you suggested. Thanks! > > >> > >> > > >> > >> Best regards, > > >> > >> Jing > > >> > >> > > >> > >> On Thu, Jun 1, 2023 at 6:00 PM Jark Wu wrote: > > >> > >> > > >> > >>> Hi Jing, > > >> > >>> > > >> > >>> This thread is for discussing the roadmap for versions 1.18, > 2.0, > > >> and > > >> > >> even > > >> > >>> more. > > >> > >>> One of the outcomes of this discussion will be an updated > version > > of > > >> > the > > >> > >>> current roadmap. > > >> > >>> Let's work together on refini
[DISCUSS] How about adding OLAP to Flink Roadmap?
, but we also provide Flink OLAP clusters in production for our users. After our improvements, we built a 500 core Flink OLAP cluster to test Flink OLAP e2e latency and QPS. There will be nearly a thousand QPS for simple queries of 128 subtasks with tens of milliseconds latency; for complex jobs with two JOIN operators and 700 subtasks, there will be about sixty QPS with hundreds of milliseconds latency. Our Flink OLAP in production supports more than ten internal businesses and there are more than 6000 cores in all clusters and 500 thousand queries will be performed each day. 5. My Proposal Above all, I personally think it will provide great value for users and is technically feasible if Flink supports OLAP and this is also what Flink engine can achieve through some improvements although this is not a simple matter. I hope the community could consider supporting OLAP as one of Flink technical directions and add it to the roadmap. As I mentioned at the beginning, this can attract more developers and users to participate and promote the improvement of Flink OLAP and build Flink as a unified engine for streaming, batch and OLAP. Looking forward to your feedback, thanks! [1] https://issues.apache.org/jira/browse/FLINK-32667 [2] https://issues.apache.org/jira/browse/FLINK-25318 Best, Shammon FY
Re: [ANNOUNCE] New Apache Flink Committer - Weihua Hu
Congratulations, Weihua! Best, Shammon FY On Fri, Aug 4, 2023 at 12:33 PM Jing Ge wrote: > congrats! Weihua! > > Best regards, > Jing > > On Fri, Aug 4, 2023 at 12:20 PM Matt Wang wrote: > > > Congratulations Weihua ~ > > > > > > -- > > > > Best, > > Matt Wang > > > > > > Replied Message > > | From | Rui Fan<1996fan...@gmail.com> | > > | Date | 08/4/2023 11:29 | > > | To | | > > | Cc | Weihua Hu | > > | Subject | Re: [ANNOUNCE] New Apache Flink Committer - Weihua Hu | > > Congratulations Weihua, well deserved! > > > > Best, > > Rui Fan > > > > On Fri, Aug 4, 2023 at 11:19 AM Xintong Song > > wrote: > > > > Hi everyone, > > > > On behalf of the PMC, I'm very happy to announce Weihua Hu as a new Flink > > Committer! > > > > Weihua has been consistently contributing to the project since May 2022. > He > > mainly works in Flink's distributed coordination areas. He is the main > > contributor of FLIP-298 and many other improvements in large-scale job > > scheduling and improvements. He is also quite active in mailing lists, > > participating discussions and answering user questions. > > > > Please join me in congratulating Weihua! > > > > Best, > > > > Xintong (on behalf of the Apache Flink PMC) > > > > >
Re: [ANNOUNCE] New Apache Flink Committer - Hong Teoh
Congratulations, Hong! Best, Shammon FY On Fri, Aug 4, 2023 at 12:33 PM Jing Ge wrote: > congrats! Hong! > > Best regards, > Jing > > On Fri, Aug 4, 2023 at 11:48 AM Qingsheng Ren wrote: > > > Congratulations and welcome aboard, Hong! > > > > Best, > > Qingsheng > > > > On Fri, Aug 4, 2023 at 11:04 AM Matt Wang wrote: > > > > > Congratulations, Hong! > > > > > > > > > -- > > > > > > Best, > > > Matt Wang > > > > > > > > > Replied Message > > > | From | Weihua Hu | > > > | Date | 08/4/2023 10:55 | > > > | To | | > > > | Subject | Re: [ANNOUNCE] New Apache Flink Committer - Hong Teoh | > > > Congratulations, Hong! > > > > > > Best, > > > Weihua > > > > > > > > > On Fri, Aug 4, 2023 at 10:49 AM Samrat Deb > > wrote: > > > > > > Congratulations, Hong Teoh > > > > > > On Fri, 4 Aug 2023 at 7:52 AM, Benchao Li > wrote: > > > > > > Congratulations, Hong! > > > > > > yuxia 于2023年8月4日周五 09:23写道： > > > > > > Congratulations, Hong Teoh! > > > > > > Best regards, > > > Yuxia > > > > > > - 原始邮件 - > > > 发件人: "Matthias Pohl" > > > 收件人: "dev" > > > 发送时间: 星期四, 2023年 8 月 03日 下午 11:24:39 > > > 主题: Re: [ANNOUNCE] New Apache Flink Committer - Hong Teoh > > > > > > Congratulations, Hong! :) > > > > > > On Thu, Aug 3, 2023 at 3:39 PM Leonard Xu wrote: > > > > > > Congratulations, Hong! > > > > > > > > > Best, > > > Leonard > > > > > > On Aug 3, 2023, at 8:45 PM, Jiabao Sun > > .INVALID> > > > wrote: > > > > > > Congratulations, Hong Teoh! > > > > > > Best, > > > Jiabao Sun > > > > > > 2023年8月3日 下午7:32，Danny Cranmer 写道： > > > > > > On behalf of the PMC, I'm very happy to announce Hong Teoh as a > > > new > > > Flink > > > Committer. > > > > > > Hong has been active in the Flink community for over 1 year and > > > has > > > played > > > a key role in developing and maintaining AWS integrations, core > > > connector > > > APIs and more recently, improvements to the Flink REST API. > > > Additionally, > > > Hong is a very active community member, supporting users and > > > participating > > > in discussions on the mailing lists, Flink slack channels and > > > speaking > > > at > > > conferences. > > > > > > Please join me in congratulating Hong for becoming a Flink > > > Committer! > > > > > > Thanks, > > > Danny Cranmer (on behalf of the Flink PMC) > > > > > > > > > > > > > > > > > > > > > -- > > > > > > Best, > > > Benchao Li > > > > > > > > > > > >
Re: [ANNOUNCE] New Apache Flink PMC Member - Matthias Pohl
Congratulations, Matthias! On Fri, Aug 4, 2023 at 1:13 PM Samrat Deb wrote: > Congrats, Matthias! > > > On Fri, 4 Aug 2023 at 10:13 AM, Benchao Li wrote: > > > Congratulations, Matthias! > > > > Jing Ge 于2023年8月4日周五 12:35写道： > > > > > Congrats! Matthias! > > > > > > Best regards, > > > Jing > > > > > > On Fri, Aug 4, 2023 at 12:09 PM Yangze Guo wrote: > > > > > > > Congrats, Matthias! > > > > > > > > Best, > > > > Yangze Guo > > > > > > > > On Fri, Aug 4, 2023 at 11:44 AM Qingsheng Ren > > wrote: > > > > > > > > > > Congratulations, Matthias! This is absolutely well deserved. > > > > > > > > > > Best, > > > > > Qingsheng > > > > > > > > > > On Fri, Aug 4, 2023 at 11:31 AM Rui Fan <1996fan...@gmail.com> > > wrote: > > > > > > > > > > > Congratulations Matthias, well deserved! > > > > > > > > > > > > Best, > > > > > > Rui Fan > > > > > > > > > > > > On Fri, Aug 4, 2023 at 11:30 AM Leonard Xu > > > wrote: > > > > > > > > > > > > > Congratulations, Matthias. > > > > > > > > > > > > > > Well deserved ^_^ > > > > > > > > > > > > > > Best, > > > > > > > Leonard > > > > > > > > > > > > > > > > > > > > > > On Aug 4, 2023, at 11:18 AM, Xintong Song < > > tonysong...@gmail.com > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > Hi everyone, > > > > > > > > > > > > > > > > On behalf of the PMC, I'm very happy to announce that > Matthias > > > > Pohl has > > > > > > > > joined the Flink PMC! > > > > > > > > > > > > > > > > Matthias has been consistently contributing to the project > > since > > > > Sep > > > > > > > 2020, > > > > > > > > and became a committer in Dec 2021. He mainly works in > Flink's > > > > > > > distributed > > > > > > > > coordination and high availability areas. He has worked on > many > > > > FLIPs > > > > > > > > including FLIP195/270/285. He helped a lot with the release > > > > management, > > > > > > > > being one of the Flink 1.17 release managers and also very > > active > > > > in > > > > > > > Flink > > > > > > > > 1.18 / 2.0 efforts. He also contributed a lot to improving > the > > > > build > > > > > > > > stability. > > > > > > > > > > > > > > > > Please join me in congratulating Matthias! > > > > > > > > > > > > > > > > Best, > > > > > > > > > > > > > > > > Xintong (on behalf of the Apache Flink PMC) > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > -- > > > > Best, > > Benchao Li > > >
Re: [DISCUSS] FLIP-314: Support Customized Job Lineage Listener
Hi devs, Thanks for all the valuable feedback. If there are no more concerns about FLIP-314[1], I would like to start voting later, thanks. [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-314%3A+Support+Customized+Job+Lineage+Listener Best, Shammon FY On Wed, Jul 12, 2023 at 11:18 AM Shammon FY wrote: > Thanks for the valuable feedback, Leonard. > > I have discussed with Leonard off-line. We have reached some conclusions > about these issues and I have updated the FLIP as follows: > > 1. Simplify the `LineageEdge` interface by creating an edge from one > source vertex to sink vertex. > 2. Remove the `TableColumnSourceLineageVertex` interface and update > `TableColumnLineageEdge` to create an edge from columns in one source to > each sink column. > 3. Rename `SupportsLineageVertex` to `LineageVertexProvider` > 4. Add method `addLineageEdges(LineageEdge ... edges)` in > `StreamExecutionEnviroment` for datastream job and remove previous methods > in `DataStreamSource` and `DataStreamSink`. > > Looking forward to your feedback, thanks. > > Best, > Shammon FY >
Re: [DISCUSS] FLIP-314: Support Customized Job Lineage Listener
Thanks for the valuable feedback, Leonard. I have discussed with Leonard off-line. We have reached some conclusions about these issues and I have updated the FLIP as follows: 1. Simplify the `LineageEdge` interface by creating an edge from one source vertex to sink vertex. 2. Remove the `TableColumnSourceLineageVertex` interface and update `TableColumnLineageEdge` to create an edge from columns in one source to each sink column. 3. Rename `SupportsLineageVertex` to `LineageVertexProvider` 4. Add method `addLineageEdges(LineageEdge ... edges)` in `StreamExecutionEnviroment` for datastream job and remove previous methods in `DataStreamSource` and `DataStreamSink`. Looking forward to your feedback, thanks. Best, Shammon FY
Re: [ANNOUNCE] Flink 1.18 Feature Freeze Extended until July 24th, 2023
Thanks for driving the release and sharing the update, looking forward to 1.18 Best, Shammon FY On Fri, Jul 7, 2023 at 10:56 AM Yun Tang wrote: > Thanks for driving this release and sharing the update on the feature > freeze extension. > > > Best > Yun Tang > > From: Jing Ge > Sent: Thursday, July 6, 2023 17:13 > To: yuxia > Cc: dev ; re...@apache.org ; > snuyan...@gmail.com ; Konstantin Knauf < > kna...@apache.org> > Subject: Re: [ANNOUNCE] Flink 1.18 Feature Freeze Extended until July > 24th, 2023 > > Thanks for driving it and sharing the update! > > Best regards, > Jing > > On Thu, Jul 6, 2023 at 9:21 AM yuxia wrote: > > > Thanks for the update and thanks for your efforts. > > > > Best regards, > > Yuxia > > > > - 原始邮件 - > > 发件人: "Rui Fan" <1996fan...@gmail.com> > > 收件人: "dev" , re...@apache.org > > 抄送: "Jing Ge" , snuyan...@gmail.com, "Konstantin > > Knauf" > > 发送时间: 星期四, 2023年 7 月 06日 下午 3:06:28 > > 主题: Re: [ANNOUNCE] Flink 1.18 Feature Freeze Extended until July 24th, > 2023 > > > > Thanks for the update, and thank you for your efforts for the 1.18 > release! > > > > Best, > > Rui Fan > > > > On Thu, Jul 6, 2023 at 2:40 PM Qingsheng Ren wrote: > > > > > Hi devs, > > > > > > Recently we collected some feedback from developers, and in order to > give > > > more time for polishing some important features in 1.18, we decide to > > > extend the feature freezing date to: > > > > > > July 24th, 2023, at 00:00 CEST(UTC+2) > > > > > > which gives us ~2 weeks for development from now. There will be no > > > extension after Jul 24, so please arrange new features in the next > > release > > > if they cannot be finished before the closing date. > > > > > > Thanks everyone for your work in 1.18! > > > > > > Best regards, > > > Qingsheng, Jing, Konstantin and Sergey > > > > > >
Re: [DISCUSS] FLIP-314: Support Customized Job Lineage Listener
Thanks Jing, sounds good to me. I have updated the FLIP and renamed the lineage related classes to `LineageGraph`, `LineageVertex` and `LineageEdge` and keep it consistent with the job definition in Flink. Best, Shammon FY On Thu, Jul 6, 2023 at 8:25 PM Jing Ge wrote: > Hi Shammon, > > Thanks for the clarification. Atlas might have his historical reason back > to the hadoop era or maybe even back to the hibernate where Entity and > Relation were commonly used. Flink already used Vertex and Edge to describe > DAG. Some popular tools like dbt are also using this convention[1] and, > afaik, most graph frameworks use vertex and edge too. It will be easier for > Flink devs and users to have a consistent naming convention for the same > concept, i.e. in this case, DAG. > > Best regards, > Jing > > [1] > > https://docs.getdbt.com/docs/dbt-cloud-apis/discovery-use-cases-and-examples#discovery > > On Wed, Jul 5, 2023 at 11:28 AM Shammon FY wrote: > > > Hi Jing, > > > > Thanks for your feedback. > > > > > 1. TableColumnLineageRelation#sinkColumn() should return > > TableColumnLineageEntity instead of String, right? > > > > The `sinkColumn()` will return `String` which is the column name in the > > sink connector. I found the name of `TableColumnLineageEntity` may > > cause ambiguity and I have renamed it to > `TableColumnSourceLineageEntity`. > > In my mind the `TableColumnLineageRelation` represents the lineage for > each > > sink column, each column may be computed from multiple sources and > columns. > > I use `TableColumnSourceLineageEntity` to manage each source and its > > columns for the sink column, so `TableColumnLineageRelation` has a sink > > column name and `TableColumnSourceLineageEntity` list. > > > > > 2. Since LineageRelation already contains all information to build the > > lineage between sources and sink, do we still need to set the > LineageEntity > > in the source? > > > > The lineage interface of `DataStream` is very flexible. We have added > > `setLineageEntity` to the source to limit and verify user behavior, > > ensuring that users have not added non-existent sources as lineage. > > > > > 3. About the "Entity" and "Relation" naming, I was confused too, like > > Qingsheng mentioned. How about LineageVertex, LineageEdge, and > LineageEdges > > which contains multiple LineageEdge? > > > > We referred to `Atlas` for the name of lineage, it uses `Entity` and > > `Relation` to represent the lineage relationship and another metadata > > service `Datahub` uses `DataSet` to represent the entity. I think > `Entity` > > and `Relation` are nicer for lineage, what do you think of it? > > > > Best, > > Shammon FY > > > > > > On Thu, Jun 29, 2023 at 4:21 AM Jing Ge > > wrote: > > > > > Hi Shammon, > > > > > > Thanks for your proposal. After reading the FLIP, I'd like to ask > > > some questions to make sure we are on the same page. Thanks! > > > > > > 1. TableColumnLineageRelation#sinkColumn() should return > > > TableColumnLineageEntity instead of String, right? > > > > > > 2. Since LineageRelation already contains all information to build the > > > lineage between sources and sink, do we still need to set the > > LineageEntity > > > in the source? > > > > > > 3. About the "Entity" and "Relation" naming, I was confused too, like > > > Qingsheng mentioned. How about LineageVertex, LineageEdge, and > > LineageEdges > > > which contains multiple LineageEdge? E.g. multiple sources join into > one > > > sink, or, edges of columns from one or different tables, etc. > > > > > > Best regards, > > > Jing > > > > > > On Sun, Jun 25, 2023 at 2:06 PM Shammon FY wrote: > > > > > > > Hi yuxia and Yun, > > > > > > > > Thanks for your input. > > > > > > > > For yuxia: > > > > > 1: What kinds of JobStatus will the `JobExecutionStatusEven` > > including? > > > > > > > > At present, we only need to notify the listener when a job goes to > > > > termination, but I think it makes sense to add generic `oldStatus` > and > > > > `newStatus` in the listener and users can update the job state in > their > > > > service as needed. > > > > > > > > > 2: I'm really confused about the `config()` included in > > > `LineageEntity`, > > > > where is it from and what
Re: [DISCUSS] FLIP-325: Support configuring end-to-end allowed latency
Hi, Thanks for your replay @Dong. I really agree with Piotr's points and I would like to share some thoughts from my side. About the latency for mini-batch mechanism in Flink SQL, I still think the description in the FLIP is not right. If there are N operators and the whole process time for data in the job is `t`, then the latency in mini-batch will be `table.exec.mini-batch.allow-latency`+`t`, not ` table.exec.mini-batch.allow-latency`*N. I think this is one of the foundations of this FLIP, and you may need to confirm it again. I think supporting similar mechanisms in the runtime and balance latency and throughput dynamically for all flink jobs is a very good idea, and I have some questions for that. 1. We encounter a situation where the workload is high when processing snapshot data and we need mini-batch in sql for performance reason. But the workload is low when processing delta data, we need to automatically adjust the mini-batch SQL for them, or even cancel the mini-batch during delta processing. I think this FLIP meets our needs, but I think we need a general solution which covers all source types in flink, and the `isBacklog` in the FLIP is only one strategy. >From the FLIP I think there should be two parts: dynamic trigger flush event in JM and dynamic trigger flush operations in Operator. We need to introduce much more general interfaces for them, such as `DynamicFlushStrategy` in JM and `DynamicFlushOperation` in TM? As Piotr mentioned above, we can collect many information from TM locally such as backpressure, queue size and `Operator` can decide whether to buffer data or process it immediately. JM is also the same, it can decide to send flush events on a regular basis or send them based on the collected metrics information and other information, such as the isBacklog in the FLIP. 2. I really don't get enough benefits for `RecordAttribute` in the FLIP and as Piotr mentioned above too, it will generate a large number of messages, affecting performance. FLIP mentions that it will be applied to Operator and Sink, I try to understand it's role and please correct me if I'm wrong. a) It tells the Operator and Sink that current most of data they are processing are from snapshot and are "insert" data? For the out of order in flink, the Operator and Sink may receive "upsert" data from other sources. b) Do Operators and Sink perform any very special operations in the above situations? What are the benefits of this special operations for "most data are insert"? c) I think the operator and sink can collect the above information locally when it receives each record without the need for `RecordAttribute` even when we need some special operations. d) Even if we do need a `RecordAttribute` events in Operator and Sink, I think broadcast them from JM is a better choice. 3. For the flush event, I also have some questions. What type of operators need to buffer data and flush them based on the flush events? In SQL mini-batch mechanism, similar processing will be added for the aggregate and join operators, while for operators such as map, it is not necessary. How can we identify different operator in the runtime layer (`Input` and `TwoInputStreamOperator`)? I think buffer data in Map/FlatMap/Filter Operator is not a good idea which makes data no longer flowing. Best, Shammon FY On Thu, Jul 6, 2023 at 1:54 AM Piotr Nowojski wrote: > Hi, > > Thanks for this proposal, this is a very much needed thing that should be > addressed in Flink. > > I think there is one thing that hasn't been discussed neither here nor in > FLIP-309. Given that we have > three dimensions: > - e2e latency/checkpointing interval > - enabling some kind of batching/buffering on the operator level > - how much resources we want to allocate to the job > > How do we want Flink to adjust itself between those three? For example: > a) Should we assume that given Job has a fixed amount of assigned > resources and make it paramount that > Flink doesn't exceed those available resources? So in case of > backpressure, we > should extend checkpointing intervals, emit records less frequently and > in batches. > b) Or should we assume that the amount of resources is flexible (up to a > point?), and the desired e2e latency > is the paramount aspect? So in case of backpressure, we should still > adhere to the configured e2e latency, > and wait for the user or autoscaler to scale up the job? > > In case of a), I think the concept of "isProcessingBacklog" is not needed, > we could steer the behaviour only > using the backpressure information. > > On the other hand, in case of b), "isProcessingBacklog" information might > be helpful, to let Flink know that > we can safely decrease the e2e latency/checkpoint interval even if there > is no backpressure, to use fewer > resources (and let the aut
Re: [ANNOUNCE] Apache Flink has won the 2023 SIGMOD Systems Award
Congratulations! Best, Shammon FY On Wed, Jul 5, 2023 at 2:38 PM Paul Lam wrote: > Congrats and cheers! > > Best, > Paul Lam > > > 2023年7月4日 18:04，Benchao Li 写道： > > > > Congratulations! > > > > Feng Jin 于2023年7月4日周二 16:17写道： > > > >> Congratulations! > >> > >> Best, > >> Feng Jin > >> > >> > >> > >> On Tue, Jul 4, 2023 at 4:13 PM Yuxin Tan > wrote: > >> > >>> Congratulations! > >>> > >>> Best, > >>> Yuxin > >>> > >>> > >>> Dunn Bangui 于2023年7月4日周二 16:04写道： > >>> > >>>> Congratulations! > >>>> > >>>> Best, > >>>> Bangui Dunn > >>>> > >>>> Yangze Guo 于2023年7月4日周二 15:59写道： > >>>> > >>>>> Congrats everyone! > >>>>> > >>>>> Best, > >>>>> Yangze Guo > >>>>> > >>>>> On Tue, Jul 4, 2023 at 3:53 PM Rui Fan <1996fan...@gmail.com> wrote: > >>>>>> > >>>>>> Congratulations! > >>>>>> > >>>>>> Best, > >>>>>> Rui Fan > >>>>>> > >>>>>> On Tue, Jul 4, 2023 at 2:08 PM Zhu Zhu wrote: > >>>>>> > >>>>>>> Congratulations everyone! > >>>>>>> > >>>>>>> Thanks, > >>>>>>> Zhu > >>>>>>> > >>>>>>> Hang Ruan 于2023年7月4日周二 14:06写道： > >>>>>>>> > >>>>>>>> Congratulations! > >>>>>>>> > >>>>>>>> Best, > >>>>>>>> Hang > >>>>>>>> > >>>>>>>> Jingsong Li 于2023年7月4日周二 13:47写道： > >>>>>>>> > >>>>>>>>> Congratulations! > >>>>>>>>> > >>>>>>>>> Thank you! All of the Flink community! > >>>>>>>>> > >>>>>>>>> Best, > >>>>>>>>> Jingsong > >>>>>>>>> > >>>>>>>>> On Tue, Jul 4, 2023 at 1:24 PM tison > >>>> wrote: > >>>>>>>>>> > >>>>>>>>>> Congrats and with honor :D > >>>>>>>>>> > >>>>>>>>>> Best, > >>>>>>>>>> tison. > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>> Mang Zhang 于2023年7月4日周二 11:08写道： > >>>>>>>>>> > >>>>>>>>>>> Congratulations!-- > >>>>>>>>>>> > >>>>>>>>>>> Best regards, > >>>>>>>>>>> Mang Zhang > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> 在 2023-07-04 01:53:46，"liu ron" 写道： > >>>>>>>>>>>> Congrats everyone > >>>>>>>>>>>> > >>>>>>>>>>>> Best, > >>>>>>>>>>>> Ron > >>>>>>>>>>>> > >>>>>>>>>>>> Jark Wu 于2023年7月3日周一 22:48写道： > >>>>>>>>>>>> > >>>>>>>>>>>>> Congrats everyone! > >>>>>>>>>>>>> > >>>>>>>>>>>>> Best, > >>>>>>>>>>>>> Jark > >>>>>>>>>>>>> > >>>>>>>>>>>>>> 2023年7月3日 22:37，Yuval Itzchakov > >>> 写道： > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> Congrats team! > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> On Mon, Jul 3, 2023, 17:28 Jing Ge via user < > >>>>>>>>> u...@flink.apache.org > >>>>
Re: [DISCUSS] FLIP-314: Support Customized Job Lineage Listener
Hi devs, Thanks for all the feedback. I have discussed with @QingSheng Ren off-line to confirm some questionable points in the FLIP. Thanks for his valuable inputs and I have updated the FLIP according to our discussion. Looking forward to your feedback, thanks, Best, Shammon FY On Wed, Jul 5, 2023 at 5:26 PM Shammon FY wrote: > Hi Jing, > > Thanks for your feedback. > > > 1. TableColumnLineageRelation#sinkColumn() should return > TableColumnLineageEntity instead of String, right? > > The `sinkColumn()` will return `String` which is the column name in the > sink connector. I found the name of `TableColumnLineageEntity` may > cause ambiguity and I have renamed it to `TableColumnSourceLineageEntity`. > In my mind the `TableColumnLineageRelation` represents the lineage for each > sink column, each column may be computed from multiple sources and columns. > I use `TableColumnSourceLineageEntity` to manage each source and its > columns for the sink column, so `TableColumnLineageRelation` has a sink > column name and `TableColumnSourceLineageEntity` list. > > > 2. Since LineageRelation already contains all information to build the > lineage between sources and sink, do we still need to set the LineageEntity > in the source? > > The lineage interface of `DataStream` is very flexible. We have added > `setLineageEntity` to the source to limit and verify user behavior, > ensuring that users have not added non-existent sources as lineage. > > > 3. About the "Entity" and "Relation" naming, I was confused too, like > Qingsheng mentioned. How about LineageVertex, LineageEdge, and LineageEdges > which contains multiple LineageEdge? > > We referred to `Atlas` for the name of lineage, it uses `Entity` and > `Relation` to represent the lineage relationship and another metadata > service `Datahub` uses `DataSet` to represent the entity. I think `Entity` > and `Relation` are nicer for lineage, what do you think of it? > > Best, > Shammon FY > > > On Thu, Jun 29, 2023 at 4:21 AM Jing Ge > wrote: > >> Hi Shammon, >> >> Thanks for your proposal. After reading the FLIP, I'd like to ask >> some questions to make sure we are on the same page. Thanks! >> >> 1. TableColumnLineageRelation#sinkColumn() should return >> TableColumnLineageEntity instead of String, right? >> >> 2. Since LineageRelation already contains all information to build the >> lineage between sources and sink, do we still need to set the >> LineageEntity >> in the source? >> >> 3. About the "Entity" and "Relation" naming, I was confused too, like >> Qingsheng mentioned. How about LineageVertex, LineageEdge, and >> LineageEdges >> which contains multiple LineageEdge? E.g. multiple sources join into one >> sink, or, edges of columns from one or different tables, etc. >> >> Best regards, >> Jing >> >> On Sun, Jun 25, 2023 at 2:06 PM Shammon FY wrote: >> >> > Hi yuxia and Yun, >> > >> > Thanks for your input. >> > >> > For yuxia: >> > > 1: What kinds of JobStatus will the `JobExecutionStatusEven` >> including? >> > >> > At present, we only need to notify the listener when a job goes to >> > termination, but I think it makes sense to add generic `oldStatus` and >> > `newStatus` in the listener and users can update the job state in their >> > service as needed. >> > >> > > 2: I'm really confused about the `config()` included in >> `LineageEntity`, >> > where is it from and what is it for ? >> > >> > The `config` in `LineageEntity` is used for users to get options for >> source >> > and sink connectors. As the examples in the FLIP, users can add >> > server/group/topic information in the config for kafka and create >> lineage >> > entities for `DataStream` jobs, then the listeners can get this >> information >> > to identify the same connector in different jobs. Otherwise, the >> `config` >> > in `TableLineageEntity` will be the same as `getOptions` in >> > `CatalogBaseTable`. >> > >> > > 3: Regardless whether `inputChangelogMode` in >> `TableSinkLineageEntity` is >> > needed or not, since `TableSinkLineageEntity` contains >> > `inputChangelogMode`, why `TableSourceLineageEntity` don't contain >> > changelogmode? >> > >> > At present, we do not actually use the changelog mode. It can be >> deleted, >> > and I have updated FLIP. >> > >> > > Btw, since there're a lot interfaces proposed, I think i
Re: [DISCUSS] FLIP-314: Support Customized Job Lineage Listener
Hi Jing, Thanks for your feedback. > 1. TableColumnLineageRelation#sinkColumn() should return TableColumnLineageEntity instead of String, right? The `sinkColumn()` will return `String` which is the column name in the sink connector. I found the name of `TableColumnLineageEntity` may cause ambiguity and I have renamed it to `TableColumnSourceLineageEntity`. In my mind the `TableColumnLineageRelation` represents the lineage for each sink column, each column may be computed from multiple sources and columns. I use `TableColumnSourceLineageEntity` to manage each source and its columns for the sink column, so `TableColumnLineageRelation` has a sink column name and `TableColumnSourceLineageEntity` list. > 2. Since LineageRelation already contains all information to build the lineage between sources and sink, do we still need to set the LineageEntity in the source? The lineage interface of `DataStream` is very flexible. We have added `setLineageEntity` to the source to limit and verify user behavior, ensuring that users have not added non-existent sources as lineage. > 3. About the "Entity" and "Relation" naming, I was confused too, like Qingsheng mentioned. How about LineageVertex, LineageEdge, and LineageEdges which contains multiple LineageEdge? We referred to `Atlas` for the name of lineage, it uses `Entity` and `Relation` to represent the lineage relationship and another metadata service `Datahub` uses `DataSet` to represent the entity. I think `Entity` and `Relation` are nicer for lineage, what do you think of it? Best, Shammon FY On Thu, Jun 29, 2023 at 4:21 AM Jing Ge wrote: > Hi Shammon, > > Thanks for your proposal. After reading the FLIP, I'd like to ask > some questions to make sure we are on the same page. Thanks! > > 1. TableColumnLineageRelation#sinkColumn() should return > TableColumnLineageEntity instead of String, right? > > 2. Since LineageRelation already contains all information to build the > lineage between sources and sink, do we still need to set the LineageEntity > in the source? > > 3. About the "Entity" and "Relation" naming, I was confused too, like > Qingsheng mentioned. How about LineageVertex, LineageEdge, and LineageEdges > which contains multiple LineageEdge? E.g. multiple sources join into one > sink, or, edges of columns from one or different tables, etc. > > Best regards, > Jing > > On Sun, Jun 25, 2023 at 2:06 PM Shammon FY wrote: > > > Hi yuxia and Yun, > > > > Thanks for your input. > > > > For yuxia: > > > 1: What kinds of JobStatus will the `JobExecutionStatusEven` including? > > > > At present, we only need to notify the listener when a job goes to > > termination, but I think it makes sense to add generic `oldStatus` and > > `newStatus` in the listener and users can update the job state in their > > service as needed. > > > > > 2: I'm really confused about the `config()` included in > `LineageEntity`, > > where is it from and what is it for ? > > > > The `config` in `LineageEntity` is used for users to get options for > source > > and sink connectors. As the examples in the FLIP, users can add > > server/group/topic information in the config for kafka and create lineage > > entities for `DataStream` jobs, then the listeners can get this > information > > to identify the same connector in different jobs. Otherwise, the `config` > > in `TableLineageEntity` will be the same as `getOptions` in > > `CatalogBaseTable`. > > > > > 3: Regardless whether `inputChangelogMode` in `TableSinkLineageEntity` > is > > needed or not, since `TableSinkLineageEntity` contains > > `inputChangelogMode`, why `TableSourceLineageEntity` don't contain > > changelogmode? > > > > At present, we do not actually use the changelog mode. It can be deleted, > > and I have updated FLIP. > > > > > Btw, since there're a lot interfaces proposed, I think it'll be better > to > > give an example about how to implement a listener in this FLIP to make us > > know better about the interfaces. > > > > I have added the example in the FLIP and the related interfaces and > > examples are in branch [1]. > > > > For Yun: > > > I have one more question on the lookup-join dim tables, it seems this > > FLIP does not touch them, and will them become part of the > > List sources() or adding another interface? > > > > You're right, currently lookup join dim tables were not considered in the > > 'proposed changed' section of this FLIP. But the interface for lineage is > > universal and we can give `TableLookupSourceLineageEntity` which > implements > > `TableSourceLineageEntity` in the fut
Re: [DISCUSS] FLIP-325: Support configuring end-to-end allowed latency
Hi Dong and Yunfeng, Thanks for bringing up this discussion. As described in the FLIP, the differences between `end-to-end latency` and `table.exec.mini-batch.allow-latency` are: "It allows users to specify the end-to-end latency, whereas table.exec.mini-batch.allow-latency applies to each operator. If there are N operators on the path from source to sink, the end-to-end latency could be up to table.exec.mini-batch.allow-latency * N". If I understand correctly, `table.exec.mini-batch.allow-latency` is also applied to the end-to-end latency for a job, maybe @Jack Wu can give more information. So, from my perspective, and please correct me if I'm misunderstand, the targets of this FLIP may include the following: 1. Support a mechanism like `mini-batch` in SQL for `DataStream`, which will collect data in the operator and flush data when it receives a `flush` event, in the FLIP it is `FlushEvent`. 2. Support dynamic `latency` according to the progress of job, such as snapshot stage and after that. To do that, I have some questions: 1. I didn't understand the purpose of public interface `RecordAttributes`. I think `FlushEvent` in the FLIP is enough, and different `DynamicFlushStrategy` can be added to generate flush events to address different needs, such as a static interval similar to mini-batch in SQL or collect the information `isProcessingBacklog` and metrics to generate `FlushEvent` which is described in your FLIP? If hudi sink needs the `isBacklog` flag, the hudi `SplitEnumerator` can create an operator event and send it to hudi source reader. 2. How is this new mechanism unified with SQL's mini-batch mechanism? As far as I am concerned, SQL implements mini-batch mechanism based on watermark, I think it is very unreasonable to have two different implementation in SQL and DataStream. 3. I notice that the `CheckpointCoordinator` will generate `FlushEvent`, which information about `FlushEvent` will be stored in `Checkpoint`? What is the alignment strategy for FlushEvent in the operator? The operator will flush the data when it receives all `FlushEvent` from upstream with the same ID or do flush for each `FlushEvent`? Can you give more detailed proposal about that? We also have a demand for this piece, thanks Best, Shammon FY On Thu, Jun 29, 2023 at 4:35 PM Martijn Visser wrote: > Hi Dong and Yunfeng, > > Thanks for the FLIP. What's not clear for me is what's the expected > behaviour when the allowed latency can't be met, for whatever reason. > Given that we're talking about an "allowed latency", it implies that > something has gone wrong and should fail? Isn't this more a minimum > latency that you're proposing? > > There's also the part about the Hudi Sink processing records > immediately upon arrival. Given that the SinkV2 API provides the > ability for custom post and pre-commit topologies [1], specifically > targeted to avoid generating multiple small files, why isn't that > sufficient for the Hudi Sink? It would be great to see that added > under Rejected Alternatives if this is indeed not sufficient. > > Best regards, > > Martijn > > [1] > https://cwiki.apache.org/confluence/display/FLINK/FLIP-191%3A+Extend+unified+Sink+interface+to+support+small+file+compaction > > On Sun, Jun 25, 2023 at 4:25 AM Yunfeng Zhou > wrote: > > > > Hi all, > > > > Dong(cc'ed) and I are opening this thread to discuss our proposal to > > support configuring end-to-end allowed latency for Flink jobs, which > > has been documented in FLIP-325 > > < > https://cwiki.apache.org/confluence/display/FLINK/FLIP-325%3A+Support+configuring+end-to-end+allowed+latency > >. > > > > By configuring the latency requirement for a Flink job, users would be > > able to optimize the throughput and overhead of the job while still > > acceptably increasing latency. This approach is particularly useful > > when dealing with records that do not require immediate processing and > > emission upon arrival. > > > > Please refer to the FLIP document for more details about the proposed > > design and implementation. We welcome any feedback and opinions on > > this proposal. > > > > Best regards. > > > > Dong and Yunfeng >
Re: [DISCUSS] FLIP-314: Support Customized Job Lineage Listener
Hi Feng, Thanks for your input. >1. we can add a lineage interface like `supportReportLineage` It's a so good idea and thanks very much. It can help users to report lineage for existing connectors in DataStream jobs without any additional operations. I will give this interface in the FLIP later and please help to review it, thanks >2. it is relatively easy to obtain column lineage through Calcite MetaQuery API It's helpful if Calcite already has some column lineage in meta, I think we can discuss and give the proposal in the column lineage FLIP Best, Shammon FY On Wednesday, June 28, 2023, Feng Jin wrote: > Hi Shammon > Thank you for proposing this FLIP. I think the Flink Job lineage is a very > useful feature. > I have few question: > > 1. For DataStream Jobs, users need to set up lineage relationships when > building DAGs for their custom sources and sinks. > However, for some common connectors such as Kafka Connector and JDBC > Connector, we can add a lineage interface like `supportReportLineage`, so > that these connectors can implement it. > This way, in the scenario of DataStream Jobs, lineages can be automatically > reported. What do you think? > > > 2. From the current design, it seems that we need to analyze column lineage > through pipeline. As far as I know, it is relatively easy to obtain column > lineage through Calcite MetaQuery API. > Would you consider using this approach? Or do we need to implement another > parsing process based on the pipeline? > ``` > RelMetadataQuery metadataQuery = relNode.getCluster().getMetadataQuery(); > metadataQuery.getColumnOrigins(inputRel, i); > ``` > Best, > Feng > > > On Sun, Jun 25, 2023 at 8:06 PM Shammon FY wrote: > > > Hi yuxia and Yun, > > > > Thanks for your input. > > > > For yuxia: > > > 1: What kinds of JobStatus will the `JobExecutionStatusEven` including? > > > > At present, we only need to notify the listener when a job goes to > > termination, but I think it makes sense to add generic `oldStatus` and > > `newStatus` in the listener and users can update the job state in their > > service as needed. > > > > > 2: I'm really confused about the `config()` included in > `LineageEntity`, > > where is it from and what is it for ? > > > > The `config` in `LineageEntity` is used for users to get options for > source > > and sink connectors. As the examples in the FLIP, users can add > > server/group/topic information in the config for kafka and create lineage > > entities for `DataStream` jobs, then the listeners can get this > information > > to identify the same connector in different jobs. Otherwise, the `config` > > in `TableLineageEntity` will be the same as `getOptions` in > > `CatalogBaseTable`. > > > > > 3: Regardless whether `inputChangelogMode` in `TableSinkLineageEntity` > is > > needed or not, since `TableSinkLineageEntity` contains > > `inputChangelogMode`, why `TableSourceLineageEntity` don't contain > > changelogmode? > > > > At present, we do not actually use the changelog mode. It can be deleted, > > and I have updated FLIP. > > > > > Btw, since there're a lot interfaces proposed, I think it'll be better > to > > give an example about how to implement a listener in this FLIP to make us > > know better about the interfaces. > > > > I have added the example in the FLIP and the related interfaces and > > examples are in branch [1]. > > > > For Yun: > > > I have one more question on the lookup-join dim tables, it seems this > > FLIP does not touch them, and will them become part of the > > List sources() or adding another interface? > > > > You're right, currently lookup join dim tables were not considered in the > > 'proposed changed' section of this FLIP. But the interface for lineage is > > universal and we can give `TableLookupSourceLineageEntity` which > implements > > `TableSourceLineageEntity` in the future without modifying the public > > interface. > > > > > By the way, if you want to focus on job lineage instead of data column > > lineage in this FLIP, why we must introduce so many column-lineage > related > > interface here? > > > > The lineage information in SQL jobs includes table lineage and column > > lineage. Although SQL jobs currently do not support column lineage, we > > would like to support this in the next step. So we have comprehensively > > considered the table lineage and column lineage interfaces here, and > > defined these two interfaces together clearly > > > > > > [1] > > > > https://github.com/Fan
Re: [DISCUSS] Persistent SQL Gateway
Hi Ferenc, If I understand correctly, there will be two types of jobs in sql-gateway: `SELECT` and `NON-SELECT` such as `INSERT`. 1. `SELECT` jobs need to collect results from Flink cluster in a corresponding session of sql gateway, and when the session is closed, the job should be canceled. These jobs are generally short queries similar to OLAP and I think it may be acceptable. 2. `NON-SELECT` jobs may be batch or streaming jobs, and when the jobs are submitted successfully, they won't be killed or canceled even if the session or sql-gateway is closed. After these assignments are successfully submitted, the lifecycle is no longer managed by SQL gateway. I don't know if it covers your usage scenario. Could you describe yours for us to test and confirm? Best, Shammon FY On Tue, Jun 27, 2023 at 6:43 PM Ferenc Csaky wrote: > Hi Jark, > > In the current implementation, any job submitted via the SQL Gateway has > to be done through a session, cause all the operations are grouped under > sessions. > > Starting from there, if I close a session, that will close the > "SessionContext", which closes the "OperationManager" [1], and the > "OperationManager" closes all submitted operations tied to that session > [2], which results closing all the jobs executed in the session. > > Maybe I am missing something, but my experience is that the jobs I submit > via the SQL Gateway are getting cleaned up on gateway session close. > > WDYT? > > Cheers, > F > > [1] > https://github.com/apache/flink/blob/149a5e34c1ed8d8943c901a98c65c70693915811/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java#L204 > [2] > https://github.com/apache/flink/blob/149a5e34c1ed8d8943c901a98c65c70693915811/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationManager.java#L194 > > > > --- Original Message --- > On Tuesday, June 27th, 2023 at 04:37, Jark Wu wrote: > > > > > > > > Hi Ferenc, > > > > But the job lifecycle doesn't tie to the SQL Gateway session. > > Even if the session is closed, all the running jobs are not affected. > > > > Best, > > Jark > > > > > > > > > > On Tue, 27 Jun 2023 at 04:14, Ferenc Csaky ferenc.cs...@pm.me.invalid > > > > wrote: > > > > > Hi Jark, > > > > > > Thank you for pointing out FLIP-295 abouth catalog persistence, I was > not > > > aware the current state. Although as far as I see, that persistent > catalogs > > > are necessary, but not sufficient achieving a "persistent gateway". > > > > > > The current implementation ties the job lifecycle to the SQL gateway > > > session, so if it gets closed, it will cancel all the jobs. So that > would > > > be the next step I think. Any work or thought regarding this aspect? > We are > > > definitely willing to help out on this front. > > > > > > Cheers, > > > F > > > > > > --- Original Message --- > > > On Sunday, June 25th, 2023 at 06:23, Jark Wu imj...@gmail.com wrote: > > > > > > > Hi Ferenc, > > > > > > > > Making SQL Gateway to be an easy-to-use platform infrastructure of > Flink > > > > SQL > > > > is one of the important roadmaps 1. > > > > > > > > The persistence ability of the SQL Gateway is a major work in 1.18 > > > > release. > > > > One of the persistence demand is that the registered catalogs are > > > > currently > > > > kept in memory and lost when Gateway restarts. There is an accepted > FLIP > > > > (FLIP-295)[2] target to resolve this issue and make Gateway can > persist > > > > the > > > > registered catalogs information into files or databases. > > > > > > > > I'm not sure whether this is something you are looking for? > > > > > > > > Best, > > > > Jark > > > > > > > > [2]: > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-295%3A+Support+lazy+initialization+of+catalogs+and+persistence+of+catalog+configurations > > > > > > > On Fri, 23 Jun 2023 at 00:25, Ferenc Csaky ferenc.cs...@pm.me.invalid > > > > > > > > wrote: > > > > > > > > > Hello devs, > > > > > > > > > > I would like to open a discussion about persistence possibilitis > for > > > > > the > > > > > SQL Gateway. At Cloudera, we are happy to see
Re: [DISCUSS] FLIP-314: Support Customized Job Lineage Listener
Hi yuxia and Yun, Thanks for your input. For yuxia: > 1: What kinds of JobStatus will the `JobExecutionStatusEven` including? At present, we only need to notify the listener when a job goes to termination, but I think it makes sense to add generic `oldStatus` and `newStatus` in the listener and users can update the job state in their service as needed. > 2: I'm really confused about the `config()` included in `LineageEntity`, where is it from and what is it for ? The `config` in `LineageEntity` is used for users to get options for source and sink connectors. As the examples in the FLIP, users can add server/group/topic information in the config for kafka and create lineage entities for `DataStream` jobs, then the listeners can get this information to identify the same connector in different jobs. Otherwise, the `config` in `TableLineageEntity` will be the same as `getOptions` in `CatalogBaseTable`. > 3: Regardless whether `inputChangelogMode` in `TableSinkLineageEntity` is needed or not, since `TableSinkLineageEntity` contains `inputChangelogMode`, why `TableSourceLineageEntity` don't contain changelogmode? At present, we do not actually use the changelog mode. It can be deleted, and I have updated FLIP. > Btw, since there're a lot interfaces proposed, I think it'll be better to give an example about how to implement a listener in this FLIP to make us know better about the interfaces. I have added the example in the FLIP and the related interfaces and examples are in branch [1]. For Yun: > I have one more question on the lookup-join dim tables, it seems this FLIP does not touch them, and will them become part of the List sources() or adding another interface? You're right, currently lookup join dim tables were not considered in the 'proposed changed' section of this FLIP. But the interface for lineage is universal and we can give `TableLookupSourceLineageEntity` which implements `TableSourceLineageEntity` in the future without modifying the public interface. > By the way, if you want to focus on job lineage instead of data column lineage in this FLIP, why we must introduce so many column-lineage related interface here? The lineage information in SQL jobs includes table lineage and column lineage. Although SQL jobs currently do not support column lineage, we would like to support this in the next step. So we have comprehensively considered the table lineage and column lineage interfaces here, and defined these two interfaces together clearly [1] https://github.com/FangYongs/flink/commit/d4bfe57e7a5315b790e79b8acef8b11e82c9187c Best, Shammon FY On Sun, Jun 25, 2023 at 4:17 PM Yun Tang wrote: > Hi Shammon, > > I like the idea in general and it will help to analysis the job lineages > no matter FlinkSQL or Flink jar jobs in production environments. > > For Qingsheng's concern, I'd like the name of JobType more than > RuntimeExecutionMode, as the latter one is not easy to understand for users. > > I have one more question on the lookup-join dim tables, it seems this FLIP > does not touch them, and will them become part of the List > sources() or adding another interface? > > By the way, if you want to focus on job lineage instead of data column > lineage in this FLIP, why we must introduce so many column-lineage related > interface here? > > > Best > Yun Tang > > From: Shammon FY > Sent: Sunday, June 25, 2023 16:13 > To: dev@flink.apache.org > Subject: Re: [DISCUSS] FLIP-314: Support Customized Job Lineage Listener > > Hi Qingsheng, > > Thanks for your valuable feedback. > > > 1. Is there any specific use case to expose the batch / streaming info to > listeners or meta services? > > I agree with you that Flink is evolving towards batch-streaming > unification, but the lifecycle of them is different. If a job processes a > bound dataset, it will end after completing the data processing, otherwise, > it will run for a long time. In our scenario, we will regularly schedule > some Flink jobs to process bound dataset and update some job information to > the lineage information for the "batch" jobs such as scheduled timestamp, > execution duration when jobs are finished, which is different from > "streaming" jobs. Currently Flink uses `RuntimeExecutionMode` and > `existsUnboundedSource` in `StreamingGraph` and `StreamingGraphGenerator` > to determine `JobType` and disjoin jobs. We can mark `JobType` as > `PublicEvolving` or use `RuntimeExecutionMode` and a boolean flag, what do > you think of it? > > > 2. it’s better to be more specific here to tell users what information > they could expect to see here, instead of just a “job configuration” as > described in JavaDoc. > > Thanks and I have updated the doc in FLIP. > > > 3. About the IO executor in JobStatusCha
Re: [DISCUSS] FLIP-314: Support Customized Job Lineage Listener
Hi Qingsheng,
Thanks for your valuable feedback.
> 1. Is there any specific use case to expose the batch / streaming info to
listeners or meta services?
I agree with you that Flink is evolving towards batch-streaming
unification, but the lifecycle of them is different. If a job processes a
bound dataset, it will end after completing the data processing, otherwise,
it will run for a long time. In our scenario, we will regularly schedule
some Flink jobs to process bound dataset and update some job information to
the lineage information for the "batch" jobs such as scheduled timestamp,
execution duration when jobs are finished, which is different from
"streaming" jobs. Currently Flink uses `RuntimeExecutionMode` and
`existsUnboundedSource` in `StreamingGraph` and `StreamingGraphGenerator`
to determine `JobType` and disjoin jobs. We can mark `JobType` as
`PublicEvolving` or use `RuntimeExecutionMode` and a boolean flag, what do
you think of it?
> 2. it’s better to be more specific here to tell users what information
they could expect to see here, instead of just a “job configuration” as
described in JavaDoc.
Thanks and I have updated the doc in FLIP.
> 3. About the IO executor in JobStatusChangedListenerFactory.Context.
I have updated the docs for io executor in
`JobStatusChangedListenerFactory.Context`, it is a regular thread pool and
executes submitted tasks in parallel. Users can submit tasks to the
executor which ensures that the submitted task can be executed before the
job exits.
> 4. I don’t quite get the LineageRelationEntity, which is just a list of
LineageEntity.
In the initial idea, the `LineageRelationEntity` is used for `DataStream`
to set additional lineage information besides source. For example, there
are table and column lineages in SQL jobs. When we build a `DataStream` job
with table source and sink, we can add table lineage in the following
method.
```
public class DataStreamSink {
public DataStreamSink setLineageSources(LineageEntity ... sources);
}
```
But we can not set column lineage for the above sink, and for the sake of
universality, we do not want to add a method similar to `addLineageColumn
(...)` in `DataStreamSink`. So I put this information into
LineageRelationEntity so that SQL and DataStream jobs can be consistent.
But as you mentioned, this approach does indeed lead to ambiguity and
complexity. So my current idea is to add the `setLineageRelation` method in
`DataStreamSink` directly without `LineageRelationEntity`, I have updated
the FLIP and please help to review it again, thanks.
> 5. I can’t find the definition of CatalogContext in the current code base
and Flink, which appears in the TableLineageEntity.
CatalogContext is defined in FLIP-294 and I have updated the FLIP
> 6. TableSinkLineageEntity exposes ModifyType, ChangelogMode and a boolean
(the “override” is quite confusing). I’m wondering if these are necessary
for meta services, as they are actually concepts defined in the runtime
level of Flink Table / SQL.
The information in `TableSinkLineageEntity` such as `ModifyType`,
`ChangelogMode` and `override` are mainly used for verification and
display. For example, Flink currently supports `INSERT`/`DELETE` and
`UPDATE`, we only want to report and update lineage for `INSERT` jobs in
our streaming & batch ETL, and display the `override` information on the UI.
Best,
Shammon FY
On Tue, Jun 20, 2023 at 6:19 PM Qingsheng Ren wrote:
> Hi Shammon,
>
> Thanks for starting this FLIP! Data lineage is a very important topic,
> which has been missing for a long time in Flink. I have some questions
> about the FLIP.
>
> About events and listeners:
>
> 1. I’m not sure if it is necessary to expose JobType to in JobCreatedEvent.
> This is an internal class in flink-runtime, and I think the correct API
> should be RuntimeExecutionMode. Furthermore, I think the boundary of batch
> and streaming becomes much more vague as Flink is evolving towards
> batch-streaming unification, so I’m concerned about exposing JobType as a
> public API. Is there any specific use case to expose the batch / streaming
> info to listeners or meta services?
>
> 2. Currently JobCreatedEvent gives a Configuration, which is quite
> ambiguous. To be honest the configuration is quite a mess in Flink, so
> maybe it’s better to be more specific here to tell users what information
> they could expect to see here, instead of just a “job configuration” as
> described in JavaDoc.
>
> 3. JobStatusChangedListenerFactory.Context provides an IO executor. I think
> more information should be provided here, such as which thread model this
> executor could promise, and whether the user should care about concurrency
> issues. Otherwise I prefer not to give such an utility that no one dares to
> use safely, and leave it to users to choose their implementation.
>
> Abo
Re: [DISCUSS] FLIP-314: Support Customized Job Lineage Listener
Hi devs, Is there any comment or feedback for this FLIP? Hope to hear from you, thanks Best, Shammon FY On Tue, Jun 6, 2023 at 8:22 PM Shammon FY wrote: > Hi devs, > > I would like to start a discussion on FLIP-314: Support Customized Job > Lineage Listener[1] which is the next stage of FLIP-294 [2]. Flink > streaming and batch jobs create lineage dependency between source and sink, > users can manage their data and jobs according to this lineage information. > For example, when there is a delay in Flink ETL or data, users can easily > trace the problematic jobs and affected data. On the other hand, when users > need to correct data or debug, they can perform operations based on lineage > too. > > In FLIP-314 we want to introduce lineage related interfaces for Flink and > users can create customized job status listeners. When job status changes, > users can get job status and information to add, update or delete lineage. > > Looking forward to your feedback, thanks. > > > [1] > https://cwiki.apache.org/confluence/display/FLINK/FLIP-314%3A+Support+Customized+Job+Lineage+Listener > [2] > https://cwiki.apache.org/confluence/display/FLINK/FLIP-294%3A+Support+Customized+Job+Meta+Data+Listener > > Best, > Shammon FY >
[RESULT][VOTE] FLIP-294: Support Customized Catalog Modification Listener
Hi devs, Happy to announce that FLIP-294 [1] has been approved ! Voting included 6 votes [2], out of which 4 were binding and no disapproving votes. - Benchao Li (binding) - Jing Ge (binding) - Leonard Xu (binding) - yuxia (binding) - liu ron (non-binding) - Feng Jin (non-binding) Thanks all for participating! [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-294%3A+Support+Customized+Catalog+Modification+Listener [2] https://lists.apache.org/thread/5f9806kw1q01536wzwgx1psh7jhmfmjw Best, Shammon FY
Re: [VOTE] FLIP-295: Support lazy initialization of catalogs and persistence of catalog configurations
Thanks Feng for the FLIP. +1(non-binding) Best, Shammon FY On Thu, Jun 15, 2023 at 6:30 PM Leonard Xu wrote: > Thanks Feng for driving this FLIP forward. > > +1(binding) > > Best, > Leonard > > > On Jun 15, 2023, at 1:42 PM, Dong Lin wrote: > > > > Thanks Feng for the FLIP. > > > > +1(binding) > > > > Cheers, > > Dong > > > > On Wed, Jun 14, 2023 at 10:35 AM Feng Jin wrote: > > > >> Hi everyone > >> > >> Thanks for all the feedback about the FLIP-295: Support lazy > initialization > >> of catalogs and persistence of catalog configurations[1]. > >> [2] is the discussion thread. > >> > >> > >> I'd like to start a vote for it. The vote will be open for at least 72 > >> hours(excluding weekends,until June 19, 10:00AM GMT) unless there is an > >> objection or an insufficient number of votes. > >> > >> > >> [1] > >> > >> > https://cwiki.apache.org/confluence/display/FLINK/FLIP-295%3A+Support+lazy+initialization+of+catalogs+and+persistence+of+catalog+configurations > >> [2]https://lists.apache.org/thread/dcwgv0gmngqt40fl3694km53pykocn5s > >> > >> > >> Best, > >> Feng > >> > >
[VOTE] FLIP-294: Support Customized Catalog Modification Listener
Hi all: Thanks for all the feedback for FLIP-294: Support Customized Catalog Modification Listener [1]. I would like to start a vote for it according to the discussion in thread [2]. The vote will be open for at least 72 hours(excluding weekends, until June 19, 19:00 PM GMT) unless there is an objection or an insufficient number of votes. [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-294%3A+Support+Customized+Catalog+Modification+Listener [2] https://lists.apache.org/thread/185mbcwnpokfop4xcb22r9bgfp2m68fx Best, Shammon FY
Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener
Hi all, Thanks for all the valuable feedback. If there's no further question or concern, I will start voting for FLIP-294 [1] later. [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-294%3A+Support+Customized+Catalog+Modification+Listener Best, Shammon FY On Fri, Jun 9, 2023 at 9:07 AM Shammon FY wrote: > Thanks Jing, it makes sense to me and I have updated the FLIP > > > Best, > Shammon FY > > > On Thu, Jun 8, 2023 at 11:15 PM Jing Ge > wrote: > >> Hi Shammon, >> >> If we take a look at the JDK Event design as a reference, we can even add >> an Object into the event [1]. Back to the CatalogModificationEvent, >> everything related to the event could be defined in the Event. If we want >> to group some information into the Context, we could also consider adding >> the CatalogModificationContext into the Event and make the onEvent() >> method >> cleaner with only one input parameter CatalogModificationEvent, because >> the >> interface CatalogModificationListener is the most often used interface for >> users. Just my two cents. >> >> Best regards, >> Jing >> >> [1] >> >> http://www.java2s.com/example/java-src/pkg/java/util/eventobject-85298.html >> >> On Thu, Jun 8, 2023 at 7:50 AM Shammon FY wrote: >> >> > Hi, >> > >> > To @Jing Ge >> > > Thanks for the clarification. Just out of curiosity, if the context is >> > not part of the event, why should it be the input parameter of each >> onEvent >> > call? >> > >> > I think it's quite strange to put some information in an Event, such as >> a >> > factory identifier for catalog, but they will be used by the listener. >> I >> > place it in the context class and I think it is more suitable than >> directly >> > placing it in the event class. >> > >> > To @Mason >> > > 1. I'm also curious about default implementations. Would >> atlas/datahub be >> > supported by default? >> > >> > We won't do that and external systems such as atlas/datahub need to >> > implement the listener themselves. >> > >> > > 2. The FLIP title is confusing to me, especially in distinguishing it >> > from FLIP-314. Would a better FLIP title be "Support Catalog Metadata >> > Listener" or something alike? >> > >> > Thanks, I think "Support Catalog Modification Listener" will be >> > more suitable, I'll update the title to it. >> > >> > >> > Best, >> > Shammon FY >> > >> > >> > On Thu, Jun 8, 2023 at 12:25 PM Mason Chen >> wrote: >> > >> > > Hi Shammon, >> > > >> > > FLIP generally looks good and I'm excited to see this feature. >> > > >> > > 1. I'm also curious about default implementations. Would >> atlas/datahub be >> > > supported by default? >> > > 2. The FLIP title is confusing to me, especially in distinguishing it >> > from >> > > FLIP-314. Would a better FLIP title be "Support Catalog Metadata >> > Listener" >> > > or something alike? >> > > >> > > Best, >> > > Mason >> > > >> > > On Tue, Jun 6, 2023 at 3:33 AM Jing Ge >> > wrote: >> > > >> > > > Hi Shammon, >> > > > >> > > > Thanks for the clarification. Just out of curiosity, if the context >> is >> > > not >> > > > part of the event, why should it be the input parameter of each >> onEvent >> > > > call? >> > > > >> > > > Best regards, >> > > > Jing >> > > > >> > > > On Tue, Jun 6, 2023 at 11:58 AM Leonard Xu >> wrote: >> > > > >> > > > > Thanks Shammon for the timely update, the updated FLIP looks good >> to >> > > me. >> > > > > >> > > > > Hope to see the vote thread and following FLIP-314 discussion >> thread. >> > > > > >> > > > > Best, >> > > > > Leonard >> > > > > >> > > > > > On Jun 6, 2023, at 5:04 PM, Shammon FY >> wrote: >> > > > > > >> > > > > > Hi, >> > > > > > >> > > > > > Thanks for all the feedback. >> > > > > > >> > > > > > For @Jing Ge, >> > > > > &
Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener
Thanks Jing, it makes sense to me and I have updated the FLIP Best, Shammon FY On Thu, Jun 8, 2023 at 11:15 PM Jing Ge wrote: > Hi Shammon, > > If we take a look at the JDK Event design as a reference, we can even add > an Object into the event [1]. Back to the CatalogModificationEvent, > everything related to the event could be defined in the Event. If we want > to group some information into the Context, we could also consider adding > the CatalogModificationContext into the Event and make the onEvent() method > cleaner with only one input parameter CatalogModificationEvent, because the > interface CatalogModificationListener is the most often used interface for > users. Just my two cents. > > Best regards, > Jing > > [1] > http://www.java2s.com/example/java-src/pkg/java/util/eventobject-85298.html > > On Thu, Jun 8, 2023 at 7:50 AM Shammon FY wrote: > > > Hi, > > > > To @Jing Ge > > > Thanks for the clarification. Just out of curiosity, if the context is > > not part of the event, why should it be the input parameter of each > onEvent > > call? > > > > I think it's quite strange to put some information in an Event, such as a > > factory identifier for catalog, but they will be used by the listener. I > > place it in the context class and I think it is more suitable than > directly > > placing it in the event class. > > > > To @Mason > > > 1. I'm also curious about default implementations. Would atlas/datahub > be > > supported by default? > > > > We won't do that and external systems such as atlas/datahub need to > > implement the listener themselves. > > > > > 2. The FLIP title is confusing to me, especially in distinguishing it > > from FLIP-314. Would a better FLIP title be "Support Catalog Metadata > > Listener" or something alike? > > > > Thanks, I think "Support Catalog Modification Listener" will be > > more suitable, I'll update the title to it. > > > > > > Best, > > Shammon FY > > > > > > On Thu, Jun 8, 2023 at 12:25 PM Mason Chen > wrote: > > > > > Hi Shammon, > > > > > > FLIP generally looks good and I'm excited to see this feature. > > > > > > 1. I'm also curious about default implementations. Would atlas/datahub > be > > > supported by default? > > > 2. The FLIP title is confusing to me, especially in distinguishing it > > from > > > FLIP-314. Would a better FLIP title be "Support Catalog Metadata > > Listener" > > > or something alike? > > > > > > Best, > > > Mason > > > > > > On Tue, Jun 6, 2023 at 3:33 AM Jing Ge > > wrote: > > > > > > > Hi Shammon, > > > > > > > > Thanks for the clarification. Just out of curiosity, if the context > is > > > not > > > > part of the event, why should it be the input parameter of each > onEvent > > > > call? > > > > > > > > Best regards, > > > > Jing > > > > > > > > On Tue, Jun 6, 2023 at 11:58 AM Leonard Xu > wrote: > > > > > > > > > Thanks Shammon for the timely update, the updated FLIP looks good > to > > > me. > > > > > > > > > > Hope to see the vote thread and following FLIP-314 discussion > thread. > > > > > > > > > > Best, > > > > > Leonard > > > > > > > > > > > On Jun 6, 2023, at 5:04 PM, Shammon FY > wrote: > > > > > > > > > > > > Hi, > > > > > > > > > > > > Thanks for all the feedback. > > > > > > > > > > > > For @Jing Ge, > > > > > > I forget to update the demo code in the FLIP, the method is > > > > > > `onEvent(CatalogModificationEvent, CatalogModificationContext)` > and > > > > there > > > > > > is no `onEvent(CatalogModificationEvent)`. I have updated the > code. > > > > > Context > > > > > > contains some additional information that is not part of an > Event, > > > but > > > > > > needs to be used in the listener, so we separate it from the > event. > > > > > > > > > > > > For @Panagiotis, > > > > > > I think `ioExecutor` make sense to me and I have added it in > > > > > > `ContextModificationContext`, thanks > > > > > > > > > > > &g
Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener
Hi, To @Jing Ge > Thanks for the clarification. Just out of curiosity, if the context is not part of the event, why should it be the input parameter of each onEvent call? I think it's quite strange to put some information in an Event, such as a factory identifier for catalog, but they will be used by the listener. I place it in the context class and I think it is more suitable than directly placing it in the event class. To @Mason > 1. I'm also curious about default implementations. Would atlas/datahub be supported by default? We won't do that and external systems such as atlas/datahub need to implement the listener themselves. > 2. The FLIP title is confusing to me, especially in distinguishing it from FLIP-314. Would a better FLIP title be "Support Catalog Metadata Listener" or something alike? Thanks, I think "Support Catalog Modification Listener" will be more suitable, I'll update the title to it. Best, Shammon FY On Thu, Jun 8, 2023 at 12:25 PM Mason Chen wrote: > Hi Shammon, > > FLIP generally looks good and I'm excited to see this feature. > > 1. I'm also curious about default implementations. Would atlas/datahub be > supported by default? > 2. The FLIP title is confusing to me, especially in distinguishing it from > FLIP-314. Would a better FLIP title be "Support Catalog Metadata Listener" > or something alike? > > Best, > Mason > > On Tue, Jun 6, 2023 at 3:33 AM Jing Ge wrote: > > > Hi Shammon, > > > > Thanks for the clarification. Just out of curiosity, if the context is > not > > part of the event, why should it be the input parameter of each onEvent > > call? > > > > Best regards, > > Jing > > > > On Tue, Jun 6, 2023 at 11:58 AM Leonard Xu wrote: > > > > > Thanks Shammon for the timely update, the updated FLIP looks good to > me. > > > > > > Hope to see the vote thread and following FLIP-314 discussion thread. > > > > > > Best, > > > Leonard > > > > > > > On Jun 6, 2023, at 5:04 PM, Shammon FY wrote: > > > > > > > > Hi, > > > > > > > > Thanks for all the feedback. > > > > > > > > For @Jing Ge, > > > > I forget to update the demo code in the FLIP, the method is > > > > `onEvent(CatalogModificationEvent, CatalogModificationContext)` and > > there > > > > is no `onEvent(CatalogModificationEvent)`. I have updated the code. > > > Context > > > > contains some additional information that is not part of an Event, > but > > > > needs to be used in the listener, so we separate it from the event. > > > > > > > > For @Panagiotis, > > > > I think `ioExecutor` make sense to me and I have added it in > > > > `ContextModificationContext`, thanks > > > > > > > > For @Leonard, > > > > Thanks for your input. > > > > 1. I have updated `CatalogModificationContext` as an interface, as > well > > > as > > > > Context in CatalogModificationListenerFactory > > > > 2. Configuration sounds good to me, I have updated the method name > and > > > > getConfiguration in Context > > > > > > > > For @David, > > > > Yes, you're right. The listener will only be used on the client side > > and > > > > won't introduce a new code path for running per-job/per-session jobs. > > The > > > > listener will be created in `TableEnvironment` and `SqlGateway` which > > > can a > > > > `CatalogManager` with the listener. > > > > > > > > > > > > Best, > > > > Shammon FY > > > > > > > > > > > > On Tue, Jun 6, 2023 at 3:33 PM David Morávek < > david.mora...@gmail.com> > > > > wrote: > > > > > > > >> Hi, > > > >> > > > >> Thanks for the FLIP! Data lineage is an important problem to tackle. > > > >> > > > >> Can you please expand on how this is planned to be wired into the > > > >> JobManager? As I understand, the listeners will be configured > globally > > > (per > > > >> cluster), so this won't introduce a new code path for running > per-job > > / > > > >> per-session user code. Is that correct? > > > >> > > > >> Best, > > > >> D > > > >> > > > >> On Tue, Jun 6, 2023 at 9:17 AM Leonard Xu > wrote: > > > >> > > > >>> Thanks Shammon for driving this FLIP
[DISCUSS] FLIP-314: Support Customized Job Lineage Listener
Hi devs, I would like to start a discussion on FLIP-314: Support Customized Job Lineage Listener[1] which is the next stage of FLIP-294 [2]. Flink streaming and batch jobs create lineage dependency between source and sink, users can manage their data and jobs according to this lineage information. For example, when there is a delay in Flink ETL or data, users can easily trace the problematic jobs and affected data. On the other hand, when users need to correct data or debug, they can perform operations based on lineage too. In FLIP-314 we want to introduce lineage related interfaces for Flink and users can create customized job status listeners. When job status changes, users can get job status and information to add, update or delete lineage. Looking forward to your feedback, thanks. [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-314%3A+Support+Customized+Job+Lineage+Listener [2] https://cwiki.apache.org/confluence/display/FLINK/FLIP-294%3A+Support+Customized+Job+Meta+Data+Listener Best, Shammon FY
Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener
Hi, Thanks for all the feedback. For @Jing Ge, I forget to update the demo code in the FLIP, the method is `onEvent(CatalogModificationEvent, CatalogModificationContext)` and there is no `onEvent(CatalogModificationEvent)`. I have updated the code. Context contains some additional information that is not part of an Event, but needs to be used in the listener, so we separate it from the event. For @Panagiotis, I think `ioExecutor` make sense to me and I have added it in `ContextModificationContext`, thanks For @Leonard, Thanks for your input. 1. I have updated `CatalogModificationContext` as an interface, as well as Context in CatalogModificationListenerFactory 2. Configuration sounds good to me, I have updated the method name and getConfiguration in Context For @David, Yes, you're right. The listener will only be used on the client side and won't introduce a new code path for running per-job/per-session jobs. The listener will be created in `TableEnvironment` and `SqlGateway` which can a `CatalogManager` with the listener. Best, Shammon FY On Tue, Jun 6, 2023 at 3:33 PM David Morávek wrote: > Hi, > > Thanks for the FLIP! Data lineage is an important problem to tackle. > > Can you please expand on how this is planned to be wired into the > JobManager? As I understand, the listeners will be configured globally (per > cluster), so this won't introduce a new code path for running per-job / > per-session user code. Is that correct? > > Best, > D > > On Tue, Jun 6, 2023 at 9:17 AM Leonard Xu wrote: > > > Thanks Shammon for driving this FLIP forward, I’ve several comments about > > the updated FLIP. > > > > 1. CatalogModificationContext is introduced as a class instead of an > > interface, is it a typo? > > > > 2. The FLIP defined multiple Map config(); methods in > > some Context classes, Could we use Configuration > getConfiguration()；Class > > org.apache.flink.configuration.Configuration is recommend as it’s public > > API and offers more useful methods as well. > > > > 3. The Context of CatalogModificationListenerFactory should be an > > interface too, and getUserClassLoder() > > would be more aligned with flink’s naming style. > > > > > > Best, > > Leonard > > > > > On May 26, 2023, at 4:08 PM, Shammon FY wrote: > > > > > > Hi devs, > > > > > > We would like to bring up a discussion about FLIP-294: Support > Customized > > > Job Meta Data Listener[1]. We have had several discussions with Jark > Wu, > > > Leonard Xu, Dong Lin, Qingsheng Ren and Poorvank about the functions > and > > > interfaces, and thanks for their valuable advice. > > > The overall job and connector information is divided into metadata and > > > lineage, this FLIP focuses on metadata and lineage will be discussed in > > > another FLIP in the future. In this FLIP we want to add a customized > > > listener in Flink to report catalog modifications to external metadata > > > systems such as datahub[2] or atlas[3]. Users can view the specific > > > information of connectors such as source and sink for Flink jobs in > these > > > systems, including fields, watermarks, partitions, etc. > > > > > > Looking forward to hearing from you, thanks. > > > > > > > > > [1] > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-294%3A+Support+Customized+Job+Meta+Data+Listener > > > [2] https://datahub.io/ > > > [3] https://atlas.apache.org/#/ > > > > >
Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener
Hi devs: Thanks for all the feedback, and if there are no more comments, I will start a vote on FLIP-294 [1] later. Thanks again. [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-294%3A+Support+Customized+Job+Meta+Data+Listener Best, Shammon FY On Tue, Jun 6, 2023 at 1:40 PM Shammon FY wrote: > Hi Martijn, > > Thanks for your attention, I will soon initiate a discussion about > FLIP-314. > > Best, > Shammon FY > > > On Fri, Jun 2, 2023 at 2:55 AM Martijn Visser > wrote: > >> Hi Shammon, >> >> Just wanted to chip-in that I like the overall FLIP. Will be interesting >> to >> see the follow-up discussion on FLIP-314. >> >> Best regards, >> >> Martijn >> >> On Thu, Jun 1, 2023 at 5:45 AM yuxia wrote: >> >> > Thanks for explanation. Make sense to me. >> > >> > Best regards, >> > Yuxia >> > >> > - 原始邮件 - >> > 发件人: "Shammon FY" >> > 收件人: "dev" >> > 发送时间: 星期四, 2023年 6 月 01日 上午 10:45:12 >> > 主题: Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener >> > >> > Thanks yuxia, you're right and I'll add the new database to >> > AlterDatabaseEvent. >> > >> > I added `ignoreIfNotExists` for AlterDatabaseEvent because it is a >> > parameter in the `Catalog.alterDatabase` method. Although this value is >> > currently always false in `AlterDatabaseOperation`, I think it's better >> > to stay consistent with `Catalog.alterDatabase`. What do you think? >> > >> > Best, >> > Shammon FY >> > >> > On Thu, Jun 1, 2023 at 10:25 AM yuxia >> wrote: >> > >> > > Hi, Shammon. >> > > I mean do we need to contain the new database after alter in >> > > AlterDatabaseEvent? So that the listener can know what has been >> modified >> > > for the database. Or the listener don't need to care about the actual >> > > modification. >> > > Also, I'm wondering whether AlterDatabaseEven need to include >> > > ignoreIfNotExists method since alter database operation don't have >> such >> > > syntax like 'alter database if exists xxx'. >> > > >> > > Best regards, >> > > Yuxia >> > > >> > > - 原始邮件 - >> > > 发件人: "Shammon FY" >> > > 收件人: "dev" >> > > 发送时间: 星期三, 2023年 5 月 31日 下午 2:55:26 >> > > 主题: Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener >> > > >> > > Hi yuxia >> > > >> > > Thanks for your input. The `AlterDatabaseEvent` extends >> > > `DatabaseModificationEvent` which has the original database. >> > > >> > > Best, >> > > Shammon FY >> > > >> > > On Wed, May 31, 2023 at 2:24 PM yuxia >> > wrote: >> > > >> > > > Thanks Shammon for driving it. >> > > > The FLIP generally looks good to me. I only have one question. >> > > > WRT AlterDatabaseEvent, IIUC, it'll contain the origin database name >> > and >> > > > the new CatalogDatabase after modified. Is it enough only pass the >> > origin >> > > > database name? Will it be better to contain the origin >> CatalogDatabase >> > so >> > > > that listener have ways to know what changes? >> > > > >> > > > Best regards, >> > > > Yuxia >> > > > >> > > > - 原始邮件 - >> > > > 发件人: "ron9 liu" >> > > > 收件人: "dev" >> > > > 发送时间: 星期三, 2023年 5 月 31日 上午 11:36:04 >> > > > 主题: Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data >> Listener >> > > > >> > > > Hi, Shammon >> > > > >> > > > Thanks for driving this FLIP, It will enforce the Flink metadata >> > > capability >> > > > from the platform produce perspective. The overall design looks >> good to >> > > me, >> > > > I just have some small question: >> > > > 1. Regarding CatalogModificationListenerFactory#createListener >> method, >> > I >> > > > think it would be better to pass Context as its parameter instead of >> > two >> > > > specific Object. In this way, we can easily extend it in the future >> and >> > > > there will be no compatibility problems. Refer to >> > > > >&
Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener
Hi Martijn, Thanks for your attention, I will soon initiate a discussion about FLIP-314. Best, Shammon FY On Fri, Jun 2, 2023 at 2:55 AM Martijn Visser wrote: > Hi Shammon, > > Just wanted to chip-in that I like the overall FLIP. Will be interesting to > see the follow-up discussion on FLIP-314. > > Best regards, > > Martijn > > On Thu, Jun 1, 2023 at 5:45 AM yuxia wrote: > > > Thanks for explanation. Make sense to me. > > > > Best regards, > > Yuxia > > > > - 原始邮件 - > > 发件人: "Shammon FY" > > 收件人: "dev" > > 发送时间: 星期四, 2023年 6 月 01日 上午 10:45:12 > > 主题: Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener > > > > Thanks yuxia, you're right and I'll add the new database to > > AlterDatabaseEvent. > > > > I added `ignoreIfNotExists` for AlterDatabaseEvent because it is a > > parameter in the `Catalog.alterDatabase` method. Although this value is > > currently always false in `AlterDatabaseOperation`, I think it's better > > to stay consistent with `Catalog.alterDatabase`. What do you think? > > > > Best, > > Shammon FY > > > > On Thu, Jun 1, 2023 at 10:25 AM yuxia > wrote: > > > > > Hi, Shammon. > > > I mean do we need to contain the new database after alter in > > > AlterDatabaseEvent? So that the listener can know what has been > modified > > > for the database. Or the listener don't need to care about the actual > > > modification. > > > Also, I'm wondering whether AlterDatabaseEven need to include > > > ignoreIfNotExists method since alter database operation don't have such > > > syntax like 'alter database if exists xxx'. > > > > > > Best regards, > > > Yuxia > > > > > > - 原始邮件 - > > > 发件人: "Shammon FY" > > > 收件人: "dev" > > > 发送时间: 星期三, 2023年 5 月 31日 下午 2:55:26 > > > 主题: Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener > > > > > > Hi yuxia > > > > > > Thanks for your input. The `AlterDatabaseEvent` extends > > > `DatabaseModificationEvent` which has the original database. > > > > > > Best, > > > Shammon FY > > > > > > On Wed, May 31, 2023 at 2:24 PM yuxia > > wrote: > > > > > > > Thanks Shammon for driving it. > > > > The FLIP generally looks good to me. I only have one question. > > > > WRT AlterDatabaseEvent, IIUC, it'll contain the origin database name > > and > > > > the new CatalogDatabase after modified. Is it enough only pass the > > origin > > > > database name? Will it be better to contain the origin > CatalogDatabase > > so > > > > that listener have ways to know what changes? > > > > > > > > Best regards, > > > > Yuxia > > > > > > > > - 原始邮件 - > > > > 发件人: "ron9 liu" > > > > 收件人: "dev" > > > > 发送时间: 星期三, 2023年 5 月 31日 上午 11:36:04 > > > > 主题: Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener > > > > > > > > Hi, Shammon > > > > > > > > Thanks for driving this FLIP, It will enforce the Flink metadata > > > capability > > > > from the platform produce perspective. The overall design looks good > to > > > me, > > > > I just have some small question: > > > > 1. Regarding CatalogModificationListenerFactory#createListener > method, > > I > > > > think it would be better to pass Context as its parameter instead of > > two > > > > specific Object. In this way, we can easily extend it in the future > and > > > > there will be no compatibility problems. Refer to > > > > > > > > > > > > > > https://github.com/apache/flink/blob/9880ba5324d4a1252d6ae1a3f0f061e4469a05ac/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/DynamicTableFactory.java#L81 > > > > 2. In FLIP, you mentioned that multiple Flink tables may refer to the > > > same > > > > physical table, so does the Listener report this physical table > > > repeatedly？ > > > > 3. When registering a Listener object, will it connect to an external > > > > system such as Datahub? If the Listener object registration times out > > due > > > > to permission issues, it will affect the execution of all subsequent > > SQL, > > > > what should we do in this case?
Re: [DISCUSS] FLIP-308: Support Time Travel In Batch Mode
Hi Feng,
I have one minor comment about the public interface `Optional
getSnapshot()` in the `CatalogTable`.
As we can get tables from the new method `Catalog.getTable(ObjectPath
tablePath, long timestamp)`, I think the returned `CatalogBaseTable` will
have the information of timestamp. Flink or connector such as
iceberg/paimon can create sources from the `CatalogBaseTable` directly
without the need to get the snapshot ID from `CatalogTable.getSnapshot()`.
What do you think of it?
Best,
Shammon FY
On Thu, Jun 1, 2023 at 7:22 AM Jing Ge wrote:
> Hi Feng,
>
> Thanks for the proposal! Very interesting feature. Would you like to update
> your thoughts described in your previous email about why SupportsTimeTravel
> has been rejected into the FLIP? This will help readers understand the
> context (in the future).
>
> Since we always directly add overload methods into Catalog according to new
> requirements, which makes the interface bloated. Just out of curiosity,
> does it make sense to introduce some DSL design? Like
> Catalog.getTable(tablePath).on(timeStamp),
> Catalog.getTable(tablePath).current() for the most current version, and
> more room for further extension like timestamp range, etc. I haven't read
> all the source code yet and I'm not sure if it is possible. But a
> design like this will keep the Catalog API lean and the API/DSL will be
> self described and easier to use.
>
> Best regards,
> Jing
>
>
> On Wed, May 31, 2023 at 12:08 PM Krzysztof Chmielewski <
> krzysiek.chmielew...@gmail.com> wrote:
>
> > Ok after second though I'm retracting my previous statement about Catalog
> > changes you proposed.
> > I do see a benefit for Delta connector actually with this change and see
> > why this could be coupled with Catalog.
> >
> > Delta Connector SQL support, also ships a Delta Catalog implementation
> for
> > Flink.
> > For Delta Catalog, table schema information is fetched from underlying
> > _delta_log and not stored in metastore. For time travel we actually had a
> > problem, that if we would like to timetravel back to some old version,
> > where schema was slightly different, then we would have a conflict since
> > Catalog would return current schema and not how it was for version X.
> >
> > With your change, our Delta Catalog can actually fetch schema for
> version X
> > and send it to DeltaTableFactory. Currency, Catalog can fetch only
> current
> > version. What we would also need however is version (number/timestamp)
> for
> > this table passed to DynamicTableFactory so we could properly set Delta
> > standalone library.
> >
> > Regards,
> > Krzysztof
> >
> > śr., 31 maj 2023 o 10:37 Krzysztof Chmielewski <
> > krzysiek.chmielew...@gmail.com> napisał(a):
> >
> > > Hi,
> > > happy to see such a feature.
> > > Small note from my end regarding Catalog changes.
> > >
> > > TL;DR
> > > I don't think it is necessary to delegate this feature to the catalog.
> I
> > > think that since "timetravel" is per job/query property, its should not
> > be
> > > coupled with the Catalog or table definition. In my opinion this is
> > > something that DynamicTableFactory only has to know about. I would
> rather
> > > see this feature as it is - SQL syntax enhancement but delegate clearly
> > to
> > > DynamicTableFactory.
> > >
> > > I've implemented timetravel feature for Delta Connector [1] using
> > > current Flink API.
> > > Docs are pending code review, but you can find them here [2] and
> examples
> > > are available here [3]
> > >
> > > The timetravel feature that I've implemented is based on Flink Query
> > > hints.
> > > "SELECT * FROM sourceTable /*+ OPTIONS('versionAsOf' = '1') */"
> > >
> > > The " versionAsOf" (we also have 'timestampAsOf') parameter is handled
> > not
> > > by Catalog but by DyntamicTableFactory implementation for Delta
> > connector.
> > > The value of this property is passed to Delta standalone lib API that
> > > returns table view for given version.
> > >
> > > I'm not sure how/if proposed change could benefit Delta connector
> > > implementation for this feature.
> > >
> > > Thanks,
> > > Krzysztof
> > >
> > > [1]
> > >
> >
> https://github.com/delta-io/connectors/tree/flink_table_catalog_feature_branch/flink
> > > [2] https://github.com/kristoffSC/connectors/tree/FlinkSQL_PR_15-docs
> > > [3]
> > >
> >
> https://
Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener
Thanks yuxia, you're right and I'll add the new database to AlterDatabaseEvent. I added `ignoreIfNotExists` for AlterDatabaseEvent because it is a parameter in the `Catalog.alterDatabase` method. Although this value is currently always false in `AlterDatabaseOperation`, I think it's better to stay consistent with `Catalog.alterDatabase`. What do you think? Best, Shammon FY On Thu, Jun 1, 2023 at 10:25 AM yuxia wrote: > Hi, Shammon. > I mean do we need to contain the new database after alter in > AlterDatabaseEvent? So that the listener can know what has been modified > for the database. Or the listener don't need to care about the actual > modification. > Also, I'm wondering whether AlterDatabaseEven need to include > ignoreIfNotExists method since alter database operation don't have such > syntax like 'alter database if exists xxx'. > > Best regards, > Yuxia > > - 原始邮件 - > 发件人: "Shammon FY" > 收件人: "dev" > 发送时间: 星期三, 2023年 5 月 31日 下午 2:55:26 > 主题: Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener > > Hi yuxia > > Thanks for your input. The `AlterDatabaseEvent` extends > `DatabaseModificationEvent` which has the original database. > > Best, > Shammon FY > > On Wed, May 31, 2023 at 2:24 PM yuxia wrote: > > > Thanks Shammon for driving it. > > The FLIP generally looks good to me. I only have one question. > > WRT AlterDatabaseEvent, IIUC, it'll contain the origin database name and > > the new CatalogDatabase after modified. Is it enough only pass the origin > > database name? Will it be better to contain the origin CatalogDatabase so > > that listener have ways to know what changes? > > > > Best regards, > > Yuxia > > > > - 原始邮件 - > > 发件人: "ron9 liu" > > 收件人: "dev" > > 发送时间: 星期三, 2023年 5 月 31日 上午 11:36:04 > > 主题: Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener > > > > Hi, Shammon > > > > Thanks for driving this FLIP, It will enforce the Flink metadata > capability > > from the platform produce perspective. The overall design looks good to > me, > > I just have some small question: > > 1. Regarding CatalogModificationListenerFactory#createListener method, I > > think it would be better to pass Context as its parameter instead of two > > specific Object. In this way, we can easily extend it in the future and > > there will be no compatibility problems. Refer to > > > > > https://github.com/apache/flink/blob/9880ba5324d4a1252d6ae1a3f0f061e4469a05ac/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/DynamicTableFactory.java#L81 > > 2. In FLIP, you mentioned that multiple Flink tables may refer to the > same > > physical table, so does the Listener report this physical table > repeatedly？ > > 3. When registering a Listener object, will it connect to an external > > system such as Datahub? If the Listener object registration times out due > > to permission issues, it will affect the execution of all subsequent SQL, > > what should we do in this case? > > > > Best, > > Ron > > > > Shammon FY 于2023年5月31日周三 08:53写道： > > > > > Thanks Feng, the catalog modification listener is only used to report > > > read-only ddl information to other components or systems. > > > > > > > 1. Will an exception thrown by the listener affect the normal > execution > > > process? > > > > > > Users need to handle the exception in the listener themselves. Many > DDLs > > > such as drop tables and alter tables cannot be rolled back, Flink > cannot > > > handle these exceptions for the listener. It will cause the operation > to > > > exit if an exception is thrown, but the executed DDL will be > successful. > > > > > > > 2. What is the order of execution? Is the listener executed first or > > are > > > specific operations executed first? If I want to perform DDL > permission > > > verification(such as integrating with Ranger based on the listener) , > is > > > that possible? > > > > > > The listener will be notified to report catalog modification after DDLs > > are > > > successful, so you can not do permission verification for DDL in the > > > listener. As mentioned above, Flink will not roll back the DDL even > when > > > the listener throws an exception. I think permission verification is > > > another issue and can be discussed separately. > > > > > > > > > Best, > > > Shammon FY > > > > > > On
Re: [DISCUSS] FLIP 295: Support persistence of Catalog configuration and asynchronous registration
Hi feng, Thanks for updating, I have some minor comments 1. The modification of `CatalogManager` should not be in `Public Interfaces`, it is not a public interface. 2. `@PublicEvolving` should be added for `CatalogStore` and `CatalogStoreFactory` 3. The code `Optional optionalDescriptor = catalogStore.get(catalogName);` in the `CatalogManager` should be `Optional optionalDescriptor = catalogStore.get(catalogName);` Best, Shammon FY On Wed, May 31, 2023 at 2:24 PM liu ron wrote: > Hi, Feng > > Thanks for driving this FLIP, this proposal is very useful for catalog > management. > I have some small questions: > > 1. Regarding the CatalogStoreFactory#createCatalogStore method, do we need > to provide a default implementation? > 2. If we get Catalog from CatalogStore, after initializing it, whether we > put it to Map catalogs again? > 3. Regarding the options `sql.catalog.store.type` and > `sql.catalog.store.file.path`, how about renaming them to > `catalog.store.type` and `catalog.store.path`? > > Best, > Ron > > Feng Jin 于2023年5月29日周一 21:19写道： > > > Hi yuxia > > > > > But from the code in Proposed Changes, once we register the Catalog, > we > > initialize it and open it. right? > > > > Yes, In order to avoid inconsistent semantics of the original CREATE > > CATALOG DDL, Catalog will be directly initialized in registerCatalog so > > that parameter validation can be performed. > > > > In the current design, lazy initialization is mainly reflected in > > getCatalog. If CatalogStore has already saved some catalog > configurations, > > only initialization is required in getCatalog. > > > > > > Best, > > Feng > > > > On Mon, May 29, 2023 at 8:27 PM yuxia > wrote: > > > > > Hi, Feng. > > > I'm trying to understanding the meaning of *lazy initialization*. If > i'm > > > wrong, please correct me. > > > > > > IIUC, lazy initialization means only you need to access the catalog, > then > > > you initialize it. But from the code in Proposed Changes, once we > > register > > > the Catalog, > > > we initialize it and open it. right? > > > > > > Best regards, > > > Yuxia > > > > > > - 原始邮件 - > > > 发件人: "Jing Ge" > > > 收件人: "dev" > > > 发送时间: 星期一, 2023年 5 月 29日 下午 5:12:46 > > > 主题: Re: [DISCUSS] FLIP 295: Support persistence of Catalog > configuration > > > and asynchronous registration > > > > > > Hi Feng, > > > > > > Thanks for your effort! +1 for the proposal. > > > > > > One of the major changes is that current design will provide > > > Map catalogs as a snapshot instead of a cache, which > > means > > > once it has been initialized, any changes done by other sessions will > not > > > affect it. Point 6 described follow-up options for further improvement. > > > > > > Best regards, > > > Jing > > > > > > On Mon, May 29, 2023 at 5:31 AM Feng Jin > wrote: > > > > > > > Hi all, I would like to update you on the latest progress of the > FLIP. > > > > > > > > > > > > Last week, Leonard Xu, HangRuan, Jing Ge, Shammon FY, ShengKai Fang > > and I > > > > had an offline discussion regarding the overall solution for Flink > > > > CatalogStore. We have reached a consensus and I have updated the > final > > > > solution in FLIP. > > > > > > > > Next, let me briefly describe the entire design: > > > > > > > >1. > > > > > > > >Introduce CatalogDescriptor to store catalog configuration similar > > to > > > >TableDescriptor. > > > >2. > > > > > > > >The two key functions of CatalogStore - void storeCatalog(String > > > >catalogName, CatalogDescriptor) and CatalogDescriptor > > > getCatalog(String) > > > >will both use CatalogDescriptor instead of Catalog instance. This > > way, > > > >CatalogStore will only be responsible for saving and retrieving > > > catalog > > > >configurations without having to initialize catalogs. > > > >3. > > > > > > > >The default registerCatalog(String catalogName, Catalog catalog) > > > >function in CatalogManager will be marked as deprecated. > > > >4. > > > > > > > >A new function registerCatalog(String catalogName
Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener
Hi yuxia Thanks for your input. The `AlterDatabaseEvent` extends `DatabaseModificationEvent` which has the original database. Best, Shammon FY On Wed, May 31, 2023 at 2:24 PM yuxia wrote: > Thanks Shammon for driving it. > The FLIP generally looks good to me. I only have one question. > WRT AlterDatabaseEvent, IIUC, it'll contain the origin database name and > the new CatalogDatabase after modified. Is it enough only pass the origin > database name? Will it be better to contain the origin CatalogDatabase so > that listener have ways to know what changes? > > Best regards, > Yuxia > > - 原始邮件 - > 发件人: "ron9 liu" > 收件人: "dev" > 发送时间: 星期三, 2023年 5 月 31日 上午 11:36:04 > 主题: Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener > > Hi, Shammon > > Thanks for driving this FLIP, It will enforce the Flink metadata capability > from the platform produce perspective. The overall design looks good to me, > I just have some small question: > 1. Regarding CatalogModificationListenerFactory#createListener method, I > think it would be better to pass Context as its parameter instead of two > specific Object. In this way, we can easily extend it in the future and > there will be no compatibility problems. Refer to > > https://github.com/apache/flink/blob/9880ba5324d4a1252d6ae1a3f0f061e4469a05ac/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/DynamicTableFactory.java#L81 > 2. In FLIP, you mentioned that multiple Flink tables may refer to the same > physical table, so does the Listener report this physical table repeatedly？ > 3. When registering a Listener object, will it connect to an external > system such as Datahub? If the Listener object registration times out due > to permission issues, it will affect the execution of all subsequent SQL, > what should we do in this case? > > Best, > Ron > > Shammon FY 于2023年5月31日周三 08:53写道： > > > Thanks Feng, the catalog modification listener is only used to report > > read-only ddl information to other components or systems. > > > > > 1. Will an exception thrown by the listener affect the normal execution > > process? > > > > Users need to handle the exception in the listener themselves. Many DDLs > > such as drop tables and alter tables cannot be rolled back, Flink cannot > > handle these exceptions for the listener. It will cause the operation to > > exit if an exception is thrown, but the executed DDL will be successful. > > > > > 2. What is the order of execution? Is the listener executed first or > are > > specific operations executed first? If I want to perform DDL permission > > verification(such as integrating with Ranger based on the listener) , is > > that possible? > > > > The listener will be notified to report catalog modification after DDLs > are > > successful, so you can not do permission verification for DDL in the > > listener. As mentioned above, Flink will not roll back the DDL even when > > the listener throws an exception. I think permission verification is > > another issue and can be discussed separately. > > > > > > Best, > > Shammon FY > > > > On Tue, May 30, 2023 at 1:07 AM Feng Jin wrote: > > > > > Hi, Shammon > > > > > > Thanks for driving this Flip, [Support Customized Job Meta Data > Listener] > > > will make it easier for Flink to collect lineage information. > > > I fully agree with the overall solution and have a small question: > > > > > > 1. Will an exception thrown by the listener affect the normal execution > > > process? > > > > > > 2. What is the order of execution? Is the listener executed first or > are > > > specific operations executed first? If I want to perform DDL > permission > > > verification(such as integrating with Ranger based on the listener) , > is > > > that possible? > > > > > > > > > Best, > > > Feng > > > > > > On Fri, May 26, 2023 at 4:09 PM Shammon FY wrote: > > > > > > > Hi devs, > > > > > > > > We would like to bring up a discussion about FLIP-294: Support > > Customized > > > > Job Meta Data Listener[1]. We have had several discussions with Jark > > Wu, > > > > Leonard Xu, Dong Lin, Qingsheng Ren and Poorvank about the functions > > and > > > > interfaces, and thanks for their valuable advice. > > > > The overall job and connector information is divided into metadata > and > > > > lineage, this FLIP focuses on metadata and lineage will be disc
Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener
Hi ron, Thanks for your feedback. > 1. Regarding CatalogModificationListenerFactory#createListener method, I think it would be better to pass Context as its parameter instead of two specific Object. In this way, we can easily extend it in the future and there will be no compatibility problems. It sounds good to me I will add Context in CatalogModificationListenerFactory, thanks > 2. In FLIP, you mentioned that multiple Flink tables may refer to the same physical table, so does the Listener report this physical table repeatedly? Yes, the listeners for different jobs may receive the same physical table, users should check and update the table information based on the identifier. For example, users may create tables on the same kafka topic in different jobs, which will notify listeners for the same kafka topic. > 3. When registering a Listener object, will it connect to an external system such as Datahub? If the Listener object registration times out due to permission issues, it will affect the execution of all subsequent SQL, what should we do in this case? Users should establish connections to external systems when creating a listener as needed, and they should handle the exceptions too. If users fail to create a listener and throw an exception, Flink will throw the exception too. Best, Shammon FY On Wed, May 31, 2023 at 11:36 AM liu ron wrote: > Hi, Shammon > > Thanks for driving this FLIP, It will enforce the Flink metadata capability > from the platform produce perspective. The overall design looks good to me, > I just have some small question: > 1. Regarding CatalogModificationListenerFactory#createListener method, I > think it would be better to pass Context as its parameter instead of two > specific Object. In this way, we can easily extend it in the future and > there will be no compatibility problems. Refer to > > https://github.com/apache/flink/blob/9880ba5324d4a1252d6ae1a3f0f061e4469a05ac/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/DynamicTableFactory.java#L81 > 2. In FLIP, you mentioned that multiple Flink tables may refer to the same > physical table, so does the Listener report this physical table repeatedly？ > 3. When registering a Listener object, will it connect to an external > system such as Datahub? If the Listener object registration times out due > to permission issues, it will affect the execution of all subsequent SQL, > what should we do in this case? > > Best, > Ron > > Shammon FY 于2023年5月31日周三 08:53写道： > > > Thanks Feng, the catalog modification listener is only used to report > > read-only ddl information to other components or systems. > > > > > 1. Will an exception thrown by the listener affect the normal execution > > process? > > > > Users need to handle the exception in the listener themselves. Many DDLs > > such as drop tables and alter tables cannot be rolled back, Flink cannot > > handle these exceptions for the listener. It will cause the operation to > > exit if an exception is thrown, but the executed DDL will be successful. > > > > > 2. What is the order of execution? Is the listener executed first or > are > > specific operations executed first? If I want to perform DDL permission > > verification(such as integrating with Ranger based on the listener) , is > > that possible? > > > > The listener will be notified to report catalog modification after DDLs > are > > successful, so you can not do permission verification for DDL in the > > listener. As mentioned above, Flink will not roll back the DDL even when > > the listener throws an exception. I think permission verification is > > another issue and can be discussed separately. > > > > > > Best, > > Shammon FY > > > > On Tue, May 30, 2023 at 1:07 AM Feng Jin wrote: > > > > > Hi, Shammon > > > > > > Thanks for driving this Flip, [Support Customized Job Meta Data > Listener] > > > will make it easier for Flink to collect lineage information. > > > I fully agree with the overall solution and have a small question: > > > > > > 1. Will an exception thrown by the listener affect the normal execution > > > process? > > > > > > 2. What is the order of execution? Is the listener executed first or > are > > > specific operations executed first? If I want to perform DDL > permission > > > verification(such as integrating with Ranger based on the listener) , > is > > > that possible? > > > > > > > > > Best, > > > Feng > > > > > > On Fri, May 26, 2023 at 4:09 PM Shammon FY wrote: > > > > > > > Hi devs, > > > &g
Re: [DISCUSS] FLIP-316: Introduce SQL Driver
Thanks Paul for driving this proposal. I found the sql driver has no config related options. If I understand correctly, the sql driver can be used to submit sql jobs in a 'job submission service' such as sql-gateway. In general, in addition to the default config for Flink cluster which includes k8s, ha and .etc, users may also specify configurations for SQL jobs, including parallelism, number of Task Managers, etc. For example, in sql-gateway users can `set` dynamic parameters before submitting a sql statement, and for sql files users may put their configurations in a `yaml` file. Should sql driver support dynamic parameters and specify config file? Best, Shammon FY On Tue, May 30, 2023 at 9:57 PM Weihua Hu wrote: > Thanks Paul for the proposal. > > +1 for this. It is valuable in improving ease of use. > > I have a few questions. > - Is SQLRunner the better name? We use this to run a SQL Job. (Not strong, > the SQLDriver is fine for me) > - Could we run SQL jobs using SQL in strings? Otherwise, we need to prepare > a SQL file in an image for Kubernetes application mode, which may be a bit > cumbersome. > - I noticed that we don't specify the SQLDriver jar in the > "run-application" > command. Does that mean we need to perform automatic detection in Flink? > > > Best, > Weihua > > > On Mon, May 29, 2023 at 7:24 PM Paul Lam wrote: > > > Hi team, > > > > I’d like to start a discussion about FLIP-316 [1], which introduces a SQL > > driver as the > > default main class for Flink SQL jobs. > > > > Currently, Flink SQL could be executed out of the box either via SQL > > Client/Gateway > > or embedded in a Flink Java/Python program. > > > > However, each one has its drawback: > > > > - SQL Client/Gateway doesn’t support the application deployment mode [2] > > - Flink Java/Python program requires extra work to write a non-SQL > program > > > > Therefore, I propose adding a SQL driver to act as the default main class > > for SQL jobs. > > Please see the FLIP docs for details and feel free to comment. Thanks! > > > > [1] > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-316%3A+Introduce+SQL+Driver > > < > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-316:+Introduce+SQL+Driver > > > > > [2] https://issues.apache.org/jira/browse/FLINK-26541 < > > https://issues.apache.org/jira/browse/FLINK-26541> > > > > Best, > > Paul Lam >
Re: [DISCUSS] FLIP-294: Support Customized Job Meta Data Listener
Thanks Feng, the catalog modification listener is only used to report read-only ddl information to other components or systems. > 1. Will an exception thrown by the listener affect the normal execution process? Users need to handle the exception in the listener themselves. Many DDLs such as drop tables and alter tables cannot be rolled back, Flink cannot handle these exceptions for the listener. It will cause the operation to exit if an exception is thrown, but the executed DDL will be successful. > 2. What is the order of execution? Is the listener executed first or are specific operations executed first? If I want to perform DDL permission verification(such as integrating with Ranger based on the listener) , is that possible? The listener will be notified to report catalog modification after DDLs are successful, so you can not do permission verification for DDL in the listener. As mentioned above, Flink will not roll back the DDL even when the listener throws an exception. I think permission verification is another issue and can be discussed separately. Best, Shammon FY On Tue, May 30, 2023 at 1:07 AM Feng Jin wrote: > Hi, Shammon > > Thanks for driving this Flip, [Support Customized Job Meta Data Listener] > will make it easier for Flink to collect lineage information. > I fully agree with the overall solution and have a small question: > > 1. Will an exception thrown by the listener affect the normal execution > process? > > 2. What is the order of execution? Is the listener executed first or are > specific operations executed first? If I want to perform DDL permission > verification(such as integrating with Ranger based on the listener) , is > that possible? > > > Best, > Feng > > On Fri, May 26, 2023 at 4:09 PM Shammon FY wrote: > > > Hi devs, > > > > We would like to bring up a discussion about FLIP-294: Support Customized > > Job Meta Data Listener[1]. We have had several discussions with Jark Wu, > > Leonard Xu, Dong Lin, Qingsheng Ren and Poorvank about the functions and > > interfaces, and thanks for their valuable advice. > > The overall job and connector information is divided into metadata and > > lineage, this FLIP focuses on metadata and lineage will be discussed in > > another FLIP in the future. In this FLIP we want to add a customized > > listener in Flink to report catalog modifications to external metadata > > systems such as datahub[2] or atlas[3]. Users can view the specific > > information of connectors such as source and sink for Flink jobs in these > > systems, including fields, watermarks, partitions, etc. > > > > Looking forward to hearing from you, thanks. > > > > > > [1] > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-294%3A+Support+Customized+Job+Meta+Data+Listener > > [2] https://datahub.io/ > > [3] https://atlas.apache.org/#/ > > >
[DISCUSS] FLIP-294: Support Customized Job Meta Data Listener
Hi devs, We would like to bring up a discussion about FLIP-294: Support Customized Job Meta Data Listener[1]. We have had several discussions with Jark Wu, Leonard Xu, Dong Lin, Qingsheng Ren and Poorvank about the functions and interfaces, and thanks for their valuable advice. The overall job and connector information is divided into metadata and lineage, this FLIP focuses on metadata and lineage will be discussed in another FLIP in the future. In this FLIP we want to add a customized listener in Flink to report catalog modifications to external metadata systems such as datahub[2] or atlas[3]. Users can view the specific information of connectors such as source and sink for Flink jobs in these systems, including fields, watermarks, partitions, etc. Looking forward to hearing from you, thanks. [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-294%3A+Support+Customized+Job+Meta+Data+Listener [2] https://datahub.io/ [3] https://atlas.apache.org/#/
Re: [DISCUSS] FLIP-308: Support Time Travel In Batch Mode
Thanks Feng, the feature of time travel sounds great! In addition to SYSTEM_TIME, lake houses such as paimon and iceberg support snapshot or version. For example, users can query snapshot 1 for paimon by the following statement SELECT * FROM t VERSION AS OF 1 Could we support this in Flink too? Best, Shammon FY On Fri, May 26, 2023 at 1:20 PM Benchao Li wrote: > Regarding the implementation, did you consider the pushdown abilities > compatible, e.g., projection pushdown, filter pushdown, partition pushdown. > Since `Snapshot` is not handled much in existing rules, I have a concern > about this. Of course, it depends on your implementation detail, what is > important is that we'd better add some cross tests for these. > > Regarding the interface exposed to Connector, I see there is a rejected > design for adding SupportsTimeTravel, but I didn't see the alternative in > the FLIP doc. IMO, this is an important thing we need to clarify because we > need to know whether the Connector supports this, and what column/metadata > corresponds to 'system_time'. > > Feng Jin 于2023年5月25日周四 22:50写道： > > > Thanks for your reply > > > > @Timo @BenChao @yuxia > > > > Sorry for the mistake, Currently , calcite only supports `FOR > SYSTEM_TIME > > AS OF ` syntax. We can only support `FOR SYSTEM_TIME AS OF` . I've > > updated the syntax part of the FLIP. > > > > > > @Timo > > > > > We will convert it to TIMESTAMP_LTZ? > > > > Yes, I think we need to convert TIMESTAMP to TIMESTAMP_LTZ and then > convert > > it into a long value. > > > > > How do we want to query the most recent version of a table > > > > I think we can use `AS OF CURRENT_TIMESTAMP` ，But it does cause > > inconsistency with the real-time concept. > > However, from my personal understanding, the scope of `AS OF > > CURRENT_TIMESTAMP` is the table itself, not the table record. So, I > think > > using CURRENT_TIMESTAMP should also be reasonable?. > > Additionally, if no version is specified, the latest version should be > used > > by default. > > > > > > > > Best, > > Feng > > > > > > On Thu, May 25, 2023 at 7:47 PM yuxia > wrote: > > > > > Thanks Feng for bringing this up. It'll be great to introduce time > travel > > > to Flink to have a better integration with external data soruces. > > > > > > I also share same concern about the syntax. > > > I see in the part of `Whether to support other syntax implementations` > in > > > this FLIP, seems the syntax in Calcite should be `FOR SYSTEM_TIME AS > OF`, > > > right? > > > But the the syntax part in this FLIP, it seems to be `AS OF TIMESTAMP` > > > instead of `FOR SYSTEM_TIME AS OF`. Is it just a mistake or by design? > > > > > > > > > Best regards, > > > Yuxia > > > > > > - 原始邮件 - > > > 发件人: "Benchao Li" > > > 收件人: "dev" > > > 发送时间: 星期四, 2023年 5 月 25日 下午 7:27:17 > > > 主题: Re: [DISCUSS] FLIP-308: Support Time Travel In Batch Mode > > > > > > Thanks Feng, it's exciting to have this ability. > > > > > > Regarding the syntax section, are you proposing `AS OF` instead of `FOR > > > SYSTEM AS OF` to do this? I know `FOR SYSTEM AS OF` is in the SQL > > standard > > > and has been supported in some database vendors such as SQL Server. > About > > > `AS OF`, is it in the standard or any database vendor supports this, if > > > yes, I think it's worth to add this support to Calcite, and I would > give > > a > > > hand in Calcite side. Otherwise, I think we'd better to use `FOR SYSTEM > > AS > > > OF`. > > > > > > Timo Walther 于2023年5月25日周四 19:02写道： > > > > > > > Also: How do we want to query the most recent version of a table? > > > > > > > > `AS OF CURRENT_TIMESTAMP` would be ideal, but according to the docs > > both > > > > the type is TIMESTAMP_LTZ and what is even more concerning is the it > > > > actually is evalated row-based: > > > > > > > > > Returns the current SQL timestamp in the local time zone, the > return > > > > type is TIMESTAMP_LTZ(3). It is evaluated for each record in > streaming > > > > mode. But in batch mode, it is evaluated once as the query starts and > > > > uses the same result for every row. > > > > > > > > This could make it difficult to explain in a join scenario of > multiple > > > &
Re: [ANNOUNCE] New Apache Flink PMC Member - Leonard Xu
Congratulations, Leonard! Best, Shammon FY On Sun, Apr 23, 2023 at 5:07 PM Xianxun Ye wrote: > Congratulations, Leonard! > > Best regards, > > Xianxun > > > 2023年4月23日 09:10，Lincoln Lee 写道： > > > > Congratulations, Leonard! > >
Re: [ANNOUNCE] New Apache Flink PMC Member - Qingsheng Ren
Congratulations, Qingsheng! Best, Shammon FY On Sun, Apr 23, 2023 at 4:40 PM Weihua Hu wrote: > Congratulations, Qingsheng! > > Best, > Weihua > > > On Sun, Apr 23, 2023 at 3:53 PM Yun Tang wrote: > > > Congratulations, Qingsheng! > > > > Best > > Yun Tang > > > > From: weijie guo > > Sent: Sunday, April 23, 2023 14:50 > > To: dev@flink.apache.org > > Subject: Re: [ANNOUNCE] New Apache Flink PMC Member - Qingsheng Ren > > > > Congratulations, Qingsheng! > > > > Best regards, > > > > Weijie > > > > > > Geng Biao 于2023年4月23日周日 14:29写道： > > > > > Congrats, Qingsheng! > > > Best, > > > Biao Geng > > > > > > 获取 Outlook for iOS<https://aka.ms/o0ukef> > > > > > > 发件人: Wencong Liu > > > 发送时间: Sunday, April 23, 2023 11:06:39 AM > > > 收件人: dev@flink.apache.org > > > 主题: Re:[ANNOUNCE] New Apache Flink PMC Member - Qingsheng Ren > > > > > > Congratulations, Qingsheng! > > > > > > Best, > > > Wencong LIu > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > At 2023-04-21 19:47:52, "Jark Wu" wrote: > > > >Hi everyone, > > > > > > > >We are thrilled to announce that Leonard Xu has joined the Flink PMC! > > > > > > > >Leonard has been an active member of the Apache Flink community for > many > > > >years and became a committer in Nov 2021. He has been involved in > > various > > > >areas of the project, from code contributions to community building. > His > > > >contributions are mainly focused on Flink SQL and connectors, > especially > > > >leading the flink-cdc-connectors project to receive 3.8+K GitHub > stars. > > He > > > >authored 150+ PRs, and reviewed 250+ PRs, and drove several FLIPs > (e.g., > > > >FLIP-132, FLIP-162). He has participated in plenty of discussions in > the > > > >dev mailing list, answering questions about 500+ threads in the > > > >user/user-zh mailing list. Besides that, he is community minded, such > as > > > >being the release manager of 1.17, verifying releases, managing > release > > > >syncs, etc. > > > > > > > >Congratulations and welcome Leonard! > > > > > > > >Best, > > > >Jark (on behalf of the Flink PMC) > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > At 2023-04-21 19:50:02, "Jark Wu" wrote: > > > >Hi everyone, > > > > > > > >We are thrilled to announce that Qingsheng Ren has joined the Flink > PMC! > > > > > > > >Qingsheng has been contributing to Apache Flink for a long time. He is > > the > > > >core contributor and maintainer of the Kafka connector and > > > >flink-cdc-connectors, bringing users stability and ease of use in both > > > >projects. He drove discussions and implementations in FLIP-221, > > FLIP-288, > > > >and the connector testing framework. He is continuously helping with > the > > > >expansion of the Flink community and has given several talks about > Flink > > > >connectors at many conferences, such as Flink Forward Global and Flink > > > >Forward Asia. Besides that, he is willing to help a lot in the > community > > > >work, such as being the release manager for both 1.17 and 1.18, > > verifying > > > >releases, and answering questions on the mailing list. > > > > > > > >Congratulations and welcome Qingsheng! > > > > > > > >Best, > > > >Jark (on behalf of the Flink PMC) > > > > > >
Re: [VOTE] FLIP-288: Enable Dynamic Partition Discovery by Default in Kafka Source
+1 (non-binding) Best, Shammon FY On Sun, Apr 23, 2023 at 10:35 AM Qingsheng Ren wrote: > Thanks for pushing this FLIP forward, Hongshun! > > +1 (binding) > > Best, > Qingsheng > > On Fri, Apr 21, 2023 at 2:52 PM Hongshun Wang > wrote: > > > Dear Flink Developers, > > > > > > Thank you for providing feedback on FLIP-288: Enable Dynamic Partition > > Discovery by Default in Kafka Source[1] on the discussion thread[2]. > > > > The goal of the FLIP is to enable partition discovery by default and set > > EARLIEST offset strategy for later discovered partitions. > > > > > > I am initiating a vote for this FLIP. The vote will be open for at least > 72 > > hours, unless there is an objection or insufficient votes. > > > > > > [1]: [ > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-288%3A+Enable+Dynamic+Partition+Discovery+by+Default+in+Kafka+Source](https://cwiki.apache.org/confluence/display/FLINK/FLIP-288%3A+Enable+Dynamic+Partition+Discovery+by+Default+in+Kafka+Source) > <https://cwiki.apache.org/confluence/display/FLINK/FLIP-288%3A+Enable+Dynamic+Partition+Discovery+by+Default+in+Kafka+Source%5D(https://cwiki.apache.org/confluence/display/FLINK/FLIP-288%3A+Enable+Dynamic+Partition+Discovery+by+Default+in+Kafka+Source)> > > < > https://cwiki.apache.org/confluence/display/FLINK/FLIP-288%3A+Enable+Dynamic+Partition+Discovery+by+Default+in+Kafka+Source%5D(https://cwiki.apache.org/confluence/display/FLINK/FLIP-288%3A+Enable+Dynamic+Partition+Discovery+by+Default+in+Kafka+Source) > > > > [2]: [ > > > > > https://lists.apache.org/thread/581f2xq5d1tlwc8gcr27gwkp3zp0wrg6](https://lists.apache.org/thread/581f2xq5d1tlwc8gcr27gwkp3zp0wrg6) > > > > > > Best regards, > > Hongshun > > >
Re: [DISCUSS] FLIP-288:Enable Dynamic Partition Discovery by Default in Kafka Source
Hi Hongshun Thanks for your explanation, I have got your point. I review the FLIP again and only have one minor comment which won't block this FLIP: should we need in `OffsetsInitializer newDiscoveryOffsetsInitializer` in the constructor of `KafkaSourceEnumerator`? I think we can remove it if we always use EARLIEST for new discovery partitions. Best, Shammon FY On Tue, Apr 18, 2023 at 4:59 PM Hongshun Wang wrote: > Hi Shammon, > > Thank you for your advice.I have carefully considered whether to show this > in SQL DDL. Therefore, I carefully studied whether it is feasible Recently > > However, after reading the corresponding code more thoroughly, it appears > that SpecifiedOffsetsInitializer and TimestampOffsetsInitializer do not > work as we initially thought. Finally, I have decided to only use > "EARLIEST" instead of allowing the user to make a free choice. > > Now, let me show my new understanding. > > The actual work of SpecifiedOffsetsInitializer and > TimestampOffsetsInitializer: > > >- *SpecifiedOffsetsInitializer*: Use *Specified offset* for specified >partitions while use *EARLIEST* for unspecified partitions. Specified >partitions offset should be less than the latest offset, otherwise it > will >start from the *EARLIEST*. >- *TimestampOffsetsInitializer*: Initialize the offsets based on a >timestamp. If the message meeting the requirement of the timestamp have > not >been produced to Kafka yet, just use the *LATEST* offset. > > So, some problems will occur when new partition use > SpecifiedOffsetsInitializer or TimestampOffsetsInitializer. You can find > more information in the "Rejected Alternatives" section of Flip-288, which > includes details of the code and process of deductive reasoning. > All these problems can be reproducible in the current version. The reason > why they haven't been exposed is probably because users usually set the > existing specified offset or timestamp, so it appears as earliest in > production. > > WDYT? > CC：Ruan, Shammon, Gordon, Leonard and Qingsheng. > > Yours > > Hongshun > > > > > On Fri, Apr 14, 2023 at 5:48 PM Shammon FY wrote: > > > Hi Hongshun > > > > Thanks for updating the FLIP, it totally sounds good to me. > > > > I just have one comment: How does sql job set new discovery offsets > > initializer? > > I found `DataStream` jobs can set different offsets initializers for new > > discovery partitions in `KafkaSourceBuilder.setNewDiscoveryOffsets`. Do > SQL > > jobs need to support this feature? > > > > Best, > > Shammon FY > > > > On Wed, Apr 12, 2023 at 2:27 PM Hongshun Wang > > wrote: > > > > > Hi everyone, > > > > > > I have already modified FLIP-288 to provide a > > > newDiscoveryOffsetsInitializer in the KafkaSourceBuilder and > > > KafkaSourceEnumerator. Users can use > > > KafkaSourceBuilder#setNewDiscoveryOffsets to change the strategy for > new > > > partitions. > > > > > > Surely, enabling the partition discovery strategy by default and > > modifying > > > the offset strategy for new partitions should be brought to the user's > > > attention. Therefore, it will be explained in the 1.18 release notes. > > > > > > WDYT?CC, Ruan, Shammon, Gordon and Leonard. > > > > > > > > > Best, > > > > > > Hongshun > > > > > > On Fri, Mar 31, 2023 at 2:56 PM Hongshun Wang > > > > wrote: > > > > > > > Hi everyone, > > > > Thanks for your participation. > > > > > > > > @Gordon, I looked at the several questions you raised: > > > > > > > >1. Should we use the firstDiscovery flag or two separate > > > >OffsetsInitializers? Actually, I have considered later. If we > follow > > > >my initial idea, we can provide a default earliest > > OffsetsInitializer > > > >for a new partition. However, According to @Shammon's suggestion, > > > different > > > >startup OffsetsInitializers correspond to different post-startup > > > >OffsetsInitializers for Flink's built-in offset strategies. > > > >2. "Future-time" TIMESTAMP OffsetsInitializer. I looked at the > code > > > >again, and it seems that neither @Shammon nor I have figured out . > > > >TimestampOffsetsInitializer#getPartitionOffsets has a comment: > > "First > > > >get the current end offsets of the partitions. This is going to be > > > used in &g
Re: Re: [VOTE] FLIP-302: Support TRUNCATE TABLE statement in batch mode
+1 (no-binding) Best, Shammon FY On Tue, Apr 18, 2023 at 12:56 PM Jacky Lau wrote: > +1 (no-binding) > > Best, > Jacky Lau > > Jingsong Li 于2023年4月18日周二 11:57写道： > > > +1 > > > > On Tue, Apr 18, 2023 at 9:39 AM Aitozi wrote: > > > > > > +1 > > > > > > Best, > > > Aitozi > > > > > > ron 于2023年4月18日周二 09:18写道： > > > > > > > > +1 > > > > > > > > > > > > > -原始邮件- > > > > > 发件人: "Lincoln Lee" > > > > > 发送时间: 2023-04-18 09:08:08 (星期二) > > > > > 收件人: dev@flink.apache.org > > > > > 抄送: > > > > > 主题: Re: [VOTE] FLIP-302: Support TRUNCATE TABLE statement in batch > > mode > > > > > > > > > > +1 (binding) > > > > > > > > > > Best, > > > > > Lincoln Lee > > > > > > > > > > > > > > > yuxia 于2023年4月17日周一 23:54写道： > > > > > > > > > > > Hi all. > > > > > > > > > > > > Thanks for all the feedback on FLIP-302: Support TRUNCATE TABLE > > statement > > > > > > in batch mode [1]. > > > > > > Based on the discussion [2], we have come to a consensus, so I > > would like > > > > > > to start a vote. > > > > > > > > > > > > The vote will last for at least 72 hours unless there is an > > objection or > > > > > > insufficient votes. > > > > > > > > > > > > [1]: > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-302%3A+Support+TRUNCATE+TABLE+statement+in+batch+mode > > > > > > [2]: [ > > https://lists.apache.org/thread/m4r3wrd7p96wdst3nz3ncqzog6kf51cf | > > > > > > https://lists.apache.org/thread/m4r3wrd7p96wdst3nz3ncqzog6kf51cf > ] > > > > > > > > > > > > > > > > > > Best regards, > > > > > > Yuxia > > > > > > > > > > > > > > > > > > -- > > > > Best, > > > > Ron > > >
Re: [DISCUSS] FLIP-288:Enable Dynamic Partition Discovery by Default in Kafka Source
Hi Hongshun Thanks for updating the FLIP, it totally sounds good to me. I just have one comment: How does sql job set new discovery offsets initializer? I found `DataStream` jobs can set different offsets initializers for new discovery partitions in `KafkaSourceBuilder.setNewDiscoveryOffsets`. Do SQL jobs need to support this feature? Best, Shammon FY On Wed, Apr 12, 2023 at 2:27 PM Hongshun Wang wrote: > Hi everyone, > > I have already modified FLIP-288 to provide a > newDiscoveryOffsetsInitializer in the KafkaSourceBuilder and > KafkaSourceEnumerator. Users can use > KafkaSourceBuilder#setNewDiscoveryOffsets to change the strategy for new > partitions. > > Surely, enabling the partition discovery strategy by default and modifying > the offset strategy for new partitions should be brought to the user's > attention. Therefore, it will be explained in the 1.18 release notes. > > WDYT?CC, Ruan, Shammon, Gordon and Leonard. > > > Best, > > Hongshun > > On Fri, Mar 31, 2023 at 2:56 PM Hongshun Wang > wrote: > > > Hi everyone, > > Thanks for your participation. > > > > @Gordon, I looked at the several questions you raised: > > > >1. Should we use the firstDiscovery flag or two separate > >OffsetsInitializers? Actually, I have considered later. If we follow > >my initial idea, we can provide a default earliest OffsetsInitializer > >for a new partition. However, According to @Shammon's suggestion, > different > >startup OffsetsInitializers correspond to different post-startup > >OffsetsInitializers for Flink's built-in offset strategies. > >2. "Future-time" TIMESTAMP OffsetsInitializer. I looked at the code > >again, and it seems that neither @Shammon nor I have figured out . > >TimestampOffsetsInitializer#getPartitionOffsets has a comment: "First > >get the current end offsets of the partitions. This is going to be > used in > >case we cannot find a suitable offset based on the timestamp, i.e., > the > >message meeting the requirement of the timestamp has not been > produced to > >Kafka yet. *In this case, we just use the latest offset*." Therefore, > >using the TimestampOffsetsInitializer will always have an offset at > >startup. > >3. Clarification on coupling SPECIFIC-OFFSET startup with > >SPECIFIC-OFFSET post-startup. SPECIFIC-OFFSET strategy already uses > >"auto.offset.reset" position for partitions that are not hit. > > > > @Gordon, @Shammon, @Leonard, the core issue we are concerned about is > > whether the offset specified at the beginning includes non-exist > > partitions. The previous design may have SPECIFIC-OFFSET startup with > > future partition. However, I think since different strategies have been > > used for the first discovered partition and the later discovered > partition, > > the specified offset at startup should be the partitions that have been > > confirmed to exist, if not an error will be thrown. If partitions still > not > > exist, it should be specified in the post-startup OffsetsInitializers > > (default EARLIEST). > > > > Best > > Hongshun > > > > > > On Thu, Mar 30, 2023 at 1:43 PM Shammon FY wrote: > > > >> Thanks Gordon and Leonard > >> > >> I'm sorry that there is no specific case from my side, but I consider > the > >> issue as follows > >> > >> 1. Users may set an offset later than the current time because Flink > does > >> not limit it > >> 2. If we use EARLIEST for a newly discovered partition with different > >> OFFSETs, which may be different from the previous strategy. I think it's > >> best to keep the same strategy as before if it does not cause data > losing > >> 3. I think support different OFFSETs in the FLIP will not make the > >> implementation more complexity > >> > >> Of course, if it is confirmed that this is an illegal Timestamp OFFSET > and > >> Flink validate it. Then we can use the same strategy to apply to the > newly > >> discovered partition, I think this will be nice too > >> > >> Best, > >> Shammon FY > >> > >> > >> On Thu, Mar 30, 2023 at 12:29 PM Leonard Xu wrote: > >> > >> > Thanks Hongshun and Shammon for driving the FLIP! > >> > > >> > > >> > > *2. Clarification on "future-time" TIMESTAMP OffsetsInitializer* > >> > > *3. Clarification on coupling SPECIFIC-OFFSET startup with > >> > SPECIFIC-OFFSET > >> &
Re: [DISCUSS] FLIP 295: Support persistence of Catalog configuration and asynchronous registration
Hi Feng
Thanks for your update.
I found there are two options in `Proposed Changes`, can you put the
unselected option in `Rejected Alternatives`? I think this may help us
better understand your proposal
Best,
Shammon FY
On Thu, Apr 13, 2023 at 4:49 AM Jing Ge wrote:
> Hi Feng,
>
> Thanks for raising this FLIP. I am still confused after completely reading
> the thread with following questions:
>
> 1. Naming confusion - registerCatalog() and addCatalog() have no big
> difference based on their names. One of them is responsible for data
> persistence. How about persistCatalog()?
> 2. As you mentioned that Map catalogs is used as a cache
> and catalogStore is used for data persistence. I would suggest describing
> their purpose conceptually and clearly in the FLIP. Some common cache
> features should be implemented, i.e. data in the cache and the store should
> be consistent. Same Catalog instance should be found in the store and in
> the cache(either it has been initialized or it will be lazy initialized)
> for the same catalog name. The consistency will be taken care of while
> updating the catalog.
> 3. As the above discussion moves forward, the option 2 solution looks more
> like a replacement of option 1, because, afaiu, issues mentioned
> previously with option 1 are not solved yet. Do you still want to propose
> both options and ask for suggestions for both of them?
> 4. After you updated the FLIP, there are some inconsistent descriptions in
> the content. Would you like to clean them up? Thanks!
>
> Best regards,
> Jing
>
>
> On Fri, Apr 7, 2023 at 9:24 AM Feng Jin wrote:
>
> > hi Shammon
> >
> > Thank you for your response, and I completely agree with your point of
> > view.
> > Initially, I may have over complicated the whole issue. First and
> foremost,
> > we need to consider the persistence of the Catalog's Configuration.
> > If we only need to provide persistence for Catalog Configuration, we can
> > add a toConfiguration method to the Catalog interface.
> > This method can convert a Catalog instance to a Map
> > properties, and the default implementation will throw an exception.
> >
> > public interface Catalog {
> >/**
> >* Returns a map containing the properties of the catalog object.
> >*
> >* @return a map containing the properties of the catalog object
> >* @throws UnsupportedOperationException if the implementing class does
> > not override
> >* the default implementation of this method
> >*/
> > default Map toProperties() {
> > throw new UnsupportedOperationException("Please implement
> toProperties
> > for this catalog");
> > }
> > }
> >
> > The specific process is as follows:
> >
> > 1. If the user has configured a CatalogStore, the toConfiguration()
> method
> > will be called when registering a Catalog instance with
> > registerCatalog(String catalogName, Catalog catalog). The Catalog
> instance
> > will be converted to a Map properties and saved to the
> > CatalogStore. If some Catalog instances do not implement this method, an
> > exception will be thrown.
> >
> > 2. If the user does not use a CatalogStore, the toConfiguration() method
> > will not be called, ensuring consistency with the original process.
> >
> > 3. Saving both the Map catalogs and the CatalogStore at
> > the same time can also avoid conflicts
> >
> >
> > For lazy initialization:
> > we can start from the Catalog itself and provide a dedicated Catalog
> > implementation for lazy loading, such as ConfigurationCatalog
> >
> > public class ConfigurationCatalog implements Catalog {
> > ConfigurationCatalog(Map properties) {
> > }
> > }
> >
> > I have added this design to the FLIP.
> >
> >
> > Best,
> > Feng
> >
> > On Thu, Apr 6, 2023 at 10:03 AM Shammon FY wrote:
> >
> > > Hi Feng
> > >
> > > Thanks for your answer.
> > >
> > > I have no questions about the functionality of `CatalogStore`, but the
> > > different operations of multiple `registerCatalog` or `storeCatalog` in
> > > `CatalogManager`. The implementation in Trino is also the same, the
> > > `CatalogManager` in trino only has one `createCatalog`, which will save
> > the
> > > catalog to memory and `CatalogStore`.
> > >
> > > I think we don't need to add another `registerCatalog` or `addCatalog`
> to
> > > save a catalog in `Map catalogs` or `CatalogStore
> > > catalogStore`. As you mentioned before, the `Map
> > >
Re: [DISCUSS] FLIP 295: Support persistence of Catalog configuration and asynchronous registration
Hi Feng Thanks for your answer. I have no questions about the functionality of `CatalogStore`, but the different operations of multiple `registerCatalog` or `storeCatalog` in `CatalogManager`. The implementation in Trino is also the same, the `CatalogManager` in trino only has one `createCatalog`, which will save the catalog to memory and `CatalogStore`. I think we don't need to add another `registerCatalog` or `addCatalog` to save a catalog in `Map catalogs` or `CatalogStore catalogStore`. As you mentioned before, the `Map catalogs` is a cache in `CatalogManager`. How about saving the catalog in `Map catalogs` and `CatalogStore catalogStore` together when it is registered or added in `CatalogManager`? Best, Shammon FY On Tue, Apr 4, 2023 at 5:22 PM Feng Jin wrote: > Thank you for your reply. I am very sorry for the misunderstanding caused > by my deviation from the original discussion. > > @Shammon > > I found there is a pre-discussion [1] for this FLIP > Yes, there was indeed such a discussion before. However, in designing the > whole solution, I found that the logic of CatalogManager itself doesn't > need to change much. *We cannot only persist Catalog instances themselves*, > so exposing only registerCatalog(String catalogName, Catalog catalog) might > not be enough to save Catalogs, because in the end we still need to save > the configurations corresponding to the Catalog instances. Therefore, I > decided to introduce the CatalogStore interface for configuration > persistence. Regarding this part, I also took inspiration from Trino's > implementation[1]. > > > @yuxia > > > 1: The mechanism of handling catalog with the same name looks a little of > complex to me. > Thank you for the suggestion. I will provide a detailed description and > code examples for this part, and add it to the FLIP. > > > 2: TBH, the method name `addCatalog` still looks confused to me. IIUC, > this method is for storing catalog to CatalogStore, how about renaming it > to `storeCatalog`? It's very personal opinion, you can decide to take it or > not by your self. > StoreCatalog looks more intuitive to me. I don't see any problem with it. > > > 3.For CREATE CATALOG statement, which method will be called? > `registerCatalog` or `addCatalog`? I'm wondering whether user can add a > catalog to store with SQL stement. > For CREATE CATALOG, my original design was to add it directly to the > CatalogStore, but this would disrupt the existing logic. Therefore, I think > we can do both: save the configuration to the CatalogStore and initialize > the Catalog instance at the same time > > > 3. Is it really neccessary to provide a default implmentation for > interface `CatalogStoreFactory`? > I think it is necessary, otherwise we would need to introduce an additional > Map to store the configuration for lazy loading. > > > 4: About asynchronous registration for catalog. > I don't think registerCatalog(String catalogName, Catalog catalog) can be > made into an asynchronous interface because Catalog is already an instance. > > > It looks more like lazy initialization for catalog than asynchronous > registration, right? > Indeed, my description was inaccurate. It should be lazy registration > instead of asynchronous registration. I have already updated the title of > the FLIP. > > > > > [1]. > > https://github.com/trinodb/trino/blob/master/core/trino-main/src/main/java/io/trino/connector/CatalogStore.java > > > On Tue, Apr 4, 2023 at 4:27 PM Shammon FY wrote: > > > Hi Feng > > > > I think if there is a `registerCatalog` method in `CatalogManager`, it > will > > confuse users whether a method named `addCatalog` or `storeCatalog` is > > added. > > > > And as you mentioned, the memory catalog is a `cache`, I think the > concept > > of `cache` should not be exposed to users. > > > > I found there is a pre-discussion [1] for this FLIP. Please correct me if > > I'm wrong, IIUC, the conclusion of that discussion is to use > > `CatalogManager` as an interface and implement it for different stores > such > > as memory, file and external system. > > > > I think there is a gap between the current FLIP design and that > conclusion. > > What about the proposal of the discussion in thread [1] ? > > > > > > [1] https://lists.apache.org/thread/9bnjblgd9wvrl75lkm84oo654c4lqv70 > > > > > > Best, > > Shammon FY > > > > > > On Tue, Apr 4, 2023 at 3:41 PM yuxia > wrote: > > > > > Thanks Feng for driving this FLIP. > > > I have few comments: > > > 1: The mechanism of handling catalog with the same name looks a little > of > > > complex to
Re: [DISCUSS] FLIP 295: Support persistence of Catalog configuration and asynchronous registration
Hi Feng I think if there is a `registerCatalog` method in `CatalogManager`, it will confuse users whether a method named `addCatalog` or `storeCatalog` is added. And as you mentioned, the memory catalog is a `cache`, I think the concept of `cache` should not be exposed to users. I found there is a pre-discussion [1] for this FLIP. Please correct me if I'm wrong, IIUC, the conclusion of that discussion is to use `CatalogManager` as an interface and implement it for different stores such as memory, file and external system. I think there is a gap between the current FLIP design and that conclusion. What about the proposal of the discussion in thread [1] ? [1] https://lists.apache.org/thread/9bnjblgd9wvrl75lkm84oo654c4lqv70 Best, Shammon FY On Tue, Apr 4, 2023 at 3:41 PM yuxia wrote: > Thanks Feng for driving this FLIP. > I have few comments: > 1: The mechanism of handling catalog with the same name looks a little of > complex to me. I think it'll be better to explain it in the java doc of > these methods and give a brief example in this FLIP. > > 2: TBH, the method name `addCatalog` still looks confused to me. IIUC, > this method is for storing catalog to CatalogStore, how about renaming it > to `storeCatalog`? It's very personal opinion, you can decide to take it or > not by your self. > > 3: For CREATE CATALOG statement, which method will be called? > `registerCatalog` or `addCatalog`? I'm wondering whether user can add a > catalog to store with SQL stement. > > > 3: Is it really neccessary to provide a default implmentation for > interface `CatalogStoreFactory`? > > > 4: About asynchronous registration for catalog. > > > When creating a catalog with CREATE CATALOG, the asynchronous > registration method is used by default. > If asynchronous registration is the default behavior, it there any way > that user can switch to synchronous registration just like before? > Will both method `addCatalog` and `registerCatalog` be asynchronous > registration? > > IIUC, in asynchronous registration, it may well that CREATE CATALOG > executes successfully, but then the following CREATE TABLE statement will > fail for the catalog fail to open. > I think it's a break change which should be highlighted in this FLIP, may > be in compatibility part. > > > BTW, by saying asynchronous registration, I would like to expect there > will be an executor to open or register catalog in the background, but from > your previous comments, > "the Catalog instance will be initialized if it has not been initialized > yet. If the initialization process fails, these statements will not be > executed successfully." > It looks more like lazy initialization for catalog than asynchronous > registration, right? > > > Best regards, > Yuxia > > - 原始邮件 - > 发件人: "Feng Jin" > 收件人: "dev" > 发送时间: 星期一, 2023年 4 月 03日 下午 3:27:45 > 主题: Re: [DISCUSS] FLIP 295: Support persistence of Catalog configuration > and asynchronous registration > > Hi everyone, Thank you all for your interest in this DISCUSS. > > @Shammon > > How to handle a catalog with the same name that exists for both of them? > > I believe this is a crucial point. Based on my current personal > understanding, the Map catalogs will serve as a cache > for instantiated catalogs and have the highest priority. > > There are three methods that can affect the Map catalogs: > > 1. registerCatalog(String catalogName, Catalog catalog) > > This method puts the catalog instance into the Map > catalogs. > > 2. unregisterCatalog(String catalogName) > > This method removes the catalog instance corresponding to catalogName > from the Map catalogs. > > 3. getCatalog(String catalogName) > > This method first retrieves the corresponding catalog instance from > the Map catalogs. If the catalog does not exist, it > retrieves the corresponding configuration from the CatalogStore, > initializes it, and puts the initialized Catalog instance into the > Map catalogs. > > The following two methods only modify the configuration in the > CatalogStore: > > 1. addCatalog(String catalogName, Map properties) > > This method saves the properties to the catalogStore and checks > whether there is a catalogName with the same name. > > 2. removeCatalog(String catalogName) > This method removes the specified configuration of the specified > catalogName in the catalogStore. > > The following are possible conflict scenarios: > > 1. When the corresponding catalogName already exists in the > CatalogStore but not in the Map, the > registerCatalog(String catalogName, Catalog catalog) method can > succeed and be directly saved to the Map catalogs. > > 2. When th
Re: [DISCUSS] FLIP 295: Support persistence of Catalog configuration and asynchronous registration
Hi Feng Thanks for driving this FLIP. The idea of catalog configuration is cool and I think it will be very useful. I have some comments about the FLIP 1. `CatalogStore` in the FLIP stores information of catalogs, what's the relationship between `Map catalogs` and `CatalogStore catalogStore`? How to handle a catalog with the same name that exists for both of them? 2. I think it will confuse users that `registerCatalog(String catalogName, Catalog catalog)` in the `Map catalogs` and `registerCatalog(String catalogName, Map properties)` in the `CatalogStore catalogStore`. Do users need to understand the differences between the two `registerCatalog`? 3. How about supporting multiple catalogs in one `xml` or `json` file? Best, Shammon FY On Thu, Mar 30, 2023 at 2:21 PM Feng Jin wrote: > Hi everyone, > > I would like to start a discuss on FLIP-295: Support persistence of > Catalog configuration and asynchronous registration > > Currently, when using Catalog in Flink, it is necessary to re-register > each time. Therefore, providing persistent configuration for Catalog > will make its management more convenient. > Especially for SQL management modules such as Flink SQL gateway and > SQL client, it eliminates the hassle of registering every time before > starting. > > Look forward to your opinions! > > [1] > https://cwiki.apache.org/confluence/display/FLINK/FLIP-295%3A+Support+lazy+initialization+of+catalogs+and+persistence+of+catalog+configurations > [2] https://issues.apache.org/jira/browse/FLINK-31259 > > Best, > Feng. >
Re: [DISCUSS] FLIP-288:Enable Dynamic Partition Discovery by Default in Kafka Source
Thanks Gordon and Leonard I'm sorry that there is no specific case from my side, but I consider the issue as follows 1. Users may set an offset later than the current time because Flink does not limit it 2. If we use EARLIEST for a newly discovered partition with different OFFSETs, which may be different from the previous strategy. I think it's best to keep the same strategy as before if it does not cause data losing 3. I think support different OFFSETs in the FLIP will not make the implementation more complexity Of course, if it is confirmed that this is an illegal Timestamp OFFSET and Flink validate it. Then we can use the same strategy to apply to the newly discovered partition, I think this will be nice too Best, Shammon FY On Thu, Mar 30, 2023 at 12:29 PM Leonard Xu wrote: > Thanks Hongshun and Shammon for driving the FLIP! > > > > *2. Clarification on "future-time" TIMESTAMP OffsetsInitializer* > > *3. Clarification on coupling SPECIFIC-OFFSET startup with > SPECIFIC-OFFSET > > post-startup* > > Grodan raised a good point about the future TIMESTAMP and SPECIFIC-OFFSET, > the timestamps/offset of the newly added partition is undetermined when the > job starts (the partition has not been created yet), and it is the > timestamps/offset in the future. > > I used many message queue systems like Kafka, Pulsar, xxMQ. In my past > experience, TIMESTAMP and SPECIFIC-OFFSET startup modes are usually used > to specify existing timestamps/offset, which are used for business > scenarios such as backfilling data and re-refreshing data. At present, It's > hard to imagine a user scenario specifying a future timestamp to filter > data in the current topic of message queue system. Is it overthinking to > consider future future TIMESTAMP and SPECIFIC-OFFSET? > > > Best, > Leonard
Re: [DISCUSS] FLIP-288:Enable Dynamic Partition Discovery by Default in Kafka Source
Hi hongshun Thanks for updating, the FLIP's pretty good and looks good to me! Best, Shammon FY On Tue, Mar 28, 2023 at 11:16 AM Hongshun Wang wrote: > Hi Shammon, > > > Thanks a lot for your advise. I agree with your opinion now. It seems that > I forgot to consider that it may be at a certain point in the future. > > > I will modify OffsetsInitializer to provide a different strategy for later > discovered partitions, by which users can also customize strategies for new > and old partitions. > > WDYT? > > > Yours > > Hongshun > > On Tue, Mar 28, 2023 at 9:00 AM Shammon FY wrote: > > > Hi Hongshun > > > > Thanks for your answer. > > > > I think the startup offset of Kafka such as timestamp or > > specific_offset has no relationship with `Window Operator`. Users can > > freely set the starting position according to their needs, it may be > before > > the latest Kafka data, or it may be at a certain point in the future. > > > > The offsets set by users in Kafka can be divided into four types at the > > moment: EARLIEST, LATEST, TIMESTAMP, SPECIFIC_OFFSET. The new discovered > > partitions may need to be handled with different strategies for these > four > > types: > > > > 1. EARLIEST, use EARLIEST for the new discovered partitions > > 2. LATEST, use EARLIEST for the new discovered partitions > > 3. TIMESTAMP, use TIMESTAMP for the new discovered partitions > > 4. SPECIFIC_OFFSET, use SPECIFIC_OFFSET for the new discovered partitions > > > > From above, it seems that we only need to do special processing for > > EARLIEST. What do you think of it? > > > > Best, > > Shammon FY > > > > > > On Fri, Mar 24, 2023 at 11:23 AM Hongshun Wang > > wrote: > > > > > "If all new messages in old partitions should be consumed, all new > > messages > > > in new partitions should also be consumed." > > > > > > Sorry, I wrote the last sentence incorrectly. > > > > > > On Fri, Mar 24, 2023 at 11:15 AM Hongshun Wang < > loserwang1...@gmail.com> > > > wrote: > > > > > > > Hi Shammon, > > > > > > > > Thanks for your advise! I learn a lot about > TIMESTAMP/SPECIFIC_OFFSET. > > > > That's interesting. > > > > > > > > However, I have a different opinion. > > > > > > > > If a user employs the SPECIFIC_OFFSET strategy and enables > > > auto-discovery, > > > > they will be able to find new partitions beyond the specified offset. > > > > Otherwise, enabling auto-discovery is no sense. > > > > > > > > When it comes to the TIMESTAMP strategy, it seems to be trivial. I > > > > understand your concern, however, it’s the role of time window rather > > > than > > > > partition discovery. The TIMESTAMP strategy means that the consumer > > > starts > > > > from the first record whose timestamp is greater than or equal to a > > given > > > > timestamp, rather than only consuming all records whose timestamp is > > > > greater than or equal to the given timestamp. *Thus, even disable > auto > > > > discovery or discover new partitions with TIMESTAMP strategy, same > > > problems > > > > still occur.* > > > > > > > > Above all , why use EARLIEST strategy? I believe that the strategy > > > > specified by the startup should be the strategy at the moment of > > > startup. *So > > > > there is no difference between new partitions and new messages in old > > > > partitions.* Therefore, all the new partition issues that you care > > about > > > > will still appear even if you disable the partition, as new messages > in > > > old > > > > partitions. If all new messages in old partitions should be consume, > > all > > > > new messages in old partitions should also be consume. > > > > > > > > > > > > Best, > > > > Hongshun > > > > > > > > On Thu, Mar 23, 2023 at 8:34 PM Shammon FY > wrote: > > > > > > > >> Hi Hongshun > > > >> > > > >> Thanks for driving this discussion. Automatically discovering > > partitions > > > >> without losing data sounds great! > > > >> > > > >> Currently flink supports kafka source with different startup modes, > > such > > > >> as > > > >> EARLIEST, LATEST, TIMESTAMP,
Re: [DISCUSS] FLIP-288:Enable Dynamic Partition Discovery by Default in Kafka Source
Hi Hongshun Thanks for your answer. I think the startup offset of Kafka such as timestamp or specific_offset has no relationship with `Window Operator`. Users can freely set the starting position according to their needs, it may be before the latest Kafka data, or it may be at a certain point in the future. The offsets set by users in Kafka can be divided into four types at the moment: EARLIEST, LATEST, TIMESTAMP, SPECIFIC_OFFSET. The new discovered partitions may need to be handled with different strategies for these four types: 1. EARLIEST, use EARLIEST for the new discovered partitions 2. LATEST, use EARLIEST for the new discovered partitions 3. TIMESTAMP, use TIMESTAMP for the new discovered partitions 4. SPECIFIC_OFFSET, use SPECIFIC_OFFSET for the new discovered partitions >From above, it seems that we only need to do special processing for EARLIEST. What do you think of it? Best, Shammon FY On Fri, Mar 24, 2023 at 11:23 AM Hongshun Wang wrote: > "If all new messages in old partitions should be consumed, all new messages > in new partitions should also be consumed." > > Sorry, I wrote the last sentence incorrectly. > > On Fri, Mar 24, 2023 at 11:15 AM Hongshun Wang > wrote: > > > Hi Shammon, > > > > Thanks for your advise! I learn a lot about TIMESTAMP/SPECIFIC_OFFSET. > > That's interesting. > > > > However, I have a different opinion. > > > > If a user employs the SPECIFIC_OFFSET strategy and enables > auto-discovery, > > they will be able to find new partitions beyond the specified offset. > > Otherwise, enabling auto-discovery is no sense. > > > > When it comes to the TIMESTAMP strategy, it seems to be trivial. I > > understand your concern, however, it’s the role of time window rather > than > > partition discovery. The TIMESTAMP strategy means that the consumer > starts > > from the first record whose timestamp is greater than or equal to a given > > timestamp, rather than only consuming all records whose timestamp is > > greater than or equal to the given timestamp. *Thus, even disable auto > > discovery or discover new partitions with TIMESTAMP strategy, same > problems > > still occur.* > > > > Above all , why use EARLIEST strategy? I believe that the strategy > > specified by the startup should be the strategy at the moment of > startup. *So > > there is no difference between new partitions and new messages in old > > partitions.* Therefore, all the new partition issues that you care about > > will still appear even if you disable the partition, as new messages in > old > > partitions. If all new messages in old partitions should be consume, all > > new messages in old partitions should also be consume. > > > > > > Best, > > Hongshun > > > > On Thu, Mar 23, 2023 at 8:34 PM Shammon FY wrote: > > > >> Hi Hongshun > >> > >> Thanks for driving this discussion. Automatically discovering partitions > >> without losing data sounds great! > >> > >> Currently flink supports kafka source with different startup modes, such > >> as > >> EARLIEST, LATEST, TIMESTAMP, SPECIFIC_OFFSETS and GROUP_OFFSET. > >> > >> If I understand correctly, you will set the offset of new partitions > with > >> EARLIEST? Please correct me if I'm wrong, I think the EARLIEST startup > >> mode > >> for new partitions is not suitable if users set > TIMESTAMP/SPECIFIC_OFFSET > >> for kafka in their jobs. > >> > >> For an extreme example, the current time is 2023-03-23 15:00:00 and > users > >> set the TIMESTAMP with 2023-03-23 16:00:00 for their jobs. If a > partition > >> is added during this period, jobs will generate “surprising” data. What > do > >> you think of it? > >> > >> > >> Best, > >> Shammon FY > >> > >> > >> On Tue, Mar 21, 2023 at 6:58 PM Hongshun Wang > >> wrote: > >> > >> > Hi, Hang, > >> > > >> > Thanks for your advice. > >> > > >> > When the second case will occur? Currently, there are three ways to > >> specify > >> > partitions in Kafka: by topic, by partition, and by matching the topic > >> with > >> > a regular expression. Currently, if the initial partition number is 0, > >> an > >> > error will occur for the first two methods. However, when using a > >> regular > >> > expression to match topics, it is allowed to have 0 matched topics. > >> > > >> > > I don't know when the second case will occur >
Re: [ANNOUNCE] Flink Table Store Joins Apache Incubator as Apache Paimon(incubating)
Congratulations! Best, Shammon FY On Mon, Mar 27, 2023 at 11:37 PM Samrat Deb wrote: > congratulations > > Bests, > Samrat > On Mon, 27 Mar 2023 at 8:24 PM, Alexander Fedulov < > alexander.fedu...@gmail.com> wrote: > > > Great to see this, congratulations! > > > > Best, > > Alex > > > > On Mon, 27 Mar 2023 at 11:24, Yu Li wrote: > > > > > Dear Flinkers, > > > > > > > > > > > > As you may have noticed, we are pleased to announce that Flink Table > > Store has joined the Apache Incubator as a separate project called Apache > > Paimon(incubating) [1] [2] [3]. The new project still aims at building a > > streaming data lake platform for high-speed data ingestion, change data > > tracking and efficient real-time analytics, with the vision of > supporting a > > larger ecosystem and establishing a vibrant and neutral open source > > community. > > > > > > > > > > > > We would like to thank everyone for their great support and efforts for > > the Flink Table Store project, and warmly welcome everyone to join the > > development and activities of the new project. Apache Flink will continue > > to be one of the first-class citizens supported by Paimon, and we believe > > that the Flink and Paimon communities will maintain close cooperation. > > > > > > > > > 亲爱的Flinkers， > > > > > > > > > 正如您可能已经注意到的，我们很高兴地宣布，Flink Table Store 已经正式加入 Apache > > > 孵化器独立孵化 [1] [2] [3]。新项目的名字是 > > > Apache > > > Paimon(incubating)，仍致力于打造一个支持高速数据摄入、流式数据订阅和高效实时分析的新一代流式湖仓平台。此外，新项目将支持更加丰富的生态，并建立一个充满活力和中立的开源社区。 > > > > > > > > > 在这里我们要感谢大家对 Flink Table Store 项目的大力支持和投入，并热烈欢迎大家加入新项目的开发和社区活动。Apache > > Flink > > > 将继续作为 Paimon 支持的主力计算引擎之一，我们也相信 Flink 和 Paimon 社区将继续保持密切合作。 > > > > > > > > > Best Regards, > > > > > > Yu (on behalf of the Apache Flink PMC and Apache Paimon PPMC) > > > > > > > > > 致礼， > > > > > > 李钰（谨代表 Apache Flink PMC 和 Apache Paimon PPMC） > > > > > > > > > [1] https://paimon.apache.org/ > > > > > > [2] https://github.com/apache/incubator-paimon > > > > > > [3] > https://cwiki.apache.org/confluence/display/INCUBATOR/PaimonProposal > > > > > >
Re: [ANNOUNCE] Kafka Connector Code Removal from apache/flink:main branch and code freezing
Thanks Jing and Gordon, I have closed the pr https://github.com/apache/flink/pull/21965 and will open a new one for kafka connector Best, shammon FY On Saturday, March 25, 2023, Ran Tao wrote: > Thank you Gordon and all the people who have worked on the externalized > kafka implementation. > I have another pr related to Kafka[1]. I will be very appreciative if you > can help me review it in your free time. > > [1] https://github.com/apache/flink-connector-kafka/pull/10 > > Best Regards, > Ran Tao > > > Tzu-Li (Gordon) Tai 于2023年3月24日周五 23:21写道： > > > Thanks Jing! I missed https://github.com/apache/flink/pull/21965 indeed. > > > > Please let us know if anything else was overlooked. > > > > On Fri, Mar 24, 2023 at 8:13 AM Jing Ge > > wrote: > > > > > Thanks Gordon for driving this! There is another PR related to Kafka > > > connector: https://github.com/apache/flink/pull/21965 > > > > > > Best regards, > > > Jing > > > > > > On Fri, Mar 24, 2023 at 4:06 PM Tzu-Li (Gordon) Tai < > tzuli...@apache.org > > > > > > wrote: > > > > > > > Hi all, > > > > > > > > Now that Flink 1.17 has been released, and given that we've already > > > synced > > > > the latest Kafka connector code up to Flink 1.17 to the > > > > apache/flink-connector-kafka repo (thanks to Mason and Martijn for > most > > > of > > > > the effort!), we're now in the final step of completely removing the > > > Kafka > > > > connector code from apache/flink:main branch, tracked by FLINK-30859 > > [1]. > > > > > > > > As such, we'd like to ask that no more Kafka connector changes gets > > > merged > > > > to apache/flink:main, effective now. Going forward, all Kafka > connector > > > PRs > > > > should be opened directly against the apache/flink-connector-kafka: > main > > > > branch. > > > > > > > > Meanwhile, there's a couple of "dangling" Kafka connector PRs over > the > > > last > > > > 2 months that is opened against apache/flink:main: > > > > > > > >1. [FLINK-31305] Propagate producer exceptions outside of mailbox > > > >executor [2] > > > >2. [FLINK-31049] Add support for Kafka record headers to KafkaSink > > [3] > > > >3. [FLINK-31262] Move kafka sql connector fat jar test to > > > >SmokeKafkaITCase [4 ] > > > >4. [hotfix] Add writeTimestamp option to > > > >KafkaRecordSerializationSchemaBuilder [5] > > > > > > > > Apart from 1. [FLINK-31305] which is a critical bug and is already in > > > > review closed to being merged, for the rest we will be reaching out > on > > > the > > > > PRs to ask the authors to close the PR and reopen them against > > > > apache/flink-connector-kafka:main. > > > > > > > > Thanks, > > > > Gordon > > > > > > > > [1] https://issues.apache.org/jira/browse/FLINK-30859 > > > > [2] https://github.com/apache/flink/pull/22150 > > > > [3] https://github.com/apache/flink/pull/8 > > > > [4] https://github.com/apache/flink/pull/22060 > > > > [5] https://github.com/apache/flink/pull/22037 > > > > > > > > > >
Re: [DISCUSS] FLIP-292: Support configuring state TTL at operator level for Table API & SQL programs
Hi jane Thanks for initializing this discussion. Configure TTL per operator can help users manage state more effectively. I think the `compiled json plan` proposal may need to consider the impact on the user's submission workflow. Generally, Flink jobs support two types of submission: SQL and jar. If users want to use `TTL on Operator` for SQL jobs, they need to edit the json file which is not supported by general job submission systems such as flink sql-client, apache kyuubi, apache streampark and .etc. Users need to download the file and edit it manually, but they may not have the permissions to the storage system such as HDFS in a real production environment. >From this perspective, I think it is necessary to provide a way similar to hits that users can configure the `TTL on Operator` in their sqls which help users to use it conveniently. At the same time, I agree with @Shuo's idea that for complex cases, users can combine hits and `json plan` to configure `TTL on Operator` better. What do you think? Thanks Best, Shammon FY On Thu, Mar 23, 2023 at 9:58 PM Shuo Cheng wrote: > Correction: “users can set 'scan.startup.mode' for kafka connector” -> > “users > can set 'scan.startup.mode' for kafka connector by dynamic table option” > > Shuo Cheng 于2023年3月23日 周四21:50写道： > > > Hi Jane, > > Thanks for driving this, operator level state ttl is absolutely a desired > > feature. I would share my opinion as following: > > > > If the scope of this proposal is limited as an enhancement for compiled > > json plan, it makes sense. I think it does not conflict with configuring > > state ttl > > in other ways, e.g., SQL HINT or something else, because they just work > in > > different level, SQL Hint works in the exact entrance of SQL API, while > > compiled json plan is the intermediate results for SQL. > > I think the final shape of state ttl configuring may like the that, users > > can define operator state ttl using SQL HINT (assumption...), but it may > > affects more than one stateful operators inside the same query block, > then > > users can further configure a specific one by modifying the compiled json > > plan... > > > > In a word, this proposal is in good shape as an enhancement for compiled > > json plan, and it's orthogonal with other ways like SQL Hint which works > in > > a higher level. > > > > > > Nips: > > > > > "From the SQL semantic perspective, hints cannot intervene in the > > calculation of data results." > > I think it's more properly to say that hint does not affect the > > equivalence of execution plans (hash agg vs sort agg), not the > equivalence > > of execution results, e.g., users can set 'scan.startup.mode' for kafka > > connector, which also "intervene in the calculation of data results". > > > > Sincerely, > > Shuo > > > > On Tue, Mar 21, 2023 at 7:52 PM Jane Chan wrote: > > > >> Hi devs, > >> > >> I'd like to start a discussion on FLIP-292: Support configuring state > TTL > >> at operator level for Table API & SQL programs [1]. > >> > >> Currently, we only support job-level state TTL configuration via > >> 'table.exec.state.ttl'. However, users may expect a fine-grained state > TTL > >> control to optimize state usage. Hence we propose to > serialize/deserialize > >> the state TTL as metadata of the operator's state to/from the compiled > >> JSON > >> plan, to achieve the goal that specifying different state TTL when > >> transforming the exec node to stateful operators. > >> > >> Look forward to your opinions! > >> > >> [1] > >> > https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=240883951 > >> > >> Best Regards, > >> Jane Chan > >> > > >
Re: [DISCUSS] FLIP-288:Enable Dynamic Partition Discovery by Default in Kafka Source
Hi Hongshun Thanks for driving this discussion. Automatically discovering partitions without losing data sounds great! Currently flink supports kafka source with different startup modes, such as EARLIEST, LATEST, TIMESTAMP, SPECIFIC_OFFSETS and GROUP_OFFSET. If I understand correctly, you will set the offset of new partitions with EARLIEST? Please correct me if I'm wrong, I think the EARLIEST startup mode for new partitions is not suitable if users set TIMESTAMP/SPECIFIC_OFFSET for kafka in their jobs. For an extreme example, the current time is 2023-03-23 15:00:00 and users set the TIMESTAMP with 2023-03-23 16:00:00 for their jobs. If a partition is added during this period, jobs will generate “surprising” data. What do you think of it? Best, Shammon FY On Tue, Mar 21, 2023 at 6:58 PM Hongshun Wang wrote: > Hi, Hang, > > Thanks for your advice. > > When the second case will occur? Currently, there are three ways to specify > partitions in Kafka: by topic, by partition, and by matching the topic with > a regular expression. Currently, if the initial partition number is 0, an > error will occur for the first two methods. However, when using a regular > expression to match topics, it is allowed to have 0 matched topics. > > > I don't know when the second case will occur > > > Why prefer the field `firstDiscoveryDone`? When a regular expression > initially matches 0 topics, it should consume all messages of the new > topic. If unassignedInitialPartitons and unassignedTopLevelPartitions are > used instead of firstDiscoveryDone, any new topics created during (5 > minutes discovery + job restart time) will be treated as the first > discovery, causing data loss. > > > Then when will we get the empty partition list? I think it should be > treated as the first initial discovery if both `unassignedInitialPartitons` > and `assignedPartitons` are empty without `firstDiscoveryDone`. > > Best > > Hongshun > > On Tue, Mar 21, 2023 at 5:56 PM Hang Ruan wrote: > > > Hi, Hongshun, > > > > Thank you for starting this discussion. I have some problems about the > > field `firstDiscoveryDone`. > > > > In the FLIP, why we need firstDiscoveryDone is as follows. > > > Why do we need firstDiscoveryDone? Only relying on the > > unAssignedInitialPartitons attribute cannot distinguish between the > > following two cases (which often occur in pattern mode): > > > The first partition discovery is so slow, before which the checkpoint > is > > executed and then job is restarted . At this time, the restored > > unAssignedInitialPartitons is an empty set, which means non-discovery. > The > > next discovery will be treated as first discovery. > > > The first time the partition is discovered is empty, and new partitions > > can only be found after multiple partition discoveries. If a restart > occurs > > between this period, the restored unAssignedInitialPartitons is also an > > empty set, which means empty-discovery.The next discovery will be treated > > as new discovery. > > > > I don't know when the second case will occur. The partitions must be > > greater than 0 when creating topics. And I have read this note in the > FLIP. > > > Note: The current design only applies to cases where all existing > > partitions can be discovered at once. If all old partitions cannot be > > discovered at once, the subsequent old partitions discovered will be > > treated as new partitions, leading to message duplication. Therefore, > this > > point needs to be particularly noted. > > > > Then when will we get the empty partition list? I think it should be > > treated as the first initial discovery if both > `unassignedInitialPartitons` > > and `assignedPartitons` are empty without `firstDiscoveryDone`. > > > > Besides that, I think the `unAssignedInitialPartitons` is better to be > > named `unassignedInitialPartitons`. > > > > Best, > > Hang > > > > Hongshun Wang 于2023年3月17日周五 18:42写道： > > > > > Hi everyone, > > > > > > I would like to start a discussion on FLIP-288:Enable Dynamic Partition > > > Discovery by Default in Kafka Source[1]. > > > > > > As described in mail thread[2], dynamic partition discovery is disabled > > by > > > default and users have to explicitly specify the interval of discovery > in > > > order to turn it on. Besides, if the initial offset strategy is LATEST, > > > same strategy is used for new partitions, leading to the loss of some > > data > > > (thinking a new partition is created and might be discovered by Kafka > > > source several
Re: [DISCUSS] FLIP-302: Support TRUNCATE TABLE statement
Hi yuxia Thanks for initiating this discussion. There are usually two types of data deletion in a production environment: one is deleting data directly and the other is moving the data to the trash directory which will be deleted periodically by the underlying system. Can we distinguish between these two operations in the truncate syntax? Or support adding options in `with`? Best, Shammon FY On Wed, Mar 22, 2023 at 9:13 PM yuxia wrote: > Hi, devs. > > I'd like to start a discussion about FLIP-302: Support TRUNCATE TABLE > statement [1]. > > The TRUNCATE TABLE statement is a SQL command that allows users to quickly > and efficiently delete all rows from a table without dropping the table > itself. This statement is commonly used in data warehouse, where large data > sets are frequently loaded and unloaded from tables. > So, this FLIP is meant to support TRUNCATE TABLE statement. M ore exactly, > this FLIP will bring Flink the TRUNCATE TABLE syntax and an interface with > which the coresponding connectors can implement their own logic for > truncating table. > > Looking forwards to your feedback. > > [1]: [ > https://cwiki.apache.org/confluence/display/FLINK/FLIP-302%3A+Support+TRUNCATE+TABLE+statement > | > https://cwiki.apache.org/confluence/display/FLINK/FLIP-302%3A+Support+TRUNCATE+TABLE+statement > ] > > > Best regards, > Yuxia >
Re: [DISCUSS] FLIP-304: Pluggable failure handling for Apache Flink
Hi Panagiotis Thank you for your answer. I agree that `FailureListener` could be stateless, then I have some thoughts as follows 1. I see that listeners and tag collections are associated. When JobManager fails and restarts, how can the new listener be associated with the tag collection before failover? Is the listener loading order? 2. The tag collection may be too large, resulting in the JobManager OOM, do we need to provide a management class that supports some obsolescence strategies instead of a direct Collection? 3. Is it possible to provide a more complex data structure than a simple string collection for tags in listeners, such as key-value? Best, Shammon FY On Mon, Mar 20, 2023 at 7:48 PM Leonard Xu wrote: > Hi，Panagiotis > > > Thank you for kicking off this discussion. Overall, the proposed feature of > this FLIP makes sense to me. We have also discussed similar requirements > with our users and developers, and I believe it will help many users. > > > In terms of FLIP content, I have some thoughts: > > (1) For the FailureListenerContextget interface, the methods > FailureListenerContext#addTag and FailureListenerContextgetTags looks very > inconsistent because they imply specific implementation details, and not > all FailureListeners need to handle them, we shouldn't put them in the > interface. Minor: The comment "UDF loading" in the getUserClassLoader() > method looks like a typo, IIUC it should return the classloader of the > current job. > > (2) Regarding the implementation in ExecutionFailureHandler#handleFailure, > some custom listeners may have heavy IO operations, such as reporting to > their monitoring system. The current logic appears to be processing in the > JobMaster's main thread, and it is recommended not to do this kind of > processing in the main thread. > > (3) The results of FailureListener's processing and the > FailureHandlingResult returned by ExecutionFailureHandler are not related. > I think these two are closely related, the motivation of this FLIP is to > make current failure handling more flexible. From this perspective, > different listeners should have the opportunity to affect the job's failure > handling flow. For example, a Flink job is configured with a > RestartStrategy with huge numbers retry , but the Kafka topic of Source has > been deleted, the job will failover continuously. In this case, the user > should have their listener to determine whether this failure is recoverable > or unrecoverable, and then wrap the processing result into > FailureHandlingResult.unrecoverable(xx) and pass it to JobMaster, this > approach will be more flexible. > > (4) All FLIPs have an important section named Public Interfaces. Current > FLIP mixes the interface section and the implementation section together. > It is better for us to refer to the FLIP template[1] and separate them, > this will make the entire FLIP clearer. > > In addition, regarding the FLIP process, there is a small suggestion: The > community generally creates a JIRA issue after the FLIP vote is passed, > instead of during the FLIP preparation phase because the FLIP may be > rejected. Although this FLIP is very reasonable, it's better to follow the > process. > > Best, > > Leonard > > [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP+Template > > On Mon, Mar 20, 2023 at 7:04 PM David Morávek wrote: > > > > > > > however listeners can use previous state (tags/labels) to make > decisions > > > > > > That sounds like a very fragile contract. We should either allow passing > > tags between listeners and then need to define ordering or make all of > them > > independent. I prefer the latter because it allows us to parallelize > things > > if needed (if all listeners trigger an RCP to the external system, for > > example). > > > > Can you expand on why we need more than one classifier to be able to > output > > the same tag? > > > > system ones come first and then the ones loaded from the plugin manager > > > > > > Since they're returned as a Set, the order is completely > non-deterministic, > > no matter in which order they're loaded. > > > > just communicating with external monitoring/alerting systems > > > > > > > That makes the need for pushing things out of the main thread even > > stronger. This almost sounds like we need to return a CompletableFuture > for > > the per-throwable classification because an external system might take a > > significant time to respond. We need to unblock the main thread for other > > RPCs. > > > > Also, in the proposal, this happens in the failure handler. If that's the > > case,
[DISCUSS] FLIP-304: Pluggable failure handling for Apache Flink
Hi Panagiotis Thank you for starting this discussion. I think this FLIP is valuable and can help user to analyze the causes of job failover better! I have two comments as follows 1. How about adding more job information in FailureListenerContext? For example, job vertext, subtask, taskmanager location. And then user can do more statistics according to different dimensions. 2. Users may want to save results in listener, and then they can get the historical results even jabmanager failover. Can we provide a unified implementation for data storage requirements? Best, shammon FY On Saturday, March 18, 2023, Panagiotis Garefalakis wrote: > Hi everyone, > > This FLIP [1] proposes a pluggable interface for failure handling allowing > users to implement custom failure logic using the plugin framework. > Motivated by existing proposals [2] and tickets [3], this enables use-cases > like: assigning particular types to failures (e.g., User or System), > emitting custom metrics per type (e.g., application or platform), even > exposing errors to downstream consumers (e.g., notification systems). > > Thanks to Piotr and Anton for the initial reviews and discussions! > > For anyone interested, the starting point would be the FLIP [1] that I > created, > describing the motivation and the proposed changes (part of the core, > runtime and web). > > The intuition behind this FLIP is being able to execute custom logic on > failures by exposing a FailureListener interface. Implementation by users > can be simply loaded to the system as Jar files. FailureListeners may also > decide to assign failure tags to errors (expressed as strings), > that will then be exposed as metadata by the UI/Rest interfaces. > > Feedback is always appreciated! Looking forward to your thoughts! > > [1] > https://cwiki.apache.org/confluence/display/FLINK/FLIP-304% > 3A+Pluggable+failure+handling+for+Apache+Flink > [2] > https://docs.google.com/document/d/1pcHg9F3GoDDeVD5GIIo2wO67 > Hmjgy0-hRDeuFnrMgT4 > [3] https://issues.apache.org/jira/browse/FLINK-20833 > > Cheers, > Panagiotis >
[RESULT][VOTE] FLIP-293: Introduce Flink Jdbc Driver For Sql Gateway
Hi everyone: I am happy to announce that FLIP-293: Introduce Flink Jdbc Driver For Sql Gateway [1] has been accepted. There are 6 binding votes and 5 non-binding votes: - Benchao Li (binding) - Jingsong Li (binding) - Jing Ge (binding) - weijie guo (binding) - Leonard Xu (binding) - Dong Lin (binding) - Ran Tao (non-binding) - Jacky Lau (non-binding) - Hang Ruan (non-binding) - Samrat Deb (non-binding) - ConradJam (non-binding) There is no disapproving vote. Thanks. [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-293%3A+Introduce+Flink+Jdbc+Driver+For+Sql+Gateway Best, Shammon FY
Re: [VOTE] FLIP-293: Introduce Flink Jdbc Driver For Sql Gateway
Hi everyone I'm closing this vote now. I will follow up with the result in another email. Thanks Best, Shammon FY On Thu, Mar 16, 2023 at 10:37 AM ConradJam wrote: > Hang Ruan ruanhang1...@gmail.com 通过 > <https://support.google.com/mail/answer/1311182?hl=zh-CN>“flink.apache.org > ” > 3月15日周三 17:09 (17小时前) > 发送至 dev > 英语 > 中文 > > 翻译邮件 > +1 (non-binding) > > Samrat Deb 于2023年3月15日周三 23:29写道： > > > +1 ( non binding) > > > > Bests, > > Samrat > > > > On Wed, 15 Mar 2023 at 8:54 PM, Dong Lin wrote: > > > > > +1 (binding) > > > > > > On Mon, Mar 13, 2023 at 1:47 PM Shammon FY wrote: > > > > > > > Hi Devs, > > > > > > > > I'd like to start the vote on FLIP-293: Introduce Flink Jdbc Driver > For > > > Sql > > > > Gateway [1]. > > > > > > > > The FLIP was discussed in thread [2], and it aims to introduce Flink > > Jdbc > > > > Driver module in Flink. > > > > > > > > The vote will last for at least 72 hours (03/16, 15:00 UTC+8) unless > > > there > > > > is an objection or insufficient vote. Thank you all. > > > > > > > > > > > > [1] > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-293%3A+Introduce+Flink+Jdbc+Driver+For+Sql+Gateway > > > > [2] https://lists.apache.org/thread/d1owrg8zh77v0xygcpb93fxt0jpjdkb3 > > > > > > > > > > > > Best, > > > > Shammon.FY > > > > > > > > > > > > -- > Best > > ConradJam >
Re: [DISCUSS] Enabling dynamic partition discovery by default in Kafka source
Hi Hongshun Agree with @Etienne. And can you describe this process and the problems in a figure? Thanks Best, Shammon FY On Wed, Mar 15, 2023 at 5:15 PM Etienne Chauchot wrote: > Hi, > > Why not track this in a FLIP and a ticket and link this discussion thread. > > My 2 cents > > Etienne > > Le 15/03/2023 à 10:01, Hongshun Wang a écrit : > > Hi devs, > > I’d like to join this discussion. CC:Qingsheng > > As discussed above, new partitions after the first discovery should be > > consumed from EARLIEST offset. > > > > However, when KafkaSourceEnumerator restarts after a job failure, it > cannot > > distinguish between unassigned partitions as first-discovered or new, > > because the snapshot state currently only contains assignedPartitions > > collection (the assigned partitions). We can solve this by adding a > > unAssignedInitialPartitons collection to snapshot state, which represents > > the collection of first discovered partitions that have not yet been > > assigned. Also, we can combine this two collections into a single > > collection if we add status to each item. > > > > Besides , there is also a problem which often occurs in pattern mode to > > distinguish between the following two case: > > > > 1. Case1: The first partition discovery is too slow, before which > the > > checkpoint is finished and then job is restarted .At this time, the > > restored unAssignedInitialPartitons is an empty collection, which > means > > non-discovery. The next discovery will be treated as the first > discovery. > > 2. Case2: The first time the partition is obtained is empty, and new > > partitions can only be obtained after multiple partition > discoveries. If a > > restart occurs between this period, the restored > > *unAssignedInitialPartitons* is also an empty collection, which means > > empty-discovery. However, the next discovery should be treated as a > new > > discovery. > > > > We can solve this problem by adding a boolean value(*firstDiscoveryDone*) > > to snapshot state, which represents whether the first-discovery has been > > done. > > > > Also two rejected alternatives : > > > > 1. Change the KafkaSourceEnumerator's snapshotState method to a > blocking > > one, which resumes only after the first-discovered partition has been > > successfully assigned to KafkaSourceReader. The advantage of this > approach > > is no need to change the snapshot state's variable values. However, > if > > first-discovered partitions are not assigned before checkpointing, > the > > SourceCoordinator's event-loop thread will be blocked, but partition > > assignment also requires the event-loop thread to execute, which > will cause > > thread self-locking. > > 2. An alternative to the *firstDiscoveryDone* variable. If we change > the > > first discovery method to a synchronous method, we can ensure that > Case1 > > will never happen. Because when the event-loop thread starts, it > first adds > > a discovery event to the blocking queue. When it turns to execute the > > checkpoint event, the partition has already been discovered > successfully. > > However, If partition discovery is a heavily time-consuming > operation, the > > SourceCoordinator cannot process other event operations during the > waiting > > period, such as reader registration. It is a waste. > > > > Best regards, > > Hongshun > > > > On 2023/01/13 03:31:20 Qingsheng Ren wrote: > >> Hi devs, > >> > >> I’d like to start a discussion about enabling the dynamic partition > >> discovery feature by default in Kafka source. Dynamic partition > discovery > >> [1] is a useful feature in Kafka source especially under the scenario > when > >> the consuming Kafka topic scales out, or the source subscribes to > multiple > >> Kafka topics with a pattern. Users don’t have to restart the Flink job > to > >> consume messages in the new partition with this feature enabled. > > Currently, > >> dynamic partition discovery is disabled by default and users have to > >> explicitly specify the interval of discovery in order to turn it on. > >> > >> # Breaking changes > >> > >> For Kafka table source: > >> > >> - “scan.topic-partition-discovery.interval” will be set to 30 seconds by > >> default. > >> - As we need to provide a way for users to disable the feature, > >> “scan.topic-partition-
Re: [DISCUSS] FLIP-293: Introduce Flink Jdbc Driver For Sql Gateway
Hi Jing I think your thoughts are interesting. We can consider a flink cluster with Jdbc-Driver as a warehouse and flink jobs can interact with it by Flink-Jdbc-Connector. This may have higher requirements for the throughput and latency of the warehouse (flink cluster), but we can try it after this FLIP Best, Shammon On Mon, Mar 13, 2023 at 3:52 PM Jing Ge wrote: > Hi, > > @Benchao > you are right, no, the question did not intend to block this FLIP. > > @Shammon > It looks quite interesting, if we imagine how the Flink jdbc connector > could work with Flink jdbc driver. It makes "on-the-fly" > cross-flink-applications data consumption possible, i.e. bounded source > scan, lookup source from another flink job, and sink to another flink job. > The streaming append mode of the sink could sort of turn the data pipeline > into streaming mode. All of these could not be done in Flink jobs directly, > if I am not mistaken. > > Best regards, > Jing > > > On Mon, Mar 13, 2023 at 3:38 AM Benchao Li wrote: > > > Hi Jing & Shammon, > > > > IMO, Jdbc is a Java standard and Flink-JDBC-Driver is an implementation > to > > interact with Flink, so in theory, it would be doable to use > > Flink-JDBC-Driver in Flink-JDBC-Connector if we add it as a dialect. > > > > However, I don't see a scenario which requires this yet. All in all, it > > would be an orthogonal topic, which does not block current FLIP > discussion. > > > > Shammon FY 于2023年3月13日周一 08:47写道： > > > > > Hi Jing > > > > > > Flink-Jdbc-Driver is mainly used to facilitate users to submit flink > sql > > > jobs. I think we will not use Flink-Jdbc-Driver in Flink-Jdbc-Connector > > to > > > read/write data in general. If users can read/write specific tables by > > > Flink-Jdbc-Driver, they can use these tables in their flink jobs > > directly. > > > Thanks > > > > > > Best, > > > Shammon > > > > > > > > > On Mon, Mar 13, 2023 at 6:21 AM Jing Ge > > > wrote: > > > > > > > Hi, Shammon, > > > > > > > > This FLIP is very interesting. Thanks for driving it! Just out of > > > > curiosity, Could it be used in the Flink JDBC connector[1] after > > > > doing the appropriate implementation there? > > > > > > > > Best regards, > > > > Jing > > > > > > > > [1] > > > > > > > > > > > > > > https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/jdbc/ > > > > > > > > On Fri, Mar 3, 2023 at 8:56 AM Shammon FY wrote: > > > > > > > > > Hi, > > > > > > > > > > Thanks jinsong. I think implementing JDBC for batch mode first > sounds > > > > good. > > > > > This will simplify the implementation and we can also remove the > row > > > kind > > > > > first. We can claim this in the FLIP and docs, I will update the > > FLIP. > > > > > > > > > > Best, > > > > > Shammon > > > > > > > > > > > > > > > On Fri, Mar 3, 2023 at 2:36 PM Jingsong Li > > > > > wrote: > > > > > > > > > > > Hi, Shammon, > > > > > > > > > > > > I took a look at JDBC `ResultSet` and `Statement`. They are > > > > > > complicated and have many interfaces. Some of the interfaces may > > not > > > > > > be very suitable for streaming. > > > > > > > > > > > > I think maybe we can just implement JDBC for batch/olap only. It > is > > > > > > hard to have an integration for JDBC and streaming... > > > > > > > > > > > > Do you need to use JDBC in streaming mode? Or do we just > implement > > > > > > JDBC for batch only first? > > > > > > > > > > > > Best, > > > > > > Jingsong > > > > > > > > > > > > > > > > > > On Thu, Mar 2, 2023 at 6:22 PM Shammon FY > > wrote: > > > > > > > > > > > > > > Hi > > > > > > > > > > > > > > Thanks for the feedback from Jingsong and Benchao. > > > > > > > > > > > > > > For @Jingsong > > > > > > > > If the user does not cast into a FlinkResultSet, will there > be > > > > >
[VOTE] FLIP-293: Introduce Flink Jdbc Driver For Sql Gateway
Hi Devs, I'd like to start the vote on FLIP-293: Introduce Flink Jdbc Driver For Sql Gateway [1]. The FLIP was discussed in thread [2], and it aims to introduce Flink Jdbc Driver module in Flink. The vote will last for at least 72 hours (03/16, 15:00 UTC+8) unless there is an objection or insufficient vote. Thank you all. [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-293%3A+Introduce+Flink+Jdbc+Driver+For+Sql+Gateway [2] https://lists.apache.org/thread/d1owrg8zh77v0xygcpb93fxt0jpjdkb3 Best, Shammon.FY
Re: Re: [ANNOUNCE] New Apache Flink Committer - Yuxia Luo
Congratulations, Yuxia On Mon, Mar 13, 2023 at 12:11 PM Jingsong Li wrote: > Congratulations, Yuxia! > > On Mon, Mar 13, 2023 at 11:49 AM Juntao Hu wrote: > > > > Congratulations, Yuxia! > > > > Best, > > Juntao > > > > > > Wencong Liu 于2023年3月13日周一 11:33写道： > > > > > Congratulations, Yuxia! > > > > > > Best, > > > Wencong Liu > > > > > > > > > At 2023-03-13 11:20:21, "Qingsheng Ren" wrote: > > > >Congratulations, Yuxia! > > > > > > > >Best, > > > >Qingsheng > > > > > > > >On Mon, Mar 13, 2023 at 10:27 AM Jark Wu wrote: > > > > > > > >> Hi, everyone > > > >> > > > >> On behalf of the PMC, I'm very happy to announce Yuxia Luo as a new > > > Flink > > > >> Committer. > > > >> > > > >> Yuxia has been continuously contributing to the Flink project for > almost > > > >> two > > > >> years, authored and reviewed hundreds of PRs over this time. He is > > > >> currently > > > >> the core maintainer of the Hive component, where he contributed many > > > >> valuable > > > >> features, including the Hive dialect with 95% compatibility and > small > > > file > > > >> compaction. > > > >> In addition, Yuxia driven FLIP-282 (DELETE & UPDATE API) to better > > > >> integrate > > > >> Flink with data lakes. He actively participated in dev discussions > and > > > >> answered > > > >> many questions on the user mailing list. > > > >> > > > >> Please join me in congratulating Yuxia Luo for becoming a Flink > > > Committer! > > > >> > > > >> Best, > > > >> Jark Wu (on behalf of the Flink PMC) > > > >> > > > >
Re: [DISCUSS] FLIP-293: Introduce Flink Jdbc Driver For Sql Gateway
Hi Jing Flink-Jdbc-Driver is mainly used to facilitate users to submit flink sql jobs. I think we will not use Flink-Jdbc-Driver in Flink-Jdbc-Connector to read/write data in general. If users can read/write specific tables by Flink-Jdbc-Driver, they can use these tables in their flink jobs directly. Thanks Best, Shammon On Mon, Mar 13, 2023 at 6:21 AM Jing Ge wrote: > Hi, Shammon, > > This FLIP is very interesting. Thanks for driving it! Just out of > curiosity, Could it be used in the Flink JDBC connector[1] after > doing the appropriate implementation there? > > Best regards, > Jing > > [1] > > https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/jdbc/ > > On Fri, Mar 3, 2023 at 8:56 AM Shammon FY wrote: > > > Hi, > > > > Thanks jinsong. I think implementing JDBC for batch mode first sounds > good. > > This will simplify the implementation and we can also remove the row kind > > first. We can claim this in the FLIP and docs, I will update the FLIP. > > > > Best, > > Shammon > > > > > > On Fri, Mar 3, 2023 at 2:36 PM Jingsong Li > wrote: > > > > > Hi, Shammon, > > > > > > I took a look at JDBC `ResultSet` and `Statement`. They are > > > complicated and have many interfaces. Some of the interfaces may not > > > be very suitable for streaming. > > > > > > I think maybe we can just implement JDBC for batch/olap only. It is > > > hard to have an integration for JDBC and streaming... > > > > > > Do you need to use JDBC in streaming mode? Or do we just implement > > > JDBC for batch only first? > > > > > > Best, > > > Jingsong > > > > > > > > > On Thu, Mar 2, 2023 at 6:22 PM Shammon FY wrote: > > > > > > > > Hi > > > > > > > > Thanks for the feedback from Jingsong and Benchao. > > > > > > > > For @Jingsong > > > > > If the user does not cast into a FlinkResultSet, will there be > > > > serious consequences here (RowKind is ignored)? > > > > > > > > I agree with you that it's indeed a big deal if users ignore the row > > kind > > > > when they must know it. One idea that comes to my mind is we can add > an > > > > option such as `table.result.changelog-mode`, users can set it > through > > > > connection properties or set dynamic parameters. The option value can > > be > > > > `insert-only`, `upset` or `all` and the default value is > `insert-only`. > > > > > > > > If the result does not conform to the changelog mode, the jdbc driver > > > > throws an exception. What do you think? > > > > > > > > > > > > For @Benchao > > > > > Besides `java.sql.Driver`, have you considered also adding support > > for > > > > `javax.sql.DataSource` interface? > > > > > > > > I missed the `javax.sql.DataSource` and I have added it to the FLIP, > > > thanks > > > > Benchao > > > > > > > > > > > > Best, > > > > Shammon > > > > > > > > On Wed, Mar 1, 2023 at 7:57 PM Benchao Li > > wrote: > > > > > > > > > +1 for the FLIP, thanks Shammon for driving this. > > > > > > > > > > JDBC is quite useful in OLAP scenarios, supporting JDBC would > enable > > > Flink > > > > > to be used with existing tools, such as Tableau. > > > > > > > > > > Regarding the JDBC interfaces listed in the FLIP, I think they > looks > > > good > > > > > already. Besides `java.sql.Driver`, have you considered also adding > > > support > > > > > for `javax.sql.DataSource` interface? > > > > > > > > > > Jingsong Li 于2023年3月1日周三 17:53写道： > > > > > > > > > > > Thanks Shammon for driving. > > > > > > > > > > > > Big +1 for this. > > > > > > > > > > > > I heard that many users want to use FlinkGateway + JDBC to do > some > > > > > > queries, but at present, only Hive JDBC can be used. It is Hive > > > > > > dialect by default, and the experience is also different from > > > > > > FlinkSQL. We need to have our own JDBC. > > > > > > > > > > > > I took a look at your `Public Interface` part, only > > > > > > `FlinkResultSet.getRowKind` is a true new int
Re: [Vote] FLIP-298: Unifying the Implementation of SlotManager
Thanks weihua, +1 (non-binding) Best, Shammon On Fri, Mar 10, 2023 at 10:32 AM Xintong Song wrote: > +1 (binding) > > Best, > > Xintong > > > > On Thu, Mar 9, 2023 at 1:28 PM Weihua Hu wrote: > > > Hi Everyone, > > > > I would like to start the vote on FLIP-298: Unifying the Implementation > > of SlotManager [1]. The FLIP was discussed in this thread [2]. > > > > This FLIP aims to unify the implementation of SlotManager in > > order to reduce maintenance costs. > > > > The vote will last for at least 72 hours (03/14, 15:00 UTC+8) > > unless there is an objection or insufficient votes. Thank you all. > > > > [1] > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-298%3A+Unifying+the+Implementation+of+SlotManager > > [2]https://lists.apache.org/thread/ocssfxglpc8z7cto3k8p44mrjxwr67r9 > > > > Best, > > Weihua > > >
Re: [DISCUSS] FLIP-298: Unifying the Implementation of SlotManager
otManager implementations with where > > > > >> their functionality matches and where they differ might help > > > understand > > > > the > > > > >> consequences of the changes proposed in FLIP-298 > > > > > > > > > > Good suggestion, I have updated the comparison in this FLIP. > Looking > > > > > forward to any suggestions/thoughts > > > > > if they are not described clearly. > > > > > > > > > > *@John > > > > > > > > > > 4. In addition to changing the default, would it make sense to log > a > > > > >> deprecation warning on initialization > > > > > > > > > > if the DeclarativeSlotManager is used? > > > > >> > > > > > SGTM, We should add Deprecated annotations to DSM for devs. And > log a > > > > > deprecation warning for users. > > > > > > > > > > *@Shammon > > > > > > > > > > 1. For their functional differences, can you give some detailed > tests > > > to > > > > >> verify that the new FineGrainedSlotManager has these capabilities? > > > This > > > > can > > > > >> effectively verify the new functions > > > > >> > > > > > As just maintained, there is already a CI stage of FGSM, and I will > > do > > > > more > > > > > review of unit tests for DSM. > > > > > > > > > > 2. I'm worried that many functions are not independent and it is > > > > difficult > > > > >> to migrate step-by-step. You can list the relationship between > them > > in > > > > >> detail. > > > > > > > > > > As Xintong saied the DSM is a subset of FGSM by design. But as > time > > > goes > > > > > on, FGSM has some lacking > > > > > functions as I listed in this FLIP. And I have added the comparison > > > > between > > > > > DSM and FGSM in this FLIP. > > > > > > > > > > > > > > > Thanks again for all your thoughts. Any feedback is appreciated! > > > > > > > > > > Best, > > > > > Weihua > > > > > > > > > > > > > > > On Wed, Mar 1, 2023 at 2:17 PM Xintong Song > > > > > wrote: > > > > > > > > > >> Thanks Weihua for preparing this FLIP. +1 for the proposal. > > > > >> > > > > >> > > > > >> As one of the contributors of the fine-grained slot manager, I'd > > like > > > to > > > > >> share some backgrounds here. > > > > >> > > > > >> - There used to be a defaut slot manager implementation, which is > > > > >> non-declarative and has been removed now. The two features, > > > declarative > > > > / > > > > >> reactive resource management and fine-grained resource management, > > > were > > > > >> proposed at about the same time. We were aware that by design the > > > > >> declarative slot manager is a subset of fine-grained slot manager > at > > > > that > > > > >> time, but still decided to implement two slot managers for the > > purpose > > > > of > > > > >> decoupling efforts and reducing cross-team synchronization > overhead. > > > > >> Merging the two slot managers once they are proved stable is IMO a > > > > >> technical debt that was planned at the very beginning. > > > > >> > > > > >> - The FineGrainedSlotManager has been verified in Alibaba's > internal > > > > >> production as well as Alibaba Cloud services as the default slot > > > manager > > > > >> for about 2 years. > > > > >> > > > > >> > > > > >> Concerning test cases, we currently have a ci stage for fine > grained > > > > >> resource management. To avoid adding too much burden, the stage > only > > > > >> includes tests from flink-runtime and flink-test modules. I think > > > > switching > > > > >> the default slot manager and applying the whole set of tests on > the > > > > >> fine-grained slot manager would help us to be more confident about > > it. > > > > >> &g
Re: [DISCUSS] FLIP-293: Introduce Flink Jdbc Driver For Sql Gateway
Hi, Thanks jinsong. I think implementing JDBC for batch mode first sounds good. This will simplify the implementation and we can also remove the row kind first. We can claim this in the FLIP and docs, I will update the FLIP. Best, Shammon On Fri, Mar 3, 2023 at 2:36 PM Jingsong Li wrote: > Hi, Shammon, > > I took a look at JDBC `ResultSet` and `Statement`. They are > complicated and have many interfaces. Some of the interfaces may not > be very suitable for streaming. > > I think maybe we can just implement JDBC for batch/olap only. It is > hard to have an integration for JDBC and streaming... > > Do you need to use JDBC in streaming mode? Or do we just implement > JDBC for batch only first? > > Best, > Jingsong > > > On Thu, Mar 2, 2023 at 6:22 PM Shammon FY wrote: > > > > Hi > > > > Thanks for the feedback from Jingsong and Benchao. > > > > For @Jingsong > > > If the user does not cast into a FlinkResultSet, will there be > > serious consequences here (RowKind is ignored)? > > > > I agree with you that it's indeed a big deal if users ignore the row kind > > when they must know it. One idea that comes to my mind is we can add an > > option such as `table.result.changelog-mode`, users can set it through > > connection properties or set dynamic parameters. The option value can be > > `insert-only`, `upset` or `all` and the default value is `insert-only`. > > > > If the result does not conform to the changelog mode, the jdbc driver > > throws an exception. What do you think? > > > > > > For @Benchao > > > Besides `java.sql.Driver`, have you considered also adding support for > > `javax.sql.DataSource` interface? > > > > I missed the `javax.sql.DataSource` and I have added it to the FLIP, > thanks > > Benchao > > > > > > Best, > > Shammon > > > > On Wed, Mar 1, 2023 at 7:57 PM Benchao Li wrote: > > > > > +1 for the FLIP, thanks Shammon for driving this. > > > > > > JDBC is quite useful in OLAP scenarios, supporting JDBC would enable > Flink > > > to be used with existing tools, such as Tableau. > > > > > > Regarding the JDBC interfaces listed in the FLIP, I think they looks > good > > > already. Besides `java.sql.Driver`, have you considered also adding > support > > > for `javax.sql.DataSource` interface? > > > > > > Jingsong Li 于2023年3月1日周三 17:53写道： > > > > > > > Thanks Shammon for driving. > > > > > > > > Big +1 for this. > > > > > > > > I heard that many users want to use FlinkGateway + JDBC to do some > > > > queries, but at present, only Hive JDBC can be used. It is Hive > > > > dialect by default, and the experience is also different from > > > > FlinkSQL. We need to have our own JDBC. > > > > > > > > I took a look at your `Public Interface` part, only > > > > `FlinkResultSet.getRowKind` is a true new interface, others are just > > > > implementations. > > > > > > > > If the user does not cast into a FlinkResultSet, will there be > serious > > > > consequences here (RowKind is ignored)? > > > > > > > > Best, > > > > Jingsong > > > > > > > > On Wed, Mar 1, 2023 at 4:59 PM Shammon FY wrote: > > > > > > > > > > Hi devs, > > > > > > > > > > I'd like to start a discussion about FLIP-293: Introduce Flink Jdbc > > > > Driver > > > > > For Sql Gateway[1]. > > > > > > > > > > FLIP-275[2] supports remote sql client based on gateway, users can > > > > interact > > > > > with gateway by flink console. However, for users who create > session > > > > > clusters with Flink, they'd like to use Jdbc Driver to interact > with > > > the > > > > > gateway in their applications, such as olap queries.. > > > > > > > > > > I have discussed this proposal with @shengkaifang and @jinsonglee. > In > > > > this > > > > > FLIP, we'd like to introduce Jdbc Driver for gateway. Users can use > > > Jdbc > > > > > Driver to submit their queries and get results like a database in > their > > > > > applications. > > > > > > > > > > Looking forward to your feedback, thanks. > > > > > > > > > > > > > > > [1] > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-293%3A+Introduce+Flink+Jdbc+Driver+For+Sql+Gateway > > > > > [2] > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-275%3A+Support+Remote+SQL+Client+Based+on+SQL+Gateway > > > > > > > > > > > > > > > Best, > > > > > Shammon > > > > > > > > > > > > > -- > > > > > > Best, > > > Benchao Li > > > >
Re: [DISCUSS] FLIP-293: Introduce Flink Jdbc Driver For Sql Gateway
Hi Thanks for the feedback from Jingsong and Benchao. For @Jingsong > If the user does not cast into a FlinkResultSet, will there be serious consequences here (RowKind is ignored)? I agree with you that it's indeed a big deal if users ignore the row kind when they must know it. One idea that comes to my mind is we can add an option such as `table.result.changelog-mode`, users can set it through connection properties or set dynamic parameters. The option value can be `insert-only`, `upset` or `all` and the default value is `insert-only`. If the result does not conform to the changelog mode, the jdbc driver throws an exception. What do you think? For @Benchao > Besides `java.sql.Driver`, have you considered also adding support for `javax.sql.DataSource` interface? I missed the `javax.sql.DataSource` and I have added it to the FLIP, thanks Benchao Best, Shammon On Wed, Mar 1, 2023 at 7:57 PM Benchao Li wrote: > +1 for the FLIP, thanks Shammon for driving this. > > JDBC is quite useful in OLAP scenarios, supporting JDBC would enable Flink > to be used with existing tools, such as Tableau. > > Regarding the JDBC interfaces listed in the FLIP, I think they looks good > already. Besides `java.sql.Driver`, have you considered also adding support > for `javax.sql.DataSource` interface? > > Jingsong Li 于2023年3月1日周三 17:53写道： > > > Thanks Shammon for driving. > > > > Big +1 for this. > > > > I heard that many users want to use FlinkGateway + JDBC to do some > > queries, but at present, only Hive JDBC can be used. It is Hive > > dialect by default, and the experience is also different from > > FlinkSQL. We need to have our own JDBC. > > > > I took a look at your `Public Interface` part, only > > `FlinkResultSet.getRowKind` is a true new interface, others are just > > implementations. > > > > If the user does not cast into a FlinkResultSet, will there be serious > > consequences here (RowKind is ignored)? > > > > Best, > > Jingsong > > > > On Wed, Mar 1, 2023 at 4:59 PM Shammon FY wrote: > > > > > > Hi devs, > > > > > > I'd like to start a discussion about FLIP-293: Introduce Flink Jdbc > > Driver > > > For Sql Gateway[1]. > > > > > > FLIP-275[2] supports remote sql client based on gateway, users can > > interact > > > with gateway by flink console. However, for users who create session > > > clusters with Flink, they'd like to use Jdbc Driver to interact with > the > > > gateway in their applications, such as olap queries.. > > > > > > I have discussed this proposal with @shengkaifang and @jinsonglee. In > > this > > > FLIP, we'd like to introduce Jdbc Driver for gateway. Users can use > Jdbc > > > Driver to submit their queries and get results like a database in their > > > applications. > > > > > > Looking forward to your feedback, thanks. > > > > > > > > > [1] > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-293%3A+Introduce+Flink+Jdbc+Driver+For+Sql+Gateway > > > [2] > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-275%3A+Support+Remote+SQL+Client+Based+on+SQL+Gateway > > > > > > > > > Best, > > > Shammon > > > > > -- > > Best, > Benchao Li >
Re: [VOTE] FLIP-291: Externalized Declarative Resource Management
+1 (non-binding) Best, Shammon On Wed, Mar 1, 2023 at 4:51 PM Roman Khachatryan wrote: > +1 (binding) > > Thanks David, and everyone involved :) > > Regards, > Roman > > > On Wed, Mar 1, 2023 at 8:01 AM Gyula Fóra wrote: > > > +1 (binding) > > > > Looking forward to this :) > > > > Gyula > > > > On Wed, 1 Mar 2023 at 04:02, feng xiangyu wrote: > > > > > +1 (non-binding) > > > > > > ConradJam 于2023年3月1日周三 10:37写道： > > > > > > > +1 (non-binding) > > > > > > > > Zhanghao Chen 于2023年3月1日周三 10:18写道： > > > > > > > > > Thanks for driving this. +1 (non-binding) > > > > > > > > > > Best, > > > > > Zhanghao Chen > > > > > > > > > > From: David Mor?vek > > > > > Sent: Tuesday, February 28, 2023 21:46 > > > > > To: dev > > > > > Subject: [VOTE] FLIP-291: Externalized Declarative Resource > > Management > > > > > > > > > > Hi Everyone, > > > > > > > > > > I want to start the vote on FLIP-291: Externalized Declarative > > Resource > > > > > Management [1]. The FLIP was discussed in this thread [2]. > > > > > > > > > > The goal of the FLIP is to enable external declaration of the > > resource > > > > > requirements of a running job. > > > > > > > > > > The vote will last for at least 72 hours (Friday, 3rd of March, > 15:00 > > > > CET) > > > > > unless > > > > > there is an objection or insufficient votes. > > > > > > > > > > [1] > https://lists.apache.org/thread/b8fnj127jsl5ljg6p4w3c4wvq30cnybh > > > > > [2] > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-291%3A+Externalized+Declarative+Resource+Management > > > > > > > > > > Best, > > > > > D. > > > > > > > > > > > > > > > > > -- > > > > Best > > > > > > > > ConradJam > > > > > > > > > >
[DISCUSS] FLIP-293: Introduce Flink Jdbc Driver For Sql Gateway
Hi devs, I'd like to start a discussion about FLIP-293: Introduce Flink Jdbc Driver For Sql Gateway[1]. FLIP-275[2] supports remote sql client based on gateway, users can interact with gateway by flink console. However, for users who create session clusters with Flink, they'd like to use Jdbc Driver to interact with the gateway in their applications, such as olap queries.. I have discussed this proposal with @shengkaifang and @jinsonglee. In this FLIP, we'd like to introduce Jdbc Driver for gateway. Users can use Jdbc Driver to submit their queries and get results like a database in their applications. Looking forward to your feedback, thanks. [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-293%3A+Introduce+Flink+Jdbc+Driver+For+Sql+Gateway [2] https://cwiki.apache.org/confluence/display/FLINK/FLIP-275%3A+Support+Remote+SQL+Client+Based+on+SQL+Gateway Best, Shammon
Re: [DISCUSS] FLIP-298: Unifying the Implementation of SlotManager
Hi Thanks for starting this work weihua, I think unifying DeclarativeSlotManager and FineGrainedSlotManager is valuable. I agree with @Matthias and @John that we need a way to ensure that DeclarativeSlotManager's capabilities are fully covered by FineGrainedSlotManager 1. For their functional differences, can you give some detailed tests to verify that the new FineGrainedSlotManager has these capabilities? This can effectively verify the new functions 2. I'm worried that many functions are not independent and it is difficult to migrate step-by-step. You can list the relationship between them in detail. 3. As John mentioned, give a smoke test for FineGrainedSlotManager is a good idea. Or you can add some test information to the DeclarativeSlotManager to determine how many tests have used it. In this way, we can gradually construct test cases for FineGrainedSlotManager during the development process. Best, Shammon On Tue, Feb 28, 2023 at 10:22 PM John Roesler wrote: > Thanks for the FLIP, Weihua! > > I’ve read the FLIP, and it sounds good to me. We need to avoid > proliferating alternative implementations wherever possible. I have just a > couple of comments: > > 1. I share Matthias’s concern about ensuring the behavior is really the > same. One suggestion I’ve used for this kind of thing is, as a smoke test, > to update the DeclarativeSlotManager to just delegate to the > FineGrainedSlotManager. If the full test suite still passes, you can be > pretty sure the new default is really ok. It would not be a good idea to > actually keep that in for the release, since it would remove the option to > fall back in case of bugs. Either way, we need to make sure all test > scenarios are present for the FGSM. > > 4. In addition to changing the default, would it make sense to log a > deprecation warning on initialization if the DeclarativeSlotManager is used? > > Thanks again, > John > > On Tue, Feb 28, 2023, at 07:20, Matthias Pohl wrote: > > Hi Weihua, > > Thanks for your proposal. From a conceptual point: AFAIU, the > > DeclarativeSlotManager covers a subset (i.e. only evenly sized slots) of > > what the FineGrainedSlotManager should be able to achieve (variable slot > > size per task manager). Is this the right assumption/understanding? In > this > > sense, merging both implementations into a single one sounds good. A few > > more general comments, though: > > > > 1. Did you do a proper test coverage analysis? That's not mentioned in > the > > current version of the FLIP. I'm bringing this up because we ran into the > > same issue when fixing the flaws that popped up after introducing the > > multi-component leader election (see FLIP-285 [1]). There is a risk that > by > > removing the legacy code we decrease test coverage because certain > > test cases that were covered for the legacy classes might not be > > necessarily covered in the new implementation, yet (see FLINK-30338 [2] > > which covers this issue for the leader election case). Ideally, we don't > > want to remove test cases accidentally because they were only implemented > > for the DeclarativeSlotManager but missed for the FineGrainedSlotManager. > > > > 2. DeclarativeSlotManager and FineGrainedSlotManager feel quite big in > > terms of lines of code. Without knowing whether it's actually a > reasonable > > thing to do: Instead of just adding more features to the > > FineGrainedSlotManager, have you thought of cutting out functionality > into > > smaller sub-components along this refactoring? Such a step-by-step > approach > > might improve the overall codebase and might make reviewing the > refactoring > > easier. I did a first pass over the code and struggled to identify code > > blocks that could be moved out of the SlotManager implementation(s). > > Therefore, I might be wrong with this proposal. I haven't worked on this > > codebase in that detail that it would allow me to come up with a > judgement > > call. I wanted to bring it up, anyway, because I'm curious whether that > > could be an option. There's a comment created by Chesnay (CC'd) in the > > JavaDoc of TaskExecutorManager [3] indicating something similar. I'm > > wondering whether he can add some insights here. > > > > 3. For me personally, having a more detailed summary comparing the > > subcomponents of both SlotManager implementations with where > > their functionality matches and where they differ might help understand > the > > consequences of the changes proposed in FLIP-298. > > > > Best, > > Matthias > > > > [1] > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-285%3A+Refactoring+LeaderElection+to+make+Flink+support+multi-component+leader+election+out-of-the-box > > [2] https://issues.apache.org/jira/browse/FLINK-30338 > > [3] > > > https://github.com/apache/flink/blob/f611ea8cb5deddb42429df2c99f0c68d7382e9bd/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/TaskExecutorManager.java#L66-L68 > > > > On Tue, Feb 28, 2023 at 6:14
[DISCUSS] Deprecate deserialize method in DeserializationSchema
Hi devs Currently there are two deserialization methods in `DeserializationSchema` 1. `T deserialize(byte[] message)`, only deserialize one record from binary, if there is no record it should return null. 2. `void deserialize(byte[] message, Collector out)`, supports deserializing none, one or multiple records gracefully, it can completely replace method `T deserialize(byte[] message)`. The deserialization logic in the above two methods is basically coincident, we recommend users use the second method to deserialize data. To improve code maintainability, I'd like to mark the first function as `@Deprecated`, and remove it when it is no longer used in the future. I have created an issue[1] to track it, looking forward to your feedback, thanks [1] https://issues.apache.org/jira/browse/FLINK-31251 Best, Shammon
Re: [Discuss] Some questions on flink-table-store micro benchmark
Hi jingsong Getting rid of JMH is a good idea. For the second point, how can we track the performance changes of the micro benchmark? What do you think? Best, Shammon On Mon, Feb 27, 2023 at 10:57 AM Jingsong Li wrote: > Thanks Yun. > > Another way is we can get rid of JMH, something like Spark > `org.apache.spark.benchmark.Benchmark` can replace JMH. > > Best, > Jingsong > > On Mon, Feb 27, 2023 at 1:24 AM Yun Tang wrote: > > > > Hi dev, > > > > I just noticed that flink-table-store had introduced the micro benchmark > module [1] to test the basic performance. And I have two questions here. > > First of all, we might not be able to keep the micro benchmark, which is > based on JMH, in the main repo of flink-table-store. This is because JMH is > under GPL license, which is not compliant with Apache-2 license. That's why > Flink moved the flink-benchmark out [2]. > > > > Secondly, I try to run the micro benchmark locally but it seems failed, > I just wonder can we make the flink-table-store's micro benchmark could be > periodically executed just as flink-benchmarks did on > http://codespeed.dak8s.net:8080? Please correct me if such daily > benchmark has been set up. > > Moreover, maybe we can also consider to integrate this micro benchmark > notification in the Slack channel just as what flink-benchmarks did. > > > > [1] https://issues.apache.org/jira/browse/FLINK-29636 > > [2] https://issues.apache.org/jira/browse/FLINK-2973 > > > > Best > > Yun Tang >
Re: [DISCUSS] FLIP-296: Watermark options for table API & SQL
Hi kui Thanks for your answer and +1 to yuxia too > we should not bind the watermark-related options to a connector to ensure semantic clarity. In my opinion, adding watermark-related options to a connector is much more clear. Currently users can define simple watermark strategy in DDL, adding more configuration items in connector options is easy to understand Best, Shammon On Thu, Feb 23, 2023 at 10:52 AM Jingsong Li wrote: > Thanks for your proposal. > > +1 to yuxia, consider watermark-related hints as option hints. > > Personally, I am cautious about adding SQL syntax, WATERMARK_PARAMS is > also SQL syntax to some extent. > > We can use OPTIONS to meet this requirement if possible. > > Best, > Jingsong > > On Thu, Feb 23, 2023 at 10:41 AM yuxia > wrote: > > > > Hi, Yuan Kui. > > Thanks for driving it. > > IMO, the 'OPTIONS' hint may be not only specific to the connector > options. Just as a reference, we also have `sink.parallelism`[1] as a > connector options. It enables > > user to specific the writer's parallelism dynamically per-query. > > > > Personally, I perfer to consider watermark-related hints as option > hints. So, user can define a default watermark strategy for the table, and > if user dosen't needed to changes it, they need to do nothing in their > query instead of specific it ervery time. > > > > [1] > https://nightlies.apache.org/flink/flink-docs-master/zh/docs/connectors/table/filesystem/#sink-parallelism > > > > Best regards, > > Yuxia > > > > Best regards, > > Yuxia > > > > - 原始邮件 - > > 发件人: "kui yuan" > > 收件人: "dev" > > 抄送: "Jark Wu" > > 发送时间: 星期三, 2023年 2 月 22日 下午 10:08:11 > > 主题: Re: [DISCUSS] FLIP-296: Watermark options for table API & SQL > > > > Hi all, > > > > Thanks for the lively discussion and I will respond to these questions > one > > by one. However, there are also some common questions and I will answer > > together. > > > > @郑 Thanks for your reply. The features mentioned in this flip are only > for > > those source connectors that implement the SupportsWatermarkPushDown > > interface, generating watermarks in other graph locations is not in the > > scope of this discussion. Perhaps another flip can be proposed later to > > implement this feature. > > > > @Shammon Thanks for your reply. In Flip-296, a rejected alternative is > > adding watermark related options in the connector options，we believe that > > we should not bind the watermark-related options to a connector to ensure > > semantic clarity. > > > > > What will happen if we add watermark related options in `the connector > > > options`? Will the connector ignore these options or throw an > exception? > > > How can we support this? > > > > If user defines different watermark configurations for one table in two > > places, I tend to prefer the first place would prevail, but we can also > > throw exception or just print logs to prompt the user, which are > > implementation details. > > > > > If one table is used by two operators with different watermark params, > > > what will happen? > > > > @Martijn Thanks for your reply. I'm sorry that we are not particularly > > accurate, this hint is mainly for SQL, not table API. > > > > > While the FLIP talks about watermark options for Table API & SQL, I > only > > > see proposed syntax for SQL, not for the Table API. What is your > proposal > > > for the Table API > > > > @Jane Thanks for your reply. For the first question, If the user uses > this > > hint on those sourse that does not implement the > SupportsWatermarkPushDown > > interface, it will be completely invalid. The task will run as normal as > if > > the hint had not been used. > > > > > What's the behavior if there are multiple table sources, among which > > > some do not support `SupportsWatermarkPushDown`? > > > > @Jane feedback that 'WATERMARK_PARAMS' is difficult to remember, perhaps > > the naming issue can be put to the end of the discussion, because more > > people like @Martijn @Shuo are considering whether these configurations > > should be put into the DDL or the 'OPTIONS' hint. Here's what I > > think, Putting these configs into DDL or putting them into 'OPTIONS' hint > > is actually the same thing, because the 'OPTIONS' hint is mainly used to > > configure the properties of conenctor. The reason why I want to use a new > > hint is to make sure the semantics clear, in my opinion the configuration > > of watermark should not be mixed up with connector. However, a new hint > > does make it more difficult to use to some extent, for example, when a > user > > uses both 'OPTIONS' hint and 'WATERMARK_PARAMS' hint. For this point, > maby > > it is more appropriate to use uniform 'OPTIONS' hint. > > On the other hand, if we enrich more watermark option keys in 'OPTIONS' > > hints, The question will be what we treat the definatrions of 'OPTIONS' > > hint, is this only specific to the connector options or could be more? > > Maybe @Jark could share more insights here. In my opion,
Re: [DISCUSS] FLIP-296: Watermark options for table API & SQL
Hi kui Thanks for initializing this discussion. I have two questions 1. What will happen if we add watermark related options in `the connector options`? Will the connector ignore these options or throw an exception? How can we support this? 2. If one table is used by two operators with different watermark params, what will happen? For example, there is a source table T and user submit the following job SELECT T1.id, T1.cnt_val, T2.sum_val FROM (SELECT id, count(val) as cnt_val FROM T /*+ WATERMARK PARAMS(‘rowtime’ = ’time_column’, ‘rowtime_expression’ = ’time_column - INTERVAL 5 SECOND') */ GROUP BY id) T1 JOIN (SELECT id, sum(val) as sum_val FROM T /*+ WATERMARK PARAMS(‘rowtime’ = ’time_column’, ‘rowtime_expression’ = ’time_column - INTERVAL 10 SECOND') */ GROUP BY id) T2 ON T1.id=T2.id Best, Shammon On Wed, Feb 22, 2023 at 11:28 AM 郑舒力 wrote: > Hi kui, > > There is a scenario using watermark that cannot be implemented through SQL > now： > We only support specify rowtime column in table DDL，if the rowtime field > is generated by join dimension table , it cannot be implemented > > Can we consider implement through HINTS like : > Select * from t1 join t2 /*+ WATERMARK PARAMS(‘rowtime’ = ’time_column’, > ‘rowtime_expression’ = ’time_column - INTERVAL 5 SECOND') */ > > > > > 2023年2月22日 10:22，kui yuan 写道： > > > > Hi devs, > > > > > > I'd like to start a discussion thread for FLIP-296[1]. This comes from an > > offline discussion with @Yun Tang, and we hope to enrich table API & SQL > to > > support many watermark-related features which were only implemented at > the > > datastream API level. > > > > > > Basically, we want to introduce watermark options in table API & SQL via > > SQL hint named 'WATERMARK_PARAMS' to support features: > > > > 1、Configurable watermark emit strategy > > > > 2、Dealing with idle sources > > > > 3、Watermark alignment > > > > > > Last but not least, thanks to Qingsheng and Jing Zhang for the initial > > reviews. > > > > > > Looking forward to your thoughts and any feedback is appreciated! > > > > > > [1] > > > https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=240884405 > > > > > > Best > > > > Yuan Kui > >
Re: Proposal to improve the Apache Flink website
Hi +1 too for this proposal Compared with the beginning, Flink has good applications in many directions now. The new website can give users better guidance for streaming/batch/olap/ML and so on. Looking forward to the new website :) Best, Shammon On Tue, Feb 21, 2023 at 2:24 PM Gyula Fóra wrote: > Hey! > > +1 > > The Flink website could indeed use some love both design and content-wise. > Thank you for starting this effort! > > I think a combination of approaches I) and II) could work well. Given that > this is a highly subjective area we could have a small kickoff / > brainstorming session with a few interested people. Then you could prepare > a small POC of the new design as a FLIP. > > Cheers > Gyula > > On Tue, 21 Feb 2023 at 03:30, Mohan, Deepthi > wrote: > > > Hi Jing, > > > > I am proposing a redesign of the Apache Flink website (NOT a simpler way > > to use Flink with Zeppelin or other technical solution). In general, the > > feedback we're received from customers is that the content on the Flink > > website is useful but is just not presented in an approachable way. For > > example, website redesign could involve existing content structured in a > > simpler way for new Flink users to understand and/or adding new content > > that tells technical decision makers how using Flink would solve their > use > > cases. > > > > I agree that we would need UX/UI designers to help with content > > restructuring, and potentially run usability studies with customers > before > > we implement a redesign. There are multiple ways we could approach this > > FLIP: > > (i) We could gather a small working committee who can discuss and set the > > direction for a website redesign. If there are folks in the community who > > have UX experience, they would be excellent candidates for this > committee. > > Interested engineers with website development experience could help > > implement these changes. > > (ii) I could take a stab at an initial redesign with the help of Amazon > UX > > designers who are interested to contribute and willing to help and bring > it > > up to the community for feedback and discussion. > > (iii) We could outsource to a third party to design and implement these > > changes (least favored) > > And I'm sure there are other alternative approaches as well. > > > > However, before we jump to solutioning, I'd love to get feedback on if > > this is something the community thinks is important to pursue. > > > > Thanks, > > Deepthi > > > > On 2/20/23, 6:07 PM, "yuxia" wrote: > > > > CAUTION: This email originated from outside of the organization. Do > > not click links or open attachments unless you can confirm the sender and > > know the content is safe. > > > > > > > > Hi, Deepthi. > > Thanks for bring it up. I'm looking forward to seeing what a “modern” > > user design will look like for Flink. > > > > Best regards, > > Yuxia > > > > - 原始邮件 - > > 发件人: "Jing Ge" > > 收件人: "dev" > > 发送时间: 星期二, 2023年 2 月 21日 上午 8:33:37 > > 主题: Re: Proposal to improve the Apache Flink website > > > > Hi Deepthi, > > > > Welcome to the Flink community. Thanks for the proposal. Commonly > > speaking, > > anything that could make Flink more attractive is welcome. I was > > wondering > > what your FLIP will look like. Are you going to propose a modern > UX/UI > > design via e.g. Zeplin which could be turned into a technical > solution > > that > > front end developers can work with? This will be typically done by > > UX/UI > > designers, afaik. You mentioned your role as a product manager. Or, > > will > > you propose a restructure of the Flink websites' content and then > find > > the UX/UI designer to work on the new design? I am a little bit > > confused. > > > > Best regards, > > Jing > > > > On Tue, Feb 21, 2023 at 1:13 AM Mohan, Deepthi > > > > wrote: > > > > > Hi, > > > > > > I am a product manager for Kinesis Data Analytics, AWS’s managed > > service > > > for Apache Flink. I’m creating this thread to gauge community > > interest in > > > redesigning the Flink website: https://flink.apache.org/. > > > A few of our customers have provided feedback that the Apache Flink > > > website looks dated. One customer claimed that the website looks > > > “unwelcoming” and lacks useful graphics and colors. Another said it > > looks > > > “like a wiki page thrown together by developers.” Customers also > > point out > > > other related Apache project websites: https://spark.apache.org/ > and > > > https://kafka.apache.org/ as having “modern” user design. Many > were > > > surprised to hear that Flink has been a top-5 Apache project for > > several > > > years now. > > > As https://flink.apache.org/ is likely the first page someone new > to > > > Flink encounters, it will be helpful to modernize the look and feel > > of the > > > website. Given the growing
Re: [ANNOUNCE] New Apache Flink Committer - Rui Fan
Congratulations, Rui! Best, Shammon On Tue, Feb 21, 2023 at 1:40 PM Sergey Nuyanzin wrote: > Congratulations, Rui! > > On Tue, Feb 21, 2023 at 4:53 AM Weihua Hu wrote: > > > Congratulations, Rui! > > > > Best, > > Weihua > > > > > > On Tue, Feb 21, 2023 at 11:28 AM Biao Geng wrote: > > > > > Congrats, Rui! > > > Best, > > > Biao Geng > > > > > > weijie guo 于2023年2月21日周二 11:21写道： > > > > > > > Congrats, Rui! > > > > > > > > Best regards, > > > > > > > > Weijie > > > > > > > > > > > > Leonard Xu 于2023年2月21日周二 11:03写道： > > > > > > > > > Congratulations, Rui! > > > > > > > > > > Best, > > > > > Leonard > > > > > > > > > > > On Feb 21, 2023, at 9:50 AM, Matt Wang wrote: > > > > > > > > > > > > Congrats Rui > > > > > > > > > > > > > > > > > > -- > > > > > > > > > > > > Best, > > > > > > Matt Wang > > > > > > > > > > > > > > > > > > Replied Message > > > > > > | From | yuxia | > > > > > > | Date | 02/21/2023 09:22 | > > > > > > | To | dev | > > > > > > | Subject | Re: [ANNOUNCE] New Apache Flink Committer - Rui Fan | > > > > > > Congrats Rui > > > > > > > > > > > > Best regards, > > > > > > Yuxia > > > > > > > > > > > > - 原始邮件 - > > > > > > 发件人: "Samrat Deb" > > > > > > 收件人: "dev" > > > > > > 发送时间: 星期二, 2023年 2 月 21日 上午 1:09:25 > > > > > > 主题: Re: [ANNOUNCE] New Apache Flink Committer - Rui Fan > > > > > > > > > > > > Congrats Rui > > > > > > > > > > > > On Mon, 20 Feb 2023 at 10:28 PM, Anton Kalashnikov < > > > kaa@yandex.com > > > > > > > > > > > wrote: > > > > > > > > > > > > Congrats Rui! > > > > > > > > > > > > -- > > > > > > Best regards, > > > > > > Anton Kalashnikov > > > > > > > > > > > > On 20.02.23 17:53, Matthias Pohl wrote: > > > > > > Congratulations, Rui :) > > > > > > > > > > > > On Mon, Feb 20, 2023 at 5:10 PM Jing Ge > > > > > > > > > wrote: > > > > > > > > > > > > Congrats Rui! > > > > > > > > > > > > On Mon, Feb 20, 2023 at 3:19 PM Piotr Nowojski < > > pnowoj...@apache.org > > > > > > > > > > wrote: > > > > > > > > > > > > Hi, everyone > > > > > > > > > > > > On behalf of the PMC, I'm very happy to announce Rui Fan as a new > > > Flink > > > > > > Committer. > > > > > > > > > > > > Rui Fan has been active on a small scale since August 2019, and > > > ramped > > > > > > up > > > > > > his contributions in the 2nd half of 2021. He was mostly involved > > in > > > > > > quite > > > > > > demanding performance related work around the network stack and > > > > > > checkpointing, like re-using TCP connections [1], and many > crucial > > > > > > improvements to the unaligned checkpoints. Among others: > FLIP-227: > > > > > > Support > > > > > > overdraft buffer [2], Merge small ChannelState file for Unaligned > > > > > > Checkpoint [3], Timeout aligned to unaligned checkpoint barrier > in > > > the > > > > > > output buffers [4]. > > > > > > > > > > > > Please join me in congratulating Rui Fan for becoming a Flink > > > > > > Committer! > > > > > > > > > > > > Best, > > > > > > Piotr Nowojski (on behalf of the Flink PMC) > > > > > > > > > > > > [1] https://issues.apache.org/jira/browse/FLINK-22643 > > > > > > [2] > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-227%3A+Support+overdraft+buffer > > > > > > [3] https://issues.apache.org/jira/browse/FLINK-26803 > > > > > > [4] https://issues.apache.org/jira/browse/FLINK-27251 > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > -- > Best regards, > Sergey >
Re: [ANNOUNCE] New Apache Flink Committer - Anton Kalashnikov
Congratulations, Anton! Best, Shammon On Tue, Feb 21, 2023 at 1:41 PM Sergey Nuyanzin wrote: > Congratulations, Anton! > > On Tue, Feb 21, 2023 at 4:53 AM Weihua Hu wrote: > > > Congratulations, Anton! > > > > Best, > > Weihua > > > > > > On Tue, Feb 21, 2023 at 11:22 AM weijie guo > > wrote: > > > > > Congratulations, Anton! > > > > > > Best regards, > > > > > > Weijie > > > > > > > > > Leonard Xu 于2023年2月21日周二 11:02写道： > > > > > > > Congratulations, Anton! > > > > > > > > Best, > > > > Leonard > > > > > > > > > > > > > On Feb 21, 2023, at 10:02 AM, Rui Fan wrote: > > > > > > > > > > Congratulations, Anton! > > > > > > > > > > Best, > > > > > Rui Fan > > > > > > > > > > On Tue, Feb 21, 2023 at 9:23 AM yuxia > > > > > wrote: > > > > > > > > > >> Congrats Anton! > > > > >> > > > > >> Best regards, > > > > >> Yuxia > > > > >> > > > > >> - 原始邮件 - > > > > >> 发件人: "Matthias Pohl" > > > > >> 收件人: "dev" > > > > >> 发送时间: 星期二, 2023年 2 月 21日 上午 12:52:40 > > > > >> 主题: Re: [ANNOUNCE] New Apache Flink Committer - Anton Kalashnikov > > > > >> > > > > >> Congratulations, Anton! :-) > > > > >> > > > > >> On Mon, Feb 20, 2023 at 5:09 PM Jing Ge > > > > > > > >> wrote: > > > > >> > > > > >>> Congrats Anton! > > > > >>> > > > > >>> On Mon, Feb 20, 2023 at 5:02 PM Samrat Deb < > decordea...@gmail.com> > > > > >> wrote: > > > > >>> > > > > congratulations Anton! > > > > > > > > Bests, > > > > Samrat > > > > > > > > On Mon, 20 Feb 2023 at 9:29 PM, John Roesler < > vvcep...@apache.org > > > > > > > >>> wrote: > > > > > > > > > Congratulations, Anton! > > > > > -John > > > > > > > > > > On Mon, Feb 20, 2023, at 08:18, Piotr Nowojski wrote: > > > > >> Hi, everyone > > > > >> > > > > >> On behalf of the PMC, I'm very happy to announce Anton > > Kalashnikov > > > > >>> as a > > > > > new > > > > >> Flink Committer. > > > > >> > > > > >> Anton has been very active for almost two years already, > > authored > > > > >> and > > > > >> reviewed many PRs over this time. He is active in the Flink's > > > > >>> runtime, > > > > >> being the main author of improvements like Buffer Debloating > > > > >>> (FLIP-183) > > > > >> [1], solved many bugs and fixed many test instabilities, > > generally > > > > > speaking > > > > >> helping with the maintenance of runtime components. > > > > >> > > > > >> Please join me in congratulating Anton Kalashnikov for > becoming > > a > > > > >>> Flink > > > > >> Committer! > > > > >> > > > > >> Best, > > > > >> Piotr Nowojski (on behalf of the Flink PMC) > > > > >> > > > > >> [1] > > > > >> > > > > > > > > > > > > > >>> > > > > >> > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-183%3A+Dynamic+buffer+size+adjustment > > > > > > > > > > > > > >>> > > > > >> > > > > > > > > > > > > > > > > -- > Best regards, > Sergey >
Re: [DISCUSS] Extract core autoscaling algorithm as new SubModule in flink-kubernetes-operator
Hi Samrat My team is also looking at this piece. After you give your proposal, we also hope to join it with you if possible. I hope we can improve this together for use in our production too, thanks :) Best, Shammon On Fri, Feb 17, 2023 at 9:27 PM Samrat Deb wrote: > @Gyula > Thank you > We will work on this and try to come up with an approach. > > > > > On Fri, Feb 17, 2023 at 6:12 PM Gyula Fóra wrote: > > > In case you guys feel strongly about this I suggest you try to fork the > > autoscaler implementation and make a version that works with both the > > Kubernetes operator and YARN. > > If your solution is generic and works well, we can discuss the way > forward. > > > > Unfortunately me or my team don't really have the resources to assist you > > with the YARN effort as we are mostly invested in Kubernetes but of > course > > we are happy to review your work. > > > > Gyula > > > > > > On Fri, Feb 17, 2023 at 1:09 PM Prabhu Joseph < > prabhujose.ga...@gmail.com> > > wrote: > > > > > @Gyula > > > > > > >> It is easier to make the operator work with jobs running in > different > > > types of clusters than to take the > > > autoscaler module itself and plug that in somewhere else. > > > > > > Our (part of Samrat's team) main problem is to leverage the AutoScaler > > > Recommendation Engine part of Flink-Kubernetes-Operator for our Flink > > jobs > > > running on YARN. > > > Currently, it is not feasible as the autoscaler module is tightly > coupled > > > with the operator. We agree that the operator serves the two core > > > requirements, but the operator itself > > > cannot be used for Flink jobs running on YARN. Those core requirements > > are > > > solved through other mechanisms in the case of YARN. But the main > problem > > > for us is *how to* > > > *use the AutoScaler Recommendation Engine for Flink Jobs on YARN.* > > > > > > > > > > > > > > > > > > > > > > > > > > > On Fri, Feb 17, 2023 at 6:34 AM Shammon FY wrote: > > > > > > > Hi Gyula, Samrat > > > > > > > > Thanks for your input and I totally agree with you that it's really > big > > > > work. As @Samrat mentioned above, I think it's not a short way to > make > > > the > > > > autoscaler completely independent too. But I still find some valuable > > > > points for the `completely independent autoscaler`, and I think this > > may > > > be > > > > the goal we need to achieve in the future. > > > > > > > > 1. A large k8s cluster may manage thousands of machines, and users > may > > > run > > > > tens of thousands flink jobs in one k8s cluster. If the autoscaler > > > manages > > > > all these jobs, the autoscaler should be horizontal expansion. > > > > > > > > 2. As you mentioned, "execute the job stateful upgrades safely" is > > > indeed a > > > > complexity work, but I think we should decouple it from k8s operator > > > > > > > > a) In addition to k8s, there may be some other resource management > > > > > > > > b) Flink may support more scaler operations by REST API, such as > > FLIP-291 > > > > [1] > > > > > > > > c) In our production environment, there's a 'Job Submission Gateway' > > > which > > > > stores job info and config, monitors the status of running jobs. > After > > > the > > > > autoscaler upgrades the job, it must update the config in Gateway and > > > users > > > > can restart his job with the updated config to avoid resource > conflict. > > > > Under these circumstances, the autoscaler sending upgrade requests to > > the > > > > gateway may be a good choice. > > > > > > > > > > > > [1] > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-291%3A+Externalized+Declarative+Resource+Management > > > > > > > > > > > > Best, > > > > Shammon > > > > > > > > > > > > On Thu, Feb 16, 2023 at 11:03 PM Gyula Fóra > > > wrote: > > > > > > > > > @Shammon , Samrat: > > > > > > > > > > I appreciate the enthusiasm and I wish this was only a matter of > > > > int
Re: [DISCUSS] Extract core autoscaling algorithm as new SubModule in flink-kubernetes-operator
Hi Gyula, Samrat Thanks for your input and I totally agree with you that it's really big work. As @Samrat mentioned above, I think it's not a short way to make the autoscaler completely independent too. But I still find some valuable points for the `completely independent autoscaler`, and I think this may be the goal we need to achieve in the future. 1. A large k8s cluster may manage thousands of machines, and users may run tens of thousands flink jobs in one k8s cluster. If the autoscaler manages all these jobs, the autoscaler should be horizontal expansion. 2. As you mentioned, "execute the job stateful upgrades safely" is indeed a complexity work, but I think we should decouple it from k8s operator a) In addition to k8s, there may be some other resource management b) Flink may support more scaler operations by REST API, such as FLIP-291 [1] c) In our production environment, there's a 'Job Submission Gateway' which stores job info and config, monitors the status of running jobs. After the autoscaler upgrades the job, it must update the config in Gateway and users can restart his job with the updated config to avoid resource conflict. Under these circumstances, the autoscaler sending upgrade requests to the gateway may be a good choice. [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-291%3A+Externalized+Declarative+Resource+Management Best, Shammon On Thu, Feb 16, 2023 at 11:03 PM Gyula Fóra wrote: > @Shammon , Samrat: > > I appreciate the enthusiasm and I wish this was only a matter of intention > but making the autoscaler work without the operator may be a pretty big > task. > You must not forget 2 core requirements here. > > 1. The autoscaler logic itself has to run somewhere (in this case on k8s > within the operator)S > 2. Something has to execute the job stateful upgrades safely based on the > scaling decisions (in this case the operator does that). > > 1. Can be solved almost anywhere easily however you need resiliency etc for > this to be a prod application, 2. is the really tricky part. The operator > was actually built to execute job upgrades, if you look at the code you > will appreciate the complexity of the task. > > As I said in the earlier thread. It is easier to make the operator work > with jobs running in different types of clusters than to take the > autoscaler module itself and plug that in somewhere else. > > Gyula > > > On Thu, Feb 16, 2023 at 3:12 PM Samrat Deb wrote: > > > Hi Shammon, > > > > Thank you for your input, completely aligned with you. > > > > We are fine with either of the options , > > > > but IMO, to start with it will be easy to have it in the > > flink-kubernetes-operator as a module instead of a separate repo which > > requires additional effort. > > > > Given that we would be incrementally working on making an autoscaling > > recommendation framework generic enough, > > > > Once it reaches a point where the community feels it needs to be moved > to a > > separate repo we can take a call. > > > > Bests, > > > > Samrat > > > > > > On Thu, Feb 16, 2023 at 7:37 PM Samrat Deb > wrote: > > > > > Hi Max , > > > If you are fine and aligned with the same thought , since this is going > > to > > > be very useful to us, we are ready to help / contribute additional work > > > required. > > > > > > Bests, > > > Samrat > > > > > > > > > On Thu, 16 Feb 2023 at 5:28 PM, Shammon FY wrote: > > > > > >> Hi Samrat > > >> > > >> Do you mean to create an independent module for flink scaling in > > >> flink-k8s-operator? How about creating a project such as > > >> `flink-auto-scaling` which is completely independent? Besides resource > > >> managers such as k8s and yarn, we can do more things in the project, > for > > >> example, updating config in the user's `job submission system` after > > >> scaling flink jobs. WDYT? > > >> > > >> Best, > > >> Shammon > > >> > > >> > > >> On Thu, Feb 16, 2023 at 7:38 PM Maximilian Michels > > >> wrote: > > >> > > >> > Hi Samrat, > > >> > > > >> > The autoscaling module is now pluggable but it is still tightly > > >> > coupled with Kubernetes. It will take additional work for the logic > to > > >> > work independently of the cluster manager. > > >> > > > >> > -Max > > >> > > > >> > On Thu, Feb 16, 2023 at 11:14 AM Samrat Deb > > >> wrote: > > &g
