Apache Flink 中文用户邮件列表

关于1.11Flink SQL 全新API设计的一些问题

Classic

List

Threaded

12 messages Options

刘首维

关于1.11Flink SQL 全新API设计的一些问题

Hi all,

很高兴看到Flink 1.11的发布，FLIP95和FLIP105也成功落地~

我们最近在调研基于1.11的SQL/Table API对我们旧有的SQL SDK进行升级。经过初步调研发现，基于`Factory`和`DynamicTable`的API，CatalogTable会被Planner直接变成Transformation，而不是跟之前一样可以拿到DataStream。比如之前的`StreamTableSource`#getDataStream，我可以直接获得DataStream对象的。这个动作/方法对于我们很重要，因为我们封装的SDK中，很多内部操作是在这个方法调用时触发/完成的。

所以，如果基于新的API去开发的话，我该如何获取到DataStream，或者达成类似的效果呢（尽可能不去动执行计划的话）

Jingsong Li

Re: 关于1.11Flink SQL 全新API设计的一些问题

可以分享下你们为啥要拿到DataStream吗？什么场景一定离不开DataStream吗？

Best
Jingsong

On Wed, Jul 22, 2020 at 12:36 PM 刘首维 <[hidden email]> wrote:

> Hi all,
>
>
>
> 很高兴看到Flink 1.11的发布，FLIP95和FLIP105也成功落地~
>
> 我们最近在调研基于1.11的SQL/Table API对我们旧有的SQL
> SDK进行升级。经过初步调研发现，基于`Factory`和`DynamicTable`的API，CatalogTable会被Planner直接变成Transformation，而不是跟之前一样可以拿到DataStream。比如之前的`StreamTableSource`#getDataStream，我可以直接获得DataStream对象的。这个动作/方法对于我们很重要，因为我们封装的SDK中，很多内部操作是在这个方法调用时触发/完成的。
>
>
>
> 所以，如果基于新的API去开发的话，我该如何获取到DataStream，或者达成类似的效果呢（尽可能不去动执行计划的话）
>

--
Best, Jingsong Lee

刘首维

答复: 关于1.11Flink SQL 全新API设计的一些问题

Hi JingSong,

简单介绍一下背景，我们组的一个工作就是用户在页面写Sql,我们负责将Sql处理转换成Flink作业并在我们的平台上运行，这个转换的过程依赖我们的SQL SDK
下面我举几个我们比较常用且感觉用1.11新API不太好实现的例子

1. 我们现在有某种特定的Kafka数据格式，1条Kafka数据会对应转换n（n为正整数）条Row数据，我们的做法是在emitDataStream的时候增加了一个process/FlatMap阶段，用于处理这种情况，这样对用户是透明的。
2. 我们目前封装了一些自己的Sink，我们会在Sink之前增加一个process/Filter 用来做缓冲池/微批/数据过滤等功能
3. 调整或者指定Source/Sink并行度为用户指定值，我们也是在DataStream层面上去做的
4. 对于一些特殊Source Sink，他们会和KeyBy操作组合（对用户透明），我们也是在DataStream层面上去做的

如果可以的话，能让我在API层面拿到Transformation也是能满足我需求的

________________________________
发件人: Jingsong Li <[hidden email]>
发送时间: 2020年7月22日 13:26:00
收件人: user-zh
抄送: [hidden email]
主题: Re: 关于1.11Flink SQL 全新API设计的一些问题

可以分享下你们为啥要拿到DataStream吗？什么场景一定离不开DataStream吗？

Best
Jingsong

On Wed, Jul 22, 2020 at 12:36 PM 刘首维 <[hidden email]> wrote:

--
Best, Jingsong Lee

Jingsong Li

Re: 关于1.11Flink SQL 全新API设计的一些问题

Hi 首维，

非常感谢你的信息，我们试图去掉太灵活的“DataStream”，但是也逐渐发现一些额外的需求，再考虑和评估下你的需求。

CC: @Jark Wu <[hidden email]>

Best,
Jingsong

On Wed, Jul 22, 2020 at 1:49 PM 刘首维 <[hidden email]> wrote:

> Hi JingSong,
>
>
> 简单介绍一下背景，我们组的一个工作就是用户在页面写Sql,我们负责将Sql处理转换成Flink作业并在我们的平台上运行，这个转换的过程依赖我们的SQL
> SDK
> 下面我举几个我们比较常用且感觉用1.11新API不太好实现的例子
>
>
> 1. 我们现在有某种特定的Kafka数据格式，1条Kafka数据
> 会对应转换n（n为正整数）条Row数据，我们的做法是在emitDataStream的时候增加了一个process/FlatMap阶段，用于处理这种情况，这样对用户是透明的。
> 2. 我们目前封装了一些自己的Sink，我们会在Sink之前增加一个process/Filter 用来做缓冲池/微批/数据过滤等功能
> 3. 调整或者指定Source/Sink并行度为用户指定值，我们也是在DataStream层面上去做的
> 4. 对于一些特殊Source Sink，他们会和KeyBy操作组合（对用户透明），我们也是在DataStream层面上去做的
>
>
> 如果可以的话，能让我在API层面拿到Transformation也是能满足我需求的
>
>
> ________________________________
> 发件人: Jingsong Li <[hidden email]>
> 发送时间: 2020年7月22日 13:26:00
> 收件人: user-zh
> 抄送: [hidden email]
> 主题: Re: 关于1.11Flink SQL 全新API设计的一些问题
>
> 可以分享下你们为啥要拿到DataStream吗？什么场景一定离不开DataStream吗？
>
> Best
> Jingsong
>
> On Wed, Jul 22, 2020 at 12:36 PM 刘首维 <[hidden email]> wrote:
>
> > Hi all,
> >
> >
> >
> > 很高兴看到Flink 1.11的发布，FLIP95和FLIP105也成功落地~
> >
> > 我们最近在调研基于1.11的SQL/Table API对我们旧有的SQL
> >
> SDK进行升级。经过初步调研发现，基于`Factory`和`DynamicTable`的API，CatalogTable会被Planner直接变成Transformation，而不是跟之前一样可以拿到DataStream。比如之前的`StreamTableSource`#getDataStream，我可以直接获得DataStream对象的。这个动作/方法对于我们很重要，因为我们封装的SDK中，很多内部操作是在这个方法调用时触发/完成的。
> >
> >
> >
> > 所以，如果基于新的API去开发的话，我该如何获取到DataStream，或者达成类似的效果呢（尽可能不去动执行计划的话）
> >
>
>
> --
> Best, Jingsong Lee
>

--
Best, Jingsong Lee

刘首维

答复: 关于1.11Flink SQL 全新API设计的一些问题

Hi JingSong，

感谢回复，真心期待一个理想的解决方案~

________________________________
发件人: Jingsong Li <[hidden email]>
发送时间: 2020年7月22日 13:58:51
收件人: user-zh; Jark Wu
主题: Re: 关于1.11Flink SQL 全新API设计的一些问题

Hi 首维，

非常感谢你的信息，我们试图去掉太灵活的“DataStream”，但是也逐渐发现一些额外的需求，再考虑和评估下你的需求。

CC: @Jark Wu <[hidden email]>

Best,
Jingsong

On Wed, Jul 22, 2020 at 1:49 PM 刘首维 <[hidden email]> wrote:

--
Best, Jingsong Lee

Michael Ran

Re:答复: 关于1.11Flink SQL 全新API设计的一些问题

In reply to this post by 刘首维

这个需求我们也比较类似：<br/>要获取注册的表信息，自己用stream+table 实现部分逻辑
在 2020-07-22 13:47:25，"刘首维" <[hidden email]> 写道：

>Hi JingSong,
>
> 简单介绍一下背景，我们组的一个工作就是用户在页面写Sql,我们负责将Sql处理转换成Flink作业并在我们的平台上运行，这个转换的过程依赖我们的SQL SDK
> 下面我举几个我们比较常用且感觉用1.11新API不太好实现的例子
>
>
> 1. 我们现在有某种特定的Kafka数据格式，1条Kafka数据会对应转换n（n为正整数）条Row数据，我们的做法是在emitDataStream的时候增加了一个process/FlatMap阶段，用于处理这种情况，这样对用户是透明的。
> 2. 我们目前封装了一些自己的Sink，我们会在Sink之前增加一个process/Filter 用来做缓冲池/微批/数据过滤等功能
> 3. 调整或者指定Source/Sink并行度为用户指定值，我们也是在DataStream层面上去做的
> 4. 对于一些特殊Source Sink，他们会和KeyBy操作组合（对用户透明），我们也是在DataStream层面上去做的
>
>
>如果可以的话，能让我在API层面拿到Transformation也是能满足我需求的
>
>
>________________________________
>发件人: Jingsong Li <[hidden email]>
>发送时间: 2020年7月22日 13:26:00
>收件人: user-zh
>抄送: [hidden email]
>主题: Re: 关于1.11Flink SQL 全新API设计的一些问题
>
>可以分享下你们为啥要拿到DataStream吗？什么场景一定离不开DataStream吗？
>
>Best
>Jingsong
>
>On Wed, Jul 22, 2020 at 12:36 PM 刘首维 <[hidden email]> wrote:
>
>> Hi all,
>>
>>
>>
>> 很高兴看到Flink 1.11的发布，FLIP95和FLIP105也成功落地~
>>
>> 我们最近在调研基于1.11的SQL/Table API对我们旧有的SQL
>> SDK进行升级。经过初步调研发现，基于`Factory`和`DynamicTable`的API，CatalogTable会被Planner直接变成Transformation，而不是跟之前一样可以拿到DataStream。比如之前的`StreamTableSource`#getDataStream，我可以直接获得DataStream对象的。这个动作/方法对于我们很重要，因为我们封装的SDK中，很多内部操作是在这个方法调用时触发/完成的。
>>
>>
>>
>> 所以，如果基于新的API去开发的话，我该如何获取到DataStream，或者达成类似的效果呢（尽可能不去动执行计划的话）
>>
>
>
>--
>Best, Jingsong Lee

Leonard Xu

Re: 关于1.11Flink SQL 全新API设计的一些问题

In reply to this post by 刘首维

Hi，首维， Ran

感谢分享，我理解1.11新的API主要是想把 Table API 和 DataStream API 两套尽量拆分干净, 但看起来平台级的开发工作会依赖DataStream的一些预处理和用户逻辑。
我觉得这类需求对平台开发是合理，可以收集反馈下的， cc: godfrey

祝好
Leonard Xu

> 在 2020年7月22日，13:47，刘首维 <[hidden email]> 写道：
>
> Hi JingSong,
>
> 简单介绍一下背景，我们组的一个工作就是用户在页面写Sql,我们负责将Sql处理转换成Flink作业并在我们的平台上运行，这个转换的过程依赖我们的SQL SDK
> 下面我举几个我们比较常用且感觉用1.11新API不太好实现的例子
>
>
> 1. 我们现在有某种特定的Kafka数据格式，1条Kafka数据会对应转换n（n为正整数）条Row数据，我们的做法是在emitDataStream的时候增加了一个process/FlatMap阶段，用于处理这种情况，这样对用户是透明的。
> 2. 我们目前封装了一些自己的Sink，我们会在Sink之前增加一个process/Filter 用来做缓冲池/微批/数据过滤等功能
> 3. 调整或者指定Source/Sink并行度为用户指定值，我们也是在DataStream层面上去做的
> 4. 对于一些特殊Source Sink，他们会和KeyBy操作组合（对用户透明），我们也是在DataStream层面上去做的
>
>
> 如果可以的话，能让我在API层面拿到Transformation也是能满足我需求的
>
>
> ________________________________
> 发件人: Jingsong Li <[hidden email]>
> 发送时间: 2020年7月22日 13:26:00
> 收件人: user-zh
> 抄送: [hidden email]
> 主题: Re: 关于1.11Flink SQL 全新API设计的一些问题
>
> 可以分享下你们为啥要拿到DataStream吗？什么场景一定离不开DataStream吗？
>
> Best
> Jingsong
>
> On Wed, Jul 22, 2020 at 12:36 PM 刘首维 <[hidden email]> wrote:
>
>> Hi all,
>>
>>
>>
>> 很高兴看到Flink 1.11的发布，FLIP95和FLIP105也成功落地~
>>
>> 我们最近在调研基于1.11的SQL/Table API对我们旧有的SQL
>> SDK进行升级。经过初步调研发现，基于`Factory`和`DynamicTable`的API，CatalogTable会被Planner直接变成Transformation，而不是跟之前一样可以拿到DataStream。比如之前的`StreamTableSource`#getDataStream，我可以直接获得DataStream对象的。这个动作/方法对于我们很重要，因为我们封装的SDK中，很多内部操作是在这个方法调用时触发/完成的。
>>
>>
>>
>> 所以，如果基于新的API去开发的话，我该如何获取到DataStream，或者达成类似的效果呢（尽可能不去动执行计划的话）
>>
>
>
> --
> Best, Jingsong Lee

Jark

Re: 关于1.11Flink SQL 全新API设计的一些问题

Administrator

Hi，首维，

非常感谢反馈。与 DataStream 解耦是 FLIP-95 的一个非常重要的设计目标，这让 sink/source 对于框架来说不再是黑盒，
因此将来才可以做诸如 state 兼容升级、消息顺序保障、自动并发设置等等事情。

关于你的一些需求，下面是我的建议和回复：

> 1. 我们现在有某种特定的Kafka数据格式，1条Kafka数据
会对应转换n（n为正整数）条Row数据，我们的做法是在emitDataStream的时候增加了一个process/FlatMap阶段，用于处理这种情况，这样对用户是透明的。
这个理论上还属于“数据格式”的职责，所以建议做在 DeserializationSchema 上，目前 DeserializationSchema
支持一对多的输出。可以参考 DebeziumJsonDeserializationSchema 的实现。

> 2. 我们目前封装了一些自己的Sink，我们会在Sink之前增加一个process/Filter 用来做缓冲池/微批/数据过滤等功能
这个我觉得也可以封装在 SinkFunction 里面。

> 3. 调整或者指定Source/Sink并行度为用户指定值，我们也是在DataStream层面上去做的
这个社区也有计划在 FLIP-95 之上支持，会提供并发（或者分区）推断的能力。

> 4. 对于一些特殊Source Sink，他们会和KeyBy操作组合（对用户透明），我们也是在DataStream层面上去做的
这个能在具体一点吗？目前像 SupportsPartitioning 接口，就可以指定数据在交给 sink 之前先做 group by
partition。我感觉这个可能也可以通过引入类似的接口解决。

Best,
Jark

On Wed, 22 Jul 2020 at 16:27, Leonard Xu <[hidden email]> wrote:

> Hi，首维， Ran
>
> 感谢分享，我理解1.11新的API主要是想把 Table API 和 DataStream API 两套尽量拆分干净,
> 但看起来平台级的开发工作会依赖DataStream的一些预处理和用户逻辑。
> 我觉得这类需求对平台开发是合理，可以收集反馈下的， cc: godfrey
>
> 祝好
> Leonard Xu
>
>
> > 在 2020年7月22日，13:47，刘首维 <[hidden email]> 写道：
> >
> > Hi JingSong,
> >
> >
> 简单介绍一下背景，我们组的一个工作就是用户在页面写Sql,我们负责将Sql处理转换成Flink作业并在我们的平台上运行，这个转换的过程依赖我们的SQL
> SDK
> > 下面我举几个我们比较常用且感觉用1.11新API不太好实现的例子
> >
> >
> > 1. 我们现在有某种特定的Kafka数据格式，1条Kafka数据
> 会对应转换n（n为正整数）条Row数据，我们的做法是在emitDataStream的时候增加了一个process/FlatMap阶段，用于处理这种情况，这样对用户是透明的。
> > 2. 我们目前封装了一些自己的Sink，我们会在Sink之前增加一个process/Filter 用来做缓冲池/微批/数据过滤等功能
> > 3. 调整或者指定Source/Sink并行度为用户指定值，我们也是在DataStream层面上去做的
> > 4. 对于一些特殊Source Sink，他们会和KeyBy操作组合（对用户透明），我们也是在DataStream层面上去做的
> >
> >
> > 如果可以的话，能让我在API层面拿到Transformation也是能满足我需求的
> >
> >
> > ________________________________
> > 发件人: Jingsong Li <[hidden email]>
> > 发送时间: 2020年7月22日 13:26:00
> > 收件人: user-zh
> > 抄送: [hidden email]
> > 主题: Re: 关于1.11Flink SQL 全新API设计的一些问题
> >
> > 可以分享下你们为啥要拿到DataStream吗？什么场景一定离不开DataStream吗？
> >
> > Best
> > Jingsong
> >
> > On Wed, Jul 22, 2020 at 12:36 PM 刘首维 <[hidden email]> wrote:
> >
> >> Hi all,
> >>
> >>
> >>
> >> 很高兴看到Flink 1.11的发布，FLIP95和FLIP105也成功落地~
> >>
> >> 我们最近在调研基于1.11的SQL/Table API对我们旧有的SQL
> >>
> SDK进行升级。经过初步调研发现，基于`Factory`和`DynamicTable`的API，CatalogTable会被Planner直接变成Transformation，而不是跟之前一样可以拿到DataStream。比如之前的`StreamTableSource`#getDataStream，我可以直接获得DataStream对象的。这个动作/方法对于我们很重要，因为我们封装的SDK中，很多内部操作是在这个方法调用时触发/完成的。
> >>
> >>
> >>
> >> 所以，如果基于新的API去开发的话，我该如何获取到DataStream，或者达成类似的效果呢（尽可能不去动执行计划的话）
> >>
> >
> >
> > --
> > Best, Jingsong Lee
>
>

刘首维

答复: 关于1.11Flink SQL 全新API设计的一些问题

Hi, Jark

感谢你的建议！

我们这边充分相信社区在SQL/Table上的苦心孤诣，也愿意跟进Flink在新版本的变化。

先看我遇到问题本身，你的建议确实可以帮助我解决问题。我想聊一下我对问题之外的一些想法

```

> 2. 我们目前封装了一些自己的Sink，我们会在Sink之前增加一个process/Filter 用来做缓冲池/微批/数据过滤等功能
这个我觉得也可以封装在 SinkFunction 里面。

```

比如上述这个问题2，我们确实可以把它做到SinkFunction中，但是我个人认为这可能在设计上不够理想的。我个人在设计编排Function/算子的时候习惯于遵循”算子单一职责”的原则，这也是我为什么会拆分出多个process/filter算子编排到SinkFunction前面而非将这些功能耦合到SinkFunction去做。另一方面，没了DataStream，向新的API的迁移成本相对来说变得更高了一些~ 又或者，我们现在还有一些特殊原因，算子编排的时候会去修改TaskChain Strategy，这个时候DataStream的灵活性是必不可少的

考虑到Flink Task都可以拆分成Source -> Transformation -> sink 三个阶段，那么能让用户可以对自己的作业针对（流或批）的运行模式下，可以有效灵活做一些自己的定制策略/优化/逻辑可能是会方便的~

诚然，DataStream的灵活性确实会是一把双刃剑，但就像@leonard提到的，平台层和应用层的目的和开发重点可能也不太一样，对Flink API使用侧重点也不同。我个人还是希望可以在享受全新API设计优势同时，

可以继续使用DataStream（Transformation）的灵活性，助力Flink组件在我们组的开落地

再次感谢各位的回复！

________________________________
发件人: Jark Wu <[hidden email]>
发送时间: 2020年7月22日 16:33:45
收件人: user-zh
抄送: godfrey he; [hidden email]; 刘首维
主题: Re: 关于1.11Flink SQL 全新API设计的一些问题

Hi，首维，

非常感谢反馈。与 DataStream 解耦是 FLIP-95 的一个非常重要的设计目标，这让 sink/source 对于框架来说不再是黑盒，
因此将来才可以做诸如 state 兼容升级、消息顺序保障、自动并发设置等等事情。

关于你的一些需求，下面是我的建议和回复：

> 1. 我们现在有某种特定的Kafka数据格式，1条Kafka数据会对应转换n（n为正整数）条Row数据，我们的做法是在emitDataStream的时候增加了一个process/FlatMap阶段，用于处理这种情况，这样对用户是透明的。
这个理论上还属于“数据格式”的职责，所以建议做在 DeserializationSchema 上，目前 DeserializationSchema 支持一对多的输出。可以参考 DebeziumJsonDeserializationSchema 的实现。

> 2. 我们目前封装了一些自己的Sink，我们会在Sink之前增加一个process/Filter 用来做缓冲池/微批/数据过滤等功能
这个我觉得也可以封装在 SinkFunction 里面。

> 3. 调整或者指定Source/Sink并行度为用户指定值，我们也是在DataStream层面上去做的
这个社区也有计划在 FLIP-95 之上支持，会提供并发（或者分区）推断的能力。

> 4. 对于一些特殊Source Sink，他们会和KeyBy操作组合（对用户透明），我们也是在DataStream层面上去做的
这个能在具体一点吗？目前像 SupportsPartitioning 接口，就可以指定数据在交给 sink 之前先做 group by partition。我感觉这个可能也可以通过引入类似的接口解决。

Best,
Jark

On Wed, 22 Jul 2020 at 16:27, Leonard Xu <[hidden email]<mailto:[hidden email]>> wrote:
Hi，首维， Ran

感谢分享，我理解1.11新的API主要是想把 Table API 和 DataStream API 两套尽量拆分干净, 但看起来平台级的开发工作会依赖DataStream的一些预处理和用户逻辑。
我觉得这类需求对平台开发是合理，可以收集反馈下的， cc: godfrey

祝好
Leonard Xu

> 在 2020年7月22日，13:47，刘首维 <[hidden email]<mailto:[hidden email]>> 写道：
>
> Hi JingSong,
>
> 简单介绍一下背景，我们组的一个工作就是用户在页面写Sql,我们负责将Sql处理转换成Flink作业并在我们的平台上运行，这个转换的过程依赖我们的SQL SDK
> 下面我举几个我们比较常用且感觉用1.11新API不太好实现的例子
>
>
> 1. 我们现在有某种特定的Kafka数据格式，1条Kafka数据会对应转换n（n为正整数）条Row数据，我们的做法是在emitDataStream的时候增加了一个process/FlatMap阶段，用于处理这种情况，这样对用户是透明的。
> 2. 我们目前封装了一些自己的Sink，我们会在Sink之前增加一个process/Filter 用来做缓冲池/微批/数据过滤等功能
> 3. 调整或者指定Source/Sink并行度为用户指定值，我们也是在DataStream层面上去做的
> 4. 对于一些特殊Source Sink，他们会和KeyBy操作组合（对用户透明），我们也是在DataStream层面上去做的
>
>
> 如果可以的话，能让我在API层面拿到Transformation也是能满足我需求的
>
>
> ________________________________
> 发件人: Jingsong Li <[hidden email]<mailto:[hidden email]>>
> 发送时间: 2020年7月22日 13:26:00
> 收件人: user-zh
> 抄送: [hidden email]<mailto:[hidden email]>
> 主题: Re: 关于1.11Flink SQL 全新API设计的一些问题
>
> 可以分享下你们为啥要拿到DataStream吗？什么场景一定离不开DataStream吗？
>
> Best
> Jingsong
>
> On Wed, Jul 22, 2020 at 12:36 PM 刘首维 <[hidden email]<mailto:[hidden email]>> wrote:
>
>> Hi all,
>>
>>
>>
>> 很高兴看到Flink 1.11的发布，FLIP95和FLIP105也成功落地~
>>
>> 我们最近在调研基于1.11的SQL/Table API对我们旧有的SQL
>> SDK进行升级。经过初步调研发现，基于`Factory`和`DynamicTable`的API，CatalogTable会被Planner直接变成Transformation，而不是跟之前一样可以拿到DataStream。比如之前的`StreamTableSource`#getDataStream，我可以直接获得DataStream对象的。这个动作/方法对于我们很重要，因为我们封装的SDK中，很多内部操作是在这个方法调用时触发/完成的。
>>
>>
>>
>> 所以，如果基于新的API去开发的话，我该如何获取到DataStream，或者达成类似的效果呢（尽可能不去动执行计划的话）
>>
>
>
> --
> Best, Jingsong Lee

godfrey he

Re: 关于1.11Flink SQL 全新API设计的一些问题

Hi，首维

感谢给出非常详细的反馈。这个问题我们之前内部也有一些讨论，但由于缺乏一些真实场景，最后维持了当前的接口。
我们会根据你提供的场景进行后续讨论。

Best,
Godfrey

刘首维 <[hidden email]> 于2020年7月22日周三下午5:23写道：

> Hi, Jark
>
>
>
> 感谢你的建议！
>
> 我们这边充分相信社区在SQL/Table上的苦心孤诣，也愿意跟进Flink在新版本的变化。
>
> 先看我遇到问题本身，你的建议确实可以帮助我解决问题。我想聊一下我对问题之外的一些想法
>
> ```
>
> > 2. 我们目前封装了一些自己的Sink，我们会在Sink之前增加一个process/Filter
> 用来做缓冲池/微批/数据过滤等功能
> 这个我觉得也可以封装在 SinkFunction 里面。
>
> ```
>
> 比如上述这个问题2，我们确实可以把它做到SinkFunction中，但是我个人认为这可能在设计上不够理想的。我个人在设计编排Function/算子的时候习惯于遵循”算子单一职责”的原则，这也是我为什么会拆分出多个process/filter算子编排到SinkFunction前面而非将这些功能耦合到SinkFunction去做。另一方面，没了DataStream，向新的API的迁移成本相对来说变得更高了一些~
> 又或者，我们现在还有一些特殊原因，算子编排的时候会去修改TaskChain Strategy，这个时候DataStream的灵活性是必不可少的
>
> 考虑到Flink Task都可以拆分成Source -> Transformation -> sink
> 三个阶段，那么能让用户可以对自己的作业针对（流或批）的运行模式下，可以有效灵活做一些自己的定制策略/优化/逻辑可能是会方便的~
>
> 诚然，DataStream的灵活性确实会是一把双刃剑，但就像@leonard提到的，平台层和应用层的目的和开发重点可能也不太一样，对Flink
> API使用侧重点也不同。我个人还是希望可以在享受全新API设计优势同时，
>
> 可以继续使用DataStream（Transformation）的灵活性，助力Flink组件在我们组的开落地
>
>
> 再次感谢各位的回复！
>
> ________________________________
> 发件人: Jark Wu <[hidden email]>
> 发送时间: 2020年7月22日 16:33:45
> 收件人: user-zh
> 抄送: godfrey he; [hidden email]; 刘首维
> 主题: Re: 关于1.11Flink SQL 全新API设计的一些问题
>
> Hi，首维，
>
> 非常感谢反馈。与 DataStream 解耦是 FLIP-95 的一个非常重要的设计目标，这让 sink/source 对于框架来说不再是黑盒，
> 因此将来才可以做诸如 state 兼容升级、消息顺序保障、自动并发设置等等事情。
>
> 关于你的一些需求，下面是我的建议和回复：
>
> > 1. 我们现在有某种特定的Kafka数据格式，1条Kafka数据
> 会对应转换n（n为正整数）条Row数据，我们的做法是在emitDataStream的时候增加了一个process/FlatMap阶段，用于处理这种情况，这样对用户是透明的。
> 这个理论上还属于“数据格式”的职责，所以建议做在 DeserializationSchema 上，目前 DeserializationSchema
> 支持一对多的输出。可以参考 DebeziumJsonDeserializationSchema 的实现。
>
> > 2. 我们目前封装了一些自己的Sink，我们会在Sink之前增加一个process/Filter 用来做缓冲池/微批/数据过滤等功能
> 这个我觉得也可以封装在 SinkFunction 里面。
>
> > 3. 调整或者指定Source/Sink并行度为用户指定值，我们也是在DataStream层面上去做的
> 这个社区也有计划在 FLIP-95 之上支持，会提供并发（或者分区）推断的能力。
>
> > 4. 对于一些特殊Source Sink，他们会和KeyBy操作组合（对用户透明），我们也是在DataStream层面上去做的
> 这个能在具体一点吗？目前像 SupportsPartitioning 接口，就可以指定数据在交给 sink 之前先做 group by
> partition。我感觉这个可能也可以通过引入类似的接口解决。
>
> Best,
> Jark
>
> On Wed, 22 Jul 2020 at 16:27, Leonard Xu <[hidden email]<mailto:
> [hidden email]>> wrote:
> Hi，首维， Ran
>
> 感谢分享，我理解1.11新的API主要是想把 Table API 和 DataStream API 两套尽量拆分干净,
> 但看起来平台级的开发工作会依赖DataStream的一些预处理和用户逻辑。
> 我觉得这类需求对平台开发是合理，可以收集反馈下的， cc: godfrey
>
> 祝好
> Leonard Xu
>
>
> > 在 2020年7月22日，13:47，刘首维 <[hidden email]<mailto:
> [hidden email]>> 写道：
> >
> > Hi JingSong,
> >
> >
> 简单介绍一下背景，我们组的一个工作就是用户在页面写Sql,我们负责将Sql处理转换成Flink作业并在我们的平台上运行，这个转换的过程依赖我们的SQL
> SDK
> > 下面我举几个我们比较常用且感觉用1.11新API不太好实现的例子
> >
> >
> > 1. 我们现在有某种特定的Kafka数据格式，1条Kafka数据
> 会对应转换n（n为正整数）条Row数据，我们的做法是在emitDataStream的时候增加了一个process/FlatMap阶段，用于处理这种情况，这样对用户是透明的。
> > 2. 我们目前封装了一些自己的Sink，我们会在Sink之前增加一个process/Filter 用来做缓冲池/微批/数据过滤等功能
> > 3. 调整或者指定Source/Sink并行度为用户指定值，我们也是在DataStream层面上去做的
> > 4. 对于一些特殊Source Sink，他们会和KeyBy操作组合（对用户透明），我们也是在DataStream层面上去做的
> >
> >
> > 如果可以的话，能让我在API层面拿到Transformation也是能满足我需求的
> >
> >
> > ________________________________
> > 发件人: Jingsong Li <[hidden email]<mailto:[hidden email]>>
> > 发送时间: 2020年7月22日 13:26:00
> > 收件人: user-zh
> > 抄送: [hidden email]<mailto:[hidden email]>
> > 主题: Re: 关于1.11Flink SQL 全新API设计的一些问题
> >
> > 可以分享下你们为啥要拿到DataStream吗？什么场景一定离不开DataStream吗？
> >
> > Best
> > Jingsong
> >
> > On Wed, Jul 22, 2020 at 12:36 PM 刘首维 <[hidden email]<mailto:
> [hidden email]>> wrote:
> >
> >> Hi all,
> >>
> >>
> >>
> >> 很高兴看到Flink 1.11的发布，FLIP95和FLIP105也成功落地~
> >>
> >> 我们最近在调研基于1.11的SQL/Table API对我们旧有的SQL
> >>
> SDK进行升级。经过初步调研发现，基于`Factory`和`DynamicTable`的API，CatalogTable会被Planner直接变成Transformation，而不是跟之前一样可以拿到DataStream。比如之前的`StreamTableSource`#getDataStream，我可以直接获得DataStream对象的。这个动作/方法对于我们很重要，因为我们封装的SDK中，很多内部操作是在这个方法调用时触发/完成的。
> >>
> >>
> >>
> >> 所以，如果基于新的API去开发的话，我该如何获取到DataStream，或者达成类似的效果呢（尽可能不去动执行计划的话）
> >>
> >
> >
> > --
> > Best, Jingsong Lee
>
>

刘首维

答复: 关于1.11Flink SQL 全新API设计的一些问题

Hi, godfrey

好的，如果可以的话，有了相关讨论的jira或者mail可以cc一下我吗，谢谢啦

________________________________
发件人: godfrey he <[hidden email]>
发送时间: 2020年7月22日 17:49:27
收件人: user-zh
抄送: Jark Wu; [hidden email]; [hidden email]
主题: Re: 关于1.11Flink SQL 全新API设计的一些问题

Hi，首维

感谢给出非常详细的反馈。这个问题我们之前内部也有一些讨论，但由于缺乏一些真实场景，最后维持了当前的接口。
我们会根据你提供的场景进行后续讨论。

Best,
Godfrey

刘首维 <[hidden email]> 于2020年7月22日周三下午5:23写道：

Jark

Re: 关于1.11Flink SQL 全新API设计的一些问题

Administrator

Hi 首维，

我建了一个 issue 来跟进这个问题：https://issues.apache.org/jira/browse/FLINK-18674
我们可以在这个里面继续讨论需求和评估解决方案。

On Wed, 22 Jul 2020 at 18:07, 刘首维 <[hidden email]> wrote:

> Hi, godfrey
>
>
> 好的，如果可以的话，有了相关讨论的jira或者mail可以cc一下我吗，谢谢啦
>
> ________________________________
> 发件人: godfrey he <[hidden email]>
> 发送时间: 2020年7月22日 17:49:27
> 收件人: user-zh
> 抄送: Jark Wu; [hidden email]; [hidden email]
> 主题: Re: 关于1.11Flink SQL 全新API设计的一些问题
>
> Hi，首维
>
> 感谢给出非常详细的反馈。这个问题我们之前内部也有一些讨论，但由于缺乏一些真实场景，最后维持了当前的接口。
> 我们会根据你提供的场景进行后续讨论。
>
> Best,
> Godfrey
>
> 刘首维 <[hidden email]> 于2020年7月22日周三下午5:23写道：
>
> > Hi, Jark
> >
> >
> >
> > 感谢你的建议！
> >
> > 我们这边充分相信社区在SQL/Table上的苦心孤诣，也愿意跟进Flink在新版本的变化。
> >
> > 先看我遇到问题本身，你的建议确实可以帮助我解决问题。我想聊一下我对问题之外的一些想法
> >
> > ```
> >
> > > 2. 我们目前封装了一些自己的Sink，我们会在Sink之前增加一个process/Filter
> > 用来做缓冲池/微批/数据过滤等功能
> > 这个我觉得也可以封装在 SinkFunction 里面。
> >
> > ```
> >
> >
> 比如上述这个问题2，我们确实可以把它做到SinkFunction中，但是我个人认为这可能在设计上不够理想的。我个人在设计编排Function/算子的时候习惯于遵循”算子单一职责”的原则，这也是我为什么会拆分出多个process/filter算子编排到SinkFunction前面而非将这些功能耦合到SinkFunction去做。另一方面，没了DataStream，向新的API的迁移成本相对来说变得更高了一些~
> > 又或者，我们现在还有一些特殊原因，算子编排的时候会去修改TaskChain Strategy，这个时候DataStream的灵活性是必不可少的
> >
> > 考虑到Flink Task都可以拆分成Source -> Transformation -> sink
> > 三个阶段，那么能让用户可以对自己的作业针对（流或批）的运行模式下，可以有效灵活做一些自己的定制策略/优化/逻辑可能是会方便的~
> >
> >
> 诚然，DataStream的灵活性确实会是一把双刃剑，但就像@leonard提到的，平台层和应用层的目的和开发重点可能也不太一样，对Flink
> > API使用侧重点也不同。我个人还是希望可以在享受全新API设计优势同时，
> >
> > 可以继续使用DataStream（Transformation）的灵活性，助力Flink组件在我们组的开落地
> >
> >
> > 再次感谢各位的回复！
> >
> > ________________________________
> > 发件人: Jark Wu <[hidden email]>
> > 发送时间: 2020年7月22日 16:33:45
> > 收件人: user-zh
> > 抄送: godfrey he; [hidden email]; 刘首维
> > 主题: Re: 关于1.11Flink SQL 全新API设计的一些问题
> >
> > Hi，首维，
> >
> > 非常感谢反馈。与 DataStream 解耦是 FLIP-95 的一个非常重要的设计目标，这让 sink/source 对于框架来说不再是黑盒，
> > 因此将来才可以做诸如 state 兼容升级、消息顺序保障、自动并发设置等等事情。
> >
> > 关于你的一些需求，下面是我的建议和回复：
> >
> > > 1. 我们现在有某种特定的Kafka数据格式，1条Kafka数据
> >
> 会对应转换n（n为正整数）条Row数据，我们的做法是在emitDataStream的时候增加了一个process/FlatMap阶段，用于处理这种情况，这样对用户是透明的。
> > 这个理论上还属于“数据格式”的职责，所以建议做在 DeserializationSchema 上，目前 DeserializationSchema
> > 支持一对多的输出。可以参考 DebeziumJsonDeserializationSchema 的实现。
> >
> > > 2. 我们目前封装了一些自己的Sink，我们会在Sink之前增加一个process/Filter 用来做缓冲池/微批/数据过滤等功能
> > 这个我觉得也可以封装在 SinkFunction 里面。
> >
> > > 3. 调整或者指定Source/Sink并行度为用户指定值，我们也是在DataStream层面上去做的
> > 这个社区也有计划在 FLIP-95 之上支持，会提供并发（或者分区）推断的能力。
> >
> > > 4. 对于一些特殊Source Sink，他们会和KeyBy操作组合（对用户透明），我们也是在DataStream层面上去做的
> > 这个能在具体一点吗？目前像 SupportsPartitioning 接口，就可以指定数据在交给 sink 之前先做 group by
> > partition。我感觉这个可能也可以通过引入类似的接口解决。
> >
> > Best,
> > Jark
> >
> > On Wed, 22 Jul 2020 at 16:27, Leonard Xu <[hidden email]<mailto:
> > [hidden email]>> wrote:
> > Hi，首维， Ran
> >
> > 感谢分享，我理解1.11新的API主要是想把 Table API 和 DataStream API 两套尽量拆分干净,
> > 但看起来平台级的开发工作会依赖DataStream的一些预处理和用户逻辑。
> > 我觉得这类需求对平台开发是合理，可以收集反馈下的， cc: godfrey
> >
> > 祝好
> > Leonard Xu
> >
> >
> > > 在 2020年7月22日，13:47，刘首维 <[hidden email]<mailto:
> > [hidden email]>> 写道：
> > >
> > > Hi JingSong,
> > >
> > >
> >
> 简单介绍一下背景，我们组的一个工作就是用户在页面写Sql,我们负责将Sql处理转换成Flink作业并在我们的平台上运行，这个转换的过程依赖我们的SQL
> > SDK
> > > 下面我举几个我们比较常用且感觉用1.11新API不太好实现的例子
> > >
> > >
> > > 1. 我们现在有某种特定的Kafka数据格式，1条Kafka数据
> >
> 会对应转换n（n为正整数）条Row数据，我们的做法是在emitDataStream的时候增加了一个process/FlatMap阶段，用于处理这种情况，这样对用户是透明的。
> > > 2. 我们目前封装了一些自己的Sink，我们会在Sink之前增加一个process/Filter 用来做缓冲池/微批/数据过滤等功能
> > > 3. 调整或者指定Source/Sink并行度为用户指定值，我们也是在DataStream层面上去做的
> > > 4. 对于一些特殊Source Sink，他们会和KeyBy操作组合（对用户透明），我们也是在DataStream层面上去做的
> > >
> > >
> > > 如果可以的话，能让我在API层面拿到Transformation也是能满足我需求的
> > >
> > >
> > > ________________________________
> > > 发件人: Jingsong Li <[hidden email]<mailto:[hidden email]
> >>
> > > 发送时间: 2020年7月22日 13:26:00
> > > 收件人: user-zh
> > > 抄送: [hidden email]<mailto:[hidden email]>
> > > 主题: Re: 关于1.11Flink SQL 全新API设计的一些问题
> > >
> > > 可以分享下你们为啥要拿到DataStream吗？什么场景一定离不开DataStream吗？
> > >
> > > Best
> > > Jingsong
> > >
> > > On Wed, Jul 22, 2020 at 12:36 PM 刘首维 <[hidden email]
> <mailto:
> > [hidden email]>> wrote:
> > >
> > >> Hi all,
> > >>
> > >>
> > >>
> > >> 很高兴看到Flink 1.11的发布，FLIP95和FLIP105也成功落地~
> > >>
> > >> 我们最近在调研基于1.11的SQL/Table API对我们旧有的SQL
> > >>
> >
> SDK进行升级。经过初步调研发现，基于`Factory`和`DynamicTable`的API，CatalogTable会被Planner直接变成Transformation，而不是跟之前一样可以拿到DataStream。比如之前的`StreamTableSource`#getDataStream，我可以直接获得DataStream对象的。这个动作/方法对于我们很重要，因为我们封装的SDK中，很多内部操作是在这个方法调用时触发/完成的。
> > >>
> > >>
> > >>
> > >> 所以，如果基于新的API去开发的话，我该如何获取到DataStream，或者达成类似的效果呢（尽可能不去动执行计划的话）
> > >>
> > >
> > >
> > > --
> > > Best, Jingsong Lee
> >
> >
>