退订
hive 1.1.0
-- --
??:
"user-zh"
需求:现有table t 三个字段
CREATE TABLE t (
abigint,
bbigint,
cbigint,
PRIMARY KEY (a) NOT ENFORCED
) WITH (
...
);
我们的场景只想根据主键a更新部分字段b,其余的字段c保持不变,
例如mysql 支持:insert into t(a,b) select 1,2 on duplicate key update b=2;
主键重复的时候只更新字段b,字段c的值不变。但是flink sql 目前只支持全字段更新:insert into t(a
用的什么版本这个应该已经在1.13中修复了https://issues.apache.org/jira/browse/FLINK-18726
不行的话可以在ddl中限制列的数量
--
发件人:Ye Chen
发送时间:2021年8月2日(星期一) 11:37
收件人:user-zh ; silence
主 题:Re:回复:场景题:Flink SQL 不支持 INSERT INTO… ON DUPLICATE KEY UPDATE ?
你好,我试了一下,如果
cancel with savepoint是之前的接口了,会造成kafka数据的重复。新的stop with
savepoint则会在做savepoint的时候,不再发送数据,从而避免了重复数据,哭啼可以参考
https://ci.apache.org/projects/flink/flink-docs-master/docs/ops/state/savepoints/
Jim Chen 于2021年8月2日周一 下午2:33写道:
> 我是通过savepoint的方式重启的,命令如下:
>
> cancel command:
>
> /home/datadev/flink-1.12.
你好,我们用的1.11版本。
需求:table t 有三个字段(a,b,c)
我们的场景只想根据主键a更新部分字段b,其余的字段c原有的值保持不变,
例如mysql 支持:
insert into t(a,b) select 1,2 on duplicate key update b=2;
主键重复的时候只更新字段b,字段c的值不变。
但是flink sql 目前只支持全字段更新:insert into t(a,b,c) select 1,2,3 。
我在sql-client测试了一下:insert into t(a,b) select 1,2 on duplicate key up
你好,
如果我沒有理解錯你的應用場景的話,你想達成的結果應該是類似這篇討論 [1] 裡提到的問題對吧?
從最新的 flink 文檔 [2] 中來看應該無法透過你期望的 on duplicate key 語句來實現,
或許可以嘗試在 SELECT 語句上達成,舉例來說你可以在原有的 select 語句之外多添加 group by,如下:
insert into t select a, last_value(b ignore nulls) as b, last_value(c
> ignore nulls) as c from $(original_select_statement) gro
補充一下,從代碼上來看, last_value 原本的實現似乎就是獲取 last "non-null" value 了。
https://github.com/apache/flink/blob/master/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/LastValueAggFunction.java
Tony Wei 於 2021年8月2日 週一 下午5:33寫道:
> 你好,
>
> 如果我沒有理解錯你的應用場景的話,你想達
可以自定义 KafkaAppender,然后可以从 System.getProperty("log.file")
获取你要的信息维度数据,比如这个可以提取到作业 application id,container_id,是 jobmanager 还是
taskmanager,还可以根据配置只提取想要的级别日志,最后将打的依赖放到 lib 目录下即可
yujianbo <15205029...@163.com> 于2021年6月15日周二 下午7:34写道:
> 最新详细配置,可以看看我的博客:
> https://blog.csdn.net/weixin_44500374/article/
Hi 刘建刚,
我使用了stop with savepoint,但是还是发现,下游有重复数据。
停止命令:
/home/datadev/flink-1.12.2/flink-1.12.2/bin/flink stop \
-yid application_1625497885855_703064 \
-p
hdfs://ztcluster/flink_realtime_warehouse/checkpoint/UserClickLogAll/savepoint
\
-d 55e7ebb6fa38faaba61b4b9a7cd89827
重启命令:
/home/datadev/flink-1
下游如何发现重复数据的,下游的isolation.level是不是read_committed
在 2021-08-02 18:14:27,"Jim Chen" 写道:
>Hi 刘建刚,
>我使用了stop with savepoint,但是还是发现,下游有重复数据。
>停止命令:
>/home/datadev/flink-1.12.2/flink-1.12.2/bin/flink stop \
>-yid application_1625497885855_703064 \
>-p
>hdfs://ztcluster/flink_realtime_warehouse/checkpoi
flink最新特性中有非对齐检查点的特性,可以用来解决一些反压下的任务,但如果用了这个,还能保证精确一次吗?对齐的检查点有清晰的快照N~N +
1之间的边界,这个会将数据混在一起,如何在恢复的时候保证精确一次?
不可以,会存在重复消费的问题,如果buffer没有对齐的话,job重启,那么这些buffer的数据就会清空,然后相关的subtask会重新消费一遍。
张锴 于2021年8月2日周一 下午6:53写道:
> flink最新特性中有非对齐检查点的特性,可以用来解决一些反压下的任务,但如果用了这个,还能保证精确一次吗?对齐的检查点有清晰的快照N~N +
> 1之间的边界,这个会将数据混在一起,如何在恢复的时候保证精确一次?
>
我不太懂,下游的isolation.level是不是read_committed是啥意思。
我是把topic A中的partitionId和offset写到消息体中,然后flink程序,把消息写到下游的topic B中。将topic
B实时写到hive上,然后在hive表中,根据partitionId和offset去重,发现有重复消费了
东东 于2021年8月2日周一 下午6:20写道:
> 下游如何发现重复数据的,下游的isolation.level是不是read_committed
>
>
> 在 2021-08-02 18:14:27,"Jim Chen" 写道:
> >Hi
从topic B实时写到hive,这个job需要配置 isolation.level 为
read_committed,否则会把还没有提交甚至是已经终止的事务消息读出来,这样就很难不出现重复了。
在 2021-08-02 19:00:13,"Jim Chen" 写道:
>我不太懂,下游的isolation.level是不是read_committed是啥意思。
>我是把topic A中的partitionId和offset写到消息体中,然后flink程序,把消息写到下游的topic B中。将topic
>B实时写到hive上,然后在hive表中,根据partitionId和offse
对于每一个subtask,边界仍然是清晰的,所以精确一次可以保证,只不过ck会变大。
在 2021-08-02 18:53:11,"张锴" 写道:
>flink最新特性中有非对齐检查点的特性,可以用来解决一些反压下的任务,但如果用了这个,还能保证精确一次吗?对齐的检查点有清晰的快照N~N +
>1之间的边界,这个会将数据混在一起,如何在恢复的时候保证精确一次?
这个原理能说明一下吗,咋做到的
东东 于2021年8月2日周一 下午7:16写道:
> 对于每一个subtask,边界仍然是清晰的,所以精确一次可以保证,只不过ck会变大。
>
> 在 2021-08-02 18:53:11,"张锴" 写道:
> >flink最新特性中有非对齐检查点的特性,可以用来解决一些反压下的任务,但如果用了这个,还能保证精确一次吗?对齐的检查点有清晰的快照N~N +
> >1之间的边界,这个会将数据混在一起,如何在恢复的时候保证精确一次?
>
Hi!
shimin huang 说的可能是原本的 at least once 的 checkpoint 机制,这种 checkpoint 原本就是不对齐的。
Flink 1.13 完善了 exactly once 条件下的不对齐 checkpoint 机制,因此这是能保证 exactly once
的。实现原理简单来说就是把还没处理的数据一起写到 state 里,下次恢复的时候把这些还没处理的数据也恢复出来接着处理。
张锴 于2021年8月2日周一 下午7:20写道:
> 这个原理能说明一下吗,咋做到的
>
> 东东 于2021年8月2日周一 下午7:16写道:
>
> > 对
谢谢你,受教了
Caizhi Weng 于 2021年8月2日周一 19:28写道:
> Hi!
>
> shimin huang 说的可能是原本的 at least once 的 checkpoint 机制,这种 checkpoint 原本就是不对齐的。
>
> Flink 1.13 完善了 exactly once 条件下的不对齐 checkpoint 机制,因此这是能保证 exactly once
> 的。实现原理简单来说就是把还没处理的数据一起写到 state 里,下次恢复的时候把这些还没处理的数据也恢复出来接着处理。
>
> 张锴 于2021年8月2日周一 下午7:20写道
非常感谢,受益良多。
best regards
在 2021-08-02 17:33:14,"Tony Wei" 写道:
>你好,
>
>如果我沒有理解錯你的應用場景的話,你想達成的結果應該是類似這篇討論 [1] 裡提到的問題對吧?
>從最新的 flink 文檔 [2] 中來看應該無法透過你期望的 on duplicate key 語句來實現,
>或許可以嘗試在 SELECT 語句上達成,舉例來說你可以在原有的 select 語句之外多添加 group by,如下:
>
>insert into t select a, last_value(b ignore nul
开发者您好,
我正在尝试在Kubernetes上部署Flink 1.12.2, 使用的是native
application部署模式。但是在测试中发现,当将作业并行度调大之后,各种timeout时有发生。根据监控看,JM和TM容器的cpu和内存都没有使用到k8s给分配的量。
在尝试调大akka.ask.timeout至1分钟,和heartbeat.timeout至2分钟之后,各种超时现象得以缓解。
我的问题是,当设置较大并行度(比如128)时,akka超时和心跳超时的各种现象都是正常的吗?如果不正常,需要用什么方式去troubleshot问题的根源呢?另外单纯一味调大各个组件的超时时
AKKA timeout
java.util.concurrent.CompletionException:
org.apache.flink.client.deployment.application.ApplicationExecutionException:
Could not execute application.
at
java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:292)
~[?:1.8.0_282]
更正,这个是akka timeout exception
java.util.concurrent.CompletionException:
org.apache.flink.client.deployment.application.ApplicationExecutionException:
Could not execute application.
at
java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:292)
~[?:1.8.0_282
Taskmanager timeout:
java.util.concurrent.CompletionException:
org.apache.flink.client.deployment.application.ApplicationExecutionException:
Could not execute application.
at
java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:292)
~[?:1.8.0_282]
你好,
请问一下为什么要设置128并行度,这个数值有点太大了,出于什么考虑设置的
在 2021-08-03 14:02:53,"Chenyu Zheng" 写道:
开发者您好,
我正在尝试在Kubernetes上部署Flink 1.12.2,使用的是native
application部署模式。但是在测试中发现,当将作业并行度调大之后,各种timeout时有发生。根据监控看,JM和TM容器的cpu和内存都没有使用到k8s给分配的量。
在尝试调大akka.ask.timeout至1分钟,和heartbeat.timeout至2分钟之后,各种超时现象得以缓解。
25 matches
Mail list logo