Hi:
有几个问题想咨询下大佬:
1.retract在什么时候触发呢?是有groupby或者窗口就默认retract吗,还是需要配置?
2.如果上游操作有retract,那么不是所有的下游都带有retract性质了?不然下游计算的数据就不准了。
3.sql操作的话,如果上游是有retract的,那下游select然后print,会把DELETE和INSERT这两条记录都print出来?
这个问题我建议你记一个 JIRA 然后提供一个可复现的程序。因为你如果是 Flink Standalone Session 模式,在 Client
端编译失败抛出如上异常,不应该跟放不放在 lib 下有什么关系。这边听你说感觉也很奇怪,可能需要本地复现一下比较好判断。
Best,
tison.
宇张 于2020年4月23日周四 上午11:53写道:
> 你的意思是,UberJar 不放在 lib 里,在用户程序里通过线程上下文 ClassLoader 能加载到
> KafkaTableSourceSinkFactory
> 吗?(同时 class loading 为 child-f
另外你 shaded 里面去 shaded com.ibm.icu 也意义不明...
Best,
tison.
tison 于2020年4月23日周四 下午3:34写道:
> 这个问题我建议你记一个 JIRA 然后提供一个可复现的程序。因为你如果是 Flink Standalone Session 模式,在 Client
> 端编译失败抛出如上异常,不应该跟放不放在 lib 下有什么关系。这边听你说感觉也很奇怪,可能需要本地复现一下比较好判断。
>
> Best,
> tison.
>
>
> 宇张 于2020年4月23日周四 上午11:53写道:
>
>> 你的意思是,Uber
Hi lec,
1.retract在什么时候触发呢?是有groupby或者窗口就默认retract吗,还是需要配置?
这个是某些算子会有这个行为,比如普通的group by,就会发送retract消息。
另外有一些算子是在某些特定配置下才会有这个行为,比如window operator,在配置了early fire或者late fire的时候。
还有些算子本身不会产生,但是会传递,比如calc算子
2.如果上游操作有retract,那么不是所有的下游都带有retract性质了?不然下游计算的数据就不准了。
这个也不绝对。大部分时候是。
这个取决于这个算子本身是不是会consume
> 如果是这样,听起来 client 的 classloading 策略没啥问题,似乎是 SPI 加载那边的 ClassLoader
有问题。之前FileSystem 相关解析就出过类似的 ClassLoader 的 BUG
@tison 不管怎么样,也得保证jar里的SPI文件包含Kafka的类,不然SPI没法找
@宇张 建议你仔细看下[1],这个pom是能打出正确的SPI文件的
[1]
https://github.com/apache/flink/blob/master/flink-table/flink-table-uber-blink/pom.xml#L104
Best,
这个难道没有一个列表,或者是配置开关之类的吗?难道只能一个一个地尝试?各种算子连接在一起,更难判断了。
Benchao Li 于2020年4月23日周四 下午3:39写道:
> Hi lec,
>
> 1.retract在什么时候触发呢?是有groupby或者窗口就默认retract吗,还是需要配置?
>
> 这个是某些算子会有这个行为,比如普通的group by,就会发送retract消息。
> 另外有一些算子是在某些特定配置下才会有这个行为,比如window operator,在配置了early fire或者late fire的时候。
> 还有些算子本身不会产生,但是会传递,比
下面配置中,如果不加flink-json模块是可以打出kafkafactory的,加了flink-json模块就只剩下
JsonRowFormatFactory一个类,kafka的Factory就打印不出来了,所以是不是某一部分导致冲突了,
但我看我先前flink1.9的工程,里面也无法打印kafkaFactory类,只有一个
GenericInMemoryCatalogFactory类,但flink1.9和1.10对比,看发布文档类加载策略有过改动
org.apache.flink:flink-connector-kafka-0.11_2.11
org.apache.flink:flin
这个暂时还没有一篇文档来介绍这部分内容。如果你要了解全部细节,可能只能从源码的角度来了解了。
lec ssmi 于2020年4月23日周四 下午3:45写道:
> 这个难道没有一个列表,或者是配置开关之类的吗?难道只能一个一个地尝试?各种算子连接在一起,更难判断了。
>
>
> Benchao Li 于2020年4月23日周四 下午3:39写道:
>
> > Hi lec,
> >
> > 1.retract在什么时候触发呢?是有groupby或者窗口就默认retract吗,还是需要配置?
> >
> > 这个是某些算子会有这个行为,比如普通的group by,就会发送retrac
谢谢。
其实,如果从DataStream编程的角度上来说,下游是能够收到一个Tuple2类型的数据,也就是能够硬编码处理retract的结果。
但是对于Table
API&SQL来说,特别是SQL,内置函数本身并没有一个增加处理Retract的逻辑(当然,可能内置算子已经包含了,我没有去看而已)。
我在编写UDAF的时候,里面有个retract方法,注释写的是: This function must be implemented
for datastream bounded over aggregate 。 是否说只有over window的时候才有retract?
另外,对于我们写的U
Hi 张,
加上这个[1]试试:
Apache Flink
UTF-8
[1]https://github.com/apache/flink/blob/master/pom.xml#L1654
Best,
Jingsong Lee
On Thu, Apr 23, 2020 at 3:56 PM 宇张 wrote:
> 下面配置中,如果不加flink-json模块是可以打出kafkafactory的,加了flink-json模块就只剩下
> JsonRowFormatFactory一个类,kafka的Factory就打印不出来了
内置的*聚合函数*应该是都能处理retract消息的。
普通的*scalar函数*不需要特殊处理,retract和append消息对它来说都是一样的。
我理解应该主要是UDAF可能需要注意一下是否需要处理retract消息,over window的确是会需要处理retract,除此之外,regular
group by也需要。
lec ssmi 于2020年4月23日周四 下午4:05写道:
> 谢谢。
> 其实,如果从DataStream编程的角度上来说,下游是能够收到一个Tuple2类型的数据,也就是能够硬编码处理retract的结果。
> 但是对于Table
> API&SQL来
那如果是两个retract算子后的流进行time interval join,
已经join成功并且发送出去的记录,也会先DELETE掉,再INSERT,然后将这两条记录发送下游?
Benchao Li 于2020年4月23日周四 下午4:11写道:
> 内置的*聚合函数*应该是都能处理retract消息的。
> 普通的*scalar函数*不需要特殊处理,retract和append消息对它来说都是一样的。
> 我理解应该主要是UDAF可能需要注意一下是否需要处理retract消息,over window的确是会需要处理retract,除此之外,regular
> group by也需
Hi
> 这个难道没有一个列表,或者是配置开关之类的吗?难道只能一个一个地尝试?各种算子连接在一起,更难判断了。
现在确实缺少这方面的文档,简单的办法可以看下这个类org.apache.flink.table.plan.nodes.datastream.DataStreamRel
如果需要了解可以看下都有哪些算子实现这些方法即可
def needsUpdatesAsRetraction: Boolean = false
def producesUpdates: Boolean = false
def consumesRetractions: Boolean = false
def pr
Hi,
我本地复现了下,用1.10.0发现的你的sql是ok的,结果也符合预期☺️,如下[1]:
看到你建了JIRA,我们在issue里继续跟进吧
祝好,
Leonard Xu
[1]
mysql> select * from order_state_cnt;
++--+--+
| order_date | product_code | cnt |
++--+--+
| 2020-04-01 | product1 |3 |
| 2020-04-01 | pro
》》加上 这部分配置之后对应的TableFactory文件里面有对应的KafkaFactory信息了,虽说程序还是无法运行,但是错误变为jar包冲突了,也就不是先前加载不到的错误;
但是感觉每次都配置这些貌似对用户不太友好。
org.codehaus.janino.CompilerFactory cannot be cast to
org.codehaus.commons.compiler.ICompilerFactory
org.apache.flink:flink-table-common
org.apache.
time interval join不允许输入是非append的。
lec ssmi 于2020年4月23日周四 下午4:18写道:
> 那如果是两个retract算子后的流进行time interval join,
> 已经join成功并且发送出去的记录,也会先DELETE掉,再INSERT,然后将这两条记录发送下游?
>
> Benchao Li 于2020年4月23日周四 下午4:11写道:
>
> > 内置的*聚合函数*应该是都能处理retract消息的。
> > 普通的*scalar函数*不需要特殊处理,retract和append消息对它来说都是一样的。
> > 我理解
我重新在源码里打了一些日志编译后,之前的问题不见了,试了好多次没有复现了,之前因为集成clickhouse 改过源码的delete代码,不知道是不是这个引起的
在2020年4月23日 16:23,Leonard Xu 写道:
Hi,
我本地复现了下,用1.10.0发现的你的sql是ok的,结果也符合预期☺️,如下[1]:
看到你建了JIRA,我们在issue里继续跟进吧
祝好,
Leonard Xu
[1]
mysql> select * from order_state_cnt;
++--+--+
| order_date | p
奇怪,目前我们使用阿里云的Blink,使用了join前的两个流,都是通过last_value 加上over
window做的,然后再做的join,然后将join的结果进行tumble window 聚合。
Benchao Li 于2020年4月23日周四 下午4:26写道:
> time interval join不允许输入是非append的。
>
>
> lec ssmi 于2020年4月23日周四 下午4:18写道:
>
> > 那如果是两个retract算子后的流进行time interval join,
> > 已经join成功并且发送出去的记录,也会先DELETE掉,再IN
阿里云上提供的Blink应该是内部版本,跟社区版本有些不一样。我刚才说的都是基于社区版本的。
lec ssmi 于2020年4月23日周四 下午4:29写道:
> 奇怪,目前我们使用阿里云的Blink,使用了join前的两个流,都是通过last_value 加上over
> window做的,然后再做的join,然后将join的结果进行tumble window 聚合。
>
> Benchao Li 于2020年4月23日周四 下午4:26写道:
>
> > time interval join不允许输入是非append的。
> >
> >
> > lec ssmi 于2020年4
Hi,
看起来应该是你之前改代码时引入的小bug,因为从代码路径和测试来看都不能复现这个问题。
另外,如果修改了源代码记得邮件里说明下,不然好南
祝好,
Leonard Xu
> 在 2020年4月23日,16:26,1101300123 写道:
>
> 我重新在源码里打了一些日志编译后,之前的问题不见了,试了好多次没有复现了,之前因为集成clickhouse
> 改过源码的delete代码,不知道是不是这个引起的
> 在2020年4月23日 16:23,Leonard Xu 写道:
> Hi,
> 我本地复现了下,用1.10.0发现的你的sql是ok的,结果也符合预期☺
不能把lib下有的jar打进去。
比如flink-table-planner-blink,lib下也有一份flink-table-planner-blink
把这一堆去掉吧:
org.apache.flink:flink-table-common
org.apache.flink:flink-table-api-java
org.apache.flink:flink-table-api-java-bridge_2.11
org.apache.flink:flink-table-planner-blink_2.11
Best
可以建个JIRA来更新文档吗?现在retract的文档的确有点confuse
Best,
Jingsong Lee
On Thu, Apr 23, 2020 at 4:33 PM Benchao Li wrote:
> 阿里云上提供的Blink应该是内部版本,跟社区版本有些不一样。我刚才说的都是基于社区版本的。
>
> lec ssmi 于2020年4月23日周四 下午4:29写道:
>
> > 奇怪,目前我们使用阿里云的Blink,使用了join前的两个流,都是通过last_value 加上over
> > window做的,然后再做的join,然后将join的结果进行tumbl
其实我想说,如果说sql内置的算子,包括UDF这种ScalarFunction默认都是能够处理retract的话,
我们举一个最简单的例子:sum函数,那内部实现是否需要具有一个类似于
if( type='DELETE'){
sum=sum-value
} else if(type='INSERT'){
sum=sum+value
}
的逻辑呢?
但是在ScalarFunction中,只实现了eval方法,也就是只有 INSERT的那部分相加的逻辑,没有DELETE那部分相减的逻辑。
Benchao Li 于2020年4月23
嗯嗯,刚刚把blink包改为provided后程序能正常运行了,但接下来工程打包的时候都要手动加下面这些感觉比较麻烦,不过能运行就好。
感谢答疑
org.apache.flink:flink-connector-kafka-base_2.11
On Thu, Apr 23, 2020 at 4:36 PM Jingsong Li wrote:
> 不能把lib下有的jar打进去。
>
> 比如flink-table-planner-blink,lib下也有一份flink-table-planner-blink
>
> 把这一堆去掉吧:
> org.apache.f
Hi Jingsong,
我建了一个jira[1] 来跟踪这个事情。
Hi lec,
sum函数不属于scalar函数。sum的内置实现是有retract的版本的,参考:IntSumWithRetractAggFunction
scalar function不需要这样子处理,因为它本身没有状态。scalar function对于消息的类型是不需要判断的,处理过程都是一样的。
[1] https://issues.apache.org/jira/browse/FLINK-17343
lec ssmi 于2020年4月23日周四 下午4:41写道:
> 其实我想说,如果说sql内置的算
那也就是说UDF这种ScalarFunction,是没有办法处理Retract的了?因为它把DELETE记录和INSERT记录都做了相同的操作。
Benchao Li 于2020年4月23日周四 下午4:54写道:
> Hi Jingsong,
> 我建了一个jira[1] 来跟踪这个事情。
>
> Hi lec,
> sum函数不属于scalar函数。sum的内置实现是有retract的版本的,参考:IntSumWithRetractAggFunction
> scalar function不需要这样子处理,因为它本身没有状态。scalar function对于消息的类型是不需要判断
hi,我这面在win10 基于Standlone搭建了一个Flink1.10(好吧,就是解压启动)
,然后执行start-cluster.bat启动服务,会弹出两个dos窗口,一个jm、一个tm(猜的),
但是几秒后tm 对应的dos窗口闪退导致程序没办法申请资源,这个不知道是flink问题还是win10问题,但是flink1.9是正常的
??
-- --
??: ""
不是这个意思。是说scalar function不需要额外处理,所以它一套逻辑就可以处理两种类型的消息了。
它不需要区分消息类型,只需要处理消息本身(消息类型是在header里)。
lec ssmi 于2020年4月23日周四 下午5:00写道:
> 那也就是说UDF这种ScalarFunction,是没有办法处理Retract的了?因为它把DELETE记录和INSERT记录都做了相同的操作。
>
> Benchao Li 于2020年4月23日周四 下午4:54写道:
>
> > Hi Jingsong,
> > 我建了一个jira[1] 来跟踪这个事情。
> >
> > Hi l
明白了,谢谢。
Benchao Li 于2020年4月23日周四 下午5:08写道:
> 不是这个意思。是说scalar function不需要额外处理,所以它一套逻辑就可以处理两种类型的消息了。
> 它不需要区分消息类型,只需要处理消息本身(消息类型是在header里)。
>
> lec ssmi 于2020年4月23日周四 下午5:00写道:
>
> > 那也就是说UDF这种ScalarFunction,是没有办法处理Retract的了?因为它把DELETE记录和INSERT记录都做了相同的操作。
> >
> > Benchao Li 于2020年4月23日周四 下午4:54写
呃,是的,某些设置的默认值都变为null了,所以tm启动报错,依次让设置这三个值,但这三个设置后报错变为下面的了,请问这个要怎么搞
taskmanager.cpu.cores: 3
taskmanager.memory.task.heap.size: 256mb
taskmanager.memory.managed.size: 256mb
org.apache.flink.configuration.IllegalConfigurationException: Failed to
create TaskExecutorResourceSpec
at
org.apache.flink
现在搭建测试环境都要改配置文件了,感觉还是以前的小白式启动(解压运行)友好一点,哈哈
On Thu, Apr 23, 2020 at 5:23 PM 宇张 wrote:
> 呃,是的,某些设置的默认值都变为null了,所以tm启动报错,依次让设置这三个值,但这三个设置后报错变为下面的了,请问这个要怎么搞
> taskmanager.cpu.cores: 3
> taskmanager.memory.task.heap.size: 256mb
> taskmanager.memory.managed.size: 256mb
>
> org.apache.flink.configu
Hi,1.10 的配置文件都设好了吗?可以把 log 目录下的日志贴上来看看。
蒋佳成(Jiacheng Jiang) <920334...@qq.com> 于2020年4月23日周四 下午5:04写道:
> 查看日志估计是内存没有设置
>
>
>
>
> -- 原始邮件 --
> 发件人: "宇张" 发送时间: 2020年4月23日(星期四) 下午5:03
> 收件人: "user-zh" 主题: flink1.10基于win10搭建Standlone服务
>
>
>
> hi,我这面在win10 基于Standlone搭
👍👍👍
oliver yunchang 于2020年4月23日周四 上午12:32写道:
> 非常感谢Leonard Xu和zhisheng的回复
>
> > es index 的 mapping 是否提前设置好了?
> 提前设置好了,提前创建索引的mapping如下:
> {
> "xxx-2020.04.23": {
> "mappings": {
> "doc": {
> "dynamic_templates": [
> {
> "string_fields": {
>
Caused by: org.apache.flink.configuration.IllegalConfigurationException:
The network memory min (64 mb) and max (1 gb) mismatch, the network memory
has to be resolved and set to a fixed value before task executor starts
??1.10??
-- ???
好吧,windows不在维护了。。。
https://issues.apache.org/jira/browse/FLINK-15925
On Thu, Apr 23, 2020 at 5:30 PM 蒋佳成(Jiacheng Jiang) <920334...@qq.com>
wrote:
>
> Caused by: org.apache.flink.configuration.IllegalConfigurationException:
>
> The network memory min (64 mb) and max (1 gb) mismatch, the network m
Hi 社区,
最近我在弄 Flink on k8s,使用的 Flink 版本为 Flink 1.6。作业模式为 standalone per job 模式。
我在创建启动 jobmanager 的时候,设置的 taskmanager.heap.mb 为 2000 mb,虽然在 flink web ui
上面看到的 jobmanager 的配置, taskmanager.heap.mb 的确是 2000mb,在我启动 taskmanager
deployment 的时候,我登录到 其中一个 pod 上看,发现 taskmanager 启动的 -xms 和 -xmx 都是 922mb。
不好意思,刚才看了一下源码:
[image: image.png]
这个是over window的聚合操作。
这个类实现没有实现producesUpdates 和producesRetractions,
而这两个方法的默认值都是False。是否说明,只能有INSERT类型的记录?
如果是的话,不就是说明over window操作的输出是一个Append-only stream?
lec ssmi 于2020年4月23日周四 下午5:13写道:
> 明白了,谢谢。
>
> Benchao Li 于2020年4月23日周四 下午5:08写道:
>
>> 不是这个意思。是说scalar f
嗯嗯,是的。over window的确是不支持retract输入,只支持append输入。
而且它也只有append输出。
lec ssmi 于2020年4月23日周四 下午6:32写道:
> 不好意思,刚才看了一下源码:
> [image: image.png]
> 这个是over window的聚合操作。
> 这个类实现没有实现producesUpdates 和producesRetractions,
> 而这两个方法的默认值都是False。是否说明,只能有INSERT类型的记录?
> 如果是的话,不就是说明over window操作的输出是一个Append-only stream
应该没有其他地方去写 flink-conf.yaml,能把具体用来打镜像、动态写配置的命令或者脚本发一下吗?
另外你这个问题还有一种解决方案,是 taskmanager.heap.mb 通过 -D 参数传给 taskmanager.sh。可以在
docker-compose.yaml 中 taskmanager command 处追加 -Dtaskmanager.heap.mb=2000m
Thank you~
Xintong Song
On Thu, Apr 23, 2020 at 5:59 PM LakeShen wrote:
> Hi 社区,
>
> 最近我在弄 Fli
Hi xuefli,
感谢你的建议。虽然我不是 PMC,但对其中的一些问题我也想来谈一下自己的理解。
1. 你说的是 shaded 的依赖吗?考虑到用户代码本身也可能依赖一些常见的库(例如 guava 等),为了防止和用户代码的版本冲突,Flink
才对常用的库进行了 shade,这样就相当于调用 Flink 自己的代码一样。这个机制正是为了解决版本兼容问题引入的。
2. open 应该只是进行数据源的连接操作,不同的 slot 处理的是不同的 input
split,不会重复读取数据(但的确可能重复连接数据源)。对于 OLAP 场景这个的确是个优化点,但对于 batch / stre
Hi Xintong,
非常感谢你的回复。想再请教一个问题,什么地方会使用到 docker-compose.yaml 呢。
我目前使用一种绕开的方式解决这个问题,就是在 DockerFile 打镜像的时候,先把 conf 目录拷贝出来(这个里面的 TaskManger
内存动态传入的)
然后在 config.sh 中,强行设置了 FLINK-CONF-DIR . 但是我觉得你的方式更优雅一些。
所以想问一下 什么地方会使用到 docker-compose.yaml呢 。
Best,
LakeShen
Xintong Song 于2020年4月24日周五 上午10:49写道:
抱歉,我刚刚说的是 docker-compose.yaml 是只用 docker 不用 kubernetes 的情况。
对于 kubernetes,如果你是按照官方文档[1]推荐的方法部署 flink 的,那么直接把这个参数加在
taskmanager-deployment.yaml 的 args 处应该就可以了。
> args:
- taskmanager
*- Dtaskmanager.heap.size=2000m*
Thank you~
Xintong Song
[1]
https://ci.apache.org/projects/flink/flink-docs-
hi all,
blink模式下没法将table 转为 dataset , 所以如果直接collect了, 请问有类似方法可以获取到
结果用于代码调试么?
--
Best,
Jun Su
hi all,
找到了源码中BatchTableEnvUtil类使用了CollectTableSink来做collect的逻辑,
但是下方代码运用了源码内部的private方法, 看起来不允许外部调用:
def collect[T](
tEnv: TableEnvironment,
table: Table,
sink: CollectTableSink[T],
jobName: Option[String]): Seq[T] = {
val typeSerializer = fromDataTypeToLegacyInfo(sink.getConsu
45 matches
Mail list logo