这个程序用于测试 flink kafka exactly once, 如果普通提交可以正常运行, 但如果从 savepoint 中恢复就会报下面的错误
kafka server 端, 配置了 transaction.max.timeout.ms = 360 client producer 端
配置了 transaction.timeout.ms = 90
参考代码:
https://gist.github.com/giraffe-tree/15c5f707d9dfe3221959ae37b4e9d786
2020-11-17 15:24:51
Issue when integrate with hive 2.1.1
Exception in thread "main" java.lang.IncompatibleClassChangeError:
Implementing class
at java.lang.ClassLoader.defineClass1(Native Method)
at java.lang.ClassLoader.defineClass(ClassLoader.java:763)
at
理论上一个 TM 可以拆分成多少 slot 并没有硬性的限制，但是并不是说并发越大，性能就一定越好。
增大并发，会增加作业对内存的需求。TM 上的 slot 数量过多时，可能会造成 GC 压力大、网络内存不足、OOM 等情况。另外，同一个 TM 上的
slot 多了，运行的 task 多了，也会给框架造成一定的压力。
建议先观察一下 TM 的 cpu 使用情况，如果作业确实存在处理性能不足（延迟增大、存在反压）同时 TM container 的 cpu
（多核）利用率上不去，再考虑调大并发。
Thank you~
Xintong Song
On Tue, Nov 17,
可以的。不过你在声明MapStateDescriptor的时候要用 GenericTypeInfo了，并且会有一定的性能损失。
Best,
Guowei
On Tue, Nov 17, 2020 at 11:55 AM Lei Wang wrote:
> 下面的业务逻辑
>
> robot 传感器上报的信息，先按 robotId keyBy，之后要遍历很多的规则。每个规则存储一个之前的对象，实现如下：
>
> private transient MapState state;
>
> for (Entry entry :
>
下面的业务逻辑
robot 传感器上报的信息，先按 robotId keyBy，之后要遍历很多的规则。每个规则存储一个之前的对象，实现如下：
private transient MapState state;
for (Entry entry :
RulesFactory.getChargerTwoRecordRules().entrySet()) {
String faultName = entry.getKey();
IChargerTwoRecordRule rule = entry.getValue();
RobotData old =
Hi,
你确定是在Flink SQL 里使用 upsert 语法？ 我理解是不支持的
另外你flink里声明connector DDL 中的主键应该和你在Mysql表的主键一致。
祝好
Leonard
> 在 2020年11月17日，09:12，鱼子酱 <384939...@qq.com> 写道：
>
> upsert
是的。 目前还不支持。
1.12 版本会支持。
你定义 watermark 目的是什么呢？做 window 聚合？
On Tue, 17 Nov 2020 at 10:53, shimin huang
wrote:
> 报错日志:
> ```
> Currently, defining WATERMARK on a changelog source is not supported
> ```
>
Hello,
我使用 MySQLDialect 在本地确认了下，
1. 在数据库需要建主键，因为建了主键 “INSERT INTO ... ON DUPLICATE KEY UPDATE”[1] 语句的 upsert
语义才会生效。
2. 需要在 DDL 中定义 'PRIMARY KEY'，因为需要根据 ‘PRIMARY KEY’ 确认是否使用 'upsert query' [2]
[1]
报错日志:
```
Currently, defining WATERMARK on a changelog source is not supported
```
需要建立mysql表request_date,terminal_no的联合主键，mysql的upsert是基于"INSERT INTO ... ON
DUPLICATE KEY UPDATE..."实现的。
ddl中可以不声明mysql主健。
鱼子酱 <384939...@qq.com>于2020年11月17日 周二09:13写道：
> 我写的是upsert呀。。。
> insert into 我也测试了，也不行。
>
> 是MySQL数据库本身里面的表需要建立一个主键吗？还是只有flink里面建表的时候写就行呢？
>
>
>
> --
> Sent from:
你的sql里用的是 Tumble窗口，不是一个回撤流，不会有更新的，只有insert
发件人: 鱼子酱 <384939...@qq.com>
发送时间: 2020年11月17日 1:12
收件人: user-zh@flink.apache.org
主题: Re: Re:flink 1.11.1 使用flinksql，jdbc ，后台数据库设置主键的情况下，upsert不生效
我写的是upsert呀。。。
insert into 我也测试了，也不行。
>
> 好的，谢谢回复，那请问下 taskmanager.memory.task.off-heap.size 这个参数可以通过 下面代码动态设置吗？
>
> streamTableEnv.getConfig().getConfiguration().setString(key, value);
>
不可以的，这个是集群配置。
可以通过 flink-conf.yaml 配置文件进行配置，或者在提交作业时通过 -yD key=value 的方式动态指定。
Thank you~
Xintong Song
On Tue, Nov 17, 2020 at 9:31 AM
??, ;
taskmanager.memory.task.off-heap.size taskmanager??
streamTableEnv.getConfig().getConfiguration().setString(key, value);
----
??:
我写的是upsert呀。。。
insert into 我也测试了，也不行。
是MySQL数据库本身里面的表需要建立一个主键吗？还是只有flink里面建表的时候写就行呢？
--
Sent from: http://apache-flink.147419.n8.nabble.com/
Hi，
这个版本是支持的。
其中插入语句是 "insert into " 而不是 “update into”？
在 2020-11-16 17:04:23，"鱼子酱" <384939...@qq.com> 写道：
>请问使用mysql数据库时，使用flinksql，已经设置主键的情况下，相同主键的记录没有更新，而是越来越多，
>是目前不支持还是我使用的方法不对呢？
>版本：flink 1.11.1
>
>关键的2个sql如下
>
>create table open_and_close_terminal_minute_1 (
> request_date
Hi,
据我了解会弃用的，新的connector都会用DynamicTableSourceFactory，一般稳定一两个版本后社区会弃用，
另外这个是比较底层的实现，sql用户应该感知不到，如果有自定义connector的开发都建议用DynamicTableSourceFactory。
祝好
Leonard Xu
> 在 2020年11月16日，19:54，Luna Wong 写道：
>
> FLIP-95都实现后有了DynamicTableSourceFactory那么TableSourceFactory会弃用吗？
FLIP-95都实现后有了DynamicTableSourceFactory那么TableSourceFactory会弃用吗？
好的，谢谢回复，那请问下 taskmanager.memory.task.off-heap.size 这个参数可以通过 下面代码动态设置吗？
streamTableEnv.getConfig().getConfiguration().setString(key, value);
发件人: Xintong Song
发送时间: 2020年11月16日 10:59
收件人: user-zh
主题: Re: flink-1.11.2 的 内存溢出问题
那应该不存在内存泄露的问题。应该就是 job 需要的 direct
我也遇到这种乱序问题，楼主，你那边在sql层面解决了么？
--
Sent from: http://apache-flink.147419.n8.nabble.com/
按照我在工作中经验，有过几次需要重启zk集群，我是单个zk节点逐个重启。结论是导致了flink集群中任务的全部自动重启（基于最近一次的ckpt）。这对任务还是有一定影响的，因为ckpt是10分钟一次，会导致瞬间压力变高。
问下这个合理嘛，还是我配置的有问题or操作有问题。
https://issues.apache.org/jira/browse/FLINK-20170
这是我今天提的issue。
Jackson这样反序列化会把BigDECIMAL转成Double。超过12位的小数精度丢失。这种情况我得怎么办。只能先当做STRING类型处理或修改下JSON这个包的源码重新变一下。
还有其他的最佳实践吗
flink-on-yarn .
per-job模式，重启是kafka的group.id没变，应该是接着offset消费的，但是任务启动不起来。不知道是不是一段时间后，积压导致的。
发件人: Xintong Song
发送时间: 2020年11月16日 10:11
收件人: user-zh
主题: Re: flink-1.11.2 的 内存溢出问题
是什么部署模式呢？standalone？
之前任务运行一段时间报错之后，重新运行的时候是所有 TM 都重启了吗？还是有复用之前的 TM？
Thank you~
是什么部署模式呢？standalone？
之前任务运行一段时间报错之后，重新运行的时候是所有 TM 都重启了吗？还是有复用之前的 TM？
Thank you~
Xintong Song
On Mon, Nov 16, 2020 at 5:53 PM 史 正超 wrote:
> 使用的是rocksdb, 并行度是5，1个tm, 5个slot，tm 内存给
> 10G，启动任务报下面的错误。之前有启动成功过，运行一段时间后，也是报内存溢出，然后接成原来的offset启动任务，直接启动不起来了。
>
> 2020-11-16 17:44:52
>
*环境配置如下：*
hbase-1.4.13
flink-1.11.1
python-3.6.1
pyflink-1.0
*已做配置如下：*
1.hadoop classpath下已经加入hbase路径 （:/opt/hbase/hbase-1.4.13/lib/*）
2.程序ddl配置如下：
source_ddl = """CREATE TABLE MySourceTable (
hbase_rowkey_name varchar, cf1 ROW) WITH (
'connector.type' = 'hbase',
*环境配置如下：*
hbase-1.4.13
flink-1.11.1
python-3.6.1
pyflink-1.0
*已做配置如下：*
1.hadoop classpath下已经加入hbase路径 （:/opt/hbase/hbase-1.4.13/lib/*）
2.程序ddl配置如下：
source_ddl = """CREATE TABLE MySourceTable (
hbase_rowkey_name varchar, cf1 ROW) WITH (
'connector.type' = 'hbase',
请问使用mysql数据库时，使用flinksql，已经设置主键的情况下，相同主键的记录没有更新，而是越来越多，
是目前不支持还是我使用的方法不对呢？
版本：flink 1.11.1
关键的2个sql如下
create table open_and_close_terminal_minute_1 (
request_date varchar
,terminal_no varchar
,logon_time varchar
,logout_time varchar
,insert_time varchar
请问使用mysql数据库时，使用flinksql，已经设置主键的情况下，相同主键的记录没有更新，而是越来越多，
是目前不支持还是我使用的方法不对呢？
版本：flink 1.11.1
关键的2个sql如下
create table open_and_close_terminal_minute_1 (
request_date varchar
,terminal_no varchar
,logon_time varchar
,logout_time varchar
,insert_time varchar
使用的是rocksdb, 并行度是5，1个tm, 5个slot，tm 内存给
10G，启动任务报下面的错误。之前有启动成功过，运行一段时间后，也是报内存溢出，然后接成原来的offset启动任务，直接启动不起来了。
2020-11-16 17:44:52
java.lang.OutOfMemoryError: Direct buffer memory. The direct out-of-memory
error has occurred. This can mean two things: either job(s) require(s) a larger
size of JVM
不太理解你说的Pod系统占用多少内存是什么意思，Pod并不是虚拟机，而是docker container来进行的轻量虚拟化
和宿主机是共用内核的，本身不会带来额外的内存开销
至于Pod的内存设置，你说的是对的。Pod的limit并不是和JVM的heap内存相等的，因为还有offheap的内存以及JVM的overhead
所以你会看到JVM的参数并不是和Pod的limit完全相等。Pod的limit是根据TaskManager能够使用的最大内存来设置的，具体
每部分的内存配置和你可以参考社区文档[1].
[1].
Hi，Community：
目前sql-client和sql-gateway只能支持单条SQL statement，这样就没法利用multiple insert的优化。如下：
INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE
'%Rubber%'
INSERT INTO GlassOrders SELECT product, amount FROM Orders WHERE product LIKE
'%Glass%'
select 时候带上这个字段？ 只要这个字段没有参与计算，就不会被物化。
如果是 window 处理后，还需要具有时间属性的字段，可以参考[1]
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/queries.html#group-windows
在 2020-11-16 14:57:39，"周虓岗" 写道：
>
>
>
>
>
>
>
>
>
>不是，我值得是table api可以带event time。 如果整个使用sql表达，怎么把time
1、试了下
在test表中增加一个proctime
CREATE TABLE test (
`id` INT,
`name` VARCHAR(255),
`time` TIMESTAMP(3),
`status` INT,
`proctime` AS PROCTIME(),
PRIMARY KEY(id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'localhost',
'port' = '3306',
'username' = 'no_lock',
'password' =
32 matches
Mail list logo
