Flink-1.11.2版本FileSystem connector问题

问题现象是这样的 1. flink 实时的往hdfs 目录 /warehouse/rt_ods/ed_cell_num_info/写数据，以天为分区 2. 然后我们启动了一个定时任务在 hive上新建partition，是T - 1的，比如今天6点执行，新建 昨天 的分区。 3. 会报，inprogress文件找不到的错误，错误如下 : ``` org.apache.flink.streaming.runtime.tasks.AsynchronousException: Caught exception while processing timer. at

Flink-1.11.2版本 Filesystem-connector 问题

问题现象是这样的 1. flink 实时的往hdfs 目录 /warehouse/rt_ods/ed_cell_num_info/写数据，以天为分区 2. 然后我们启动了一个定时任务在 hive上新建partition，是T - 1的，比如今天6点执行，新建 昨天 的分区。 3. 会报，inprogress文件找不到的错误，错误如下 : ``` org.apache.flink.streaming.runtime.tasks.AsynchronousException: Caught exception while processing timer. at

Flink-1.11.2版本 Filesystem-connector 问题

问题现象是这样的 1. flink 实时的往hdfs 目录 /warehouse/rt_ods/ed_cell_num_info/写数据，以天为分区 2. 然后我们启动了一个定时任务在 hive上新建partition，是T - 1的，比如今天6点执行，新建 昨天 的分区。 3. 会报，inprogress文件找不到的错误，错误如下 : ``` org.apache.flink.streaming.runtime.tasks.AsynchronousException: Caught exception while processing timer. at

flink版本：1.11.2， 从kafka读取数据写到hdfs，运行一段时间报错

flink版本：1.11.2， 从kafka读取数据写到hdfs，运行一段时间报错， sql 和 日志如下： ```sql CREATE TABLE T_ED_CELL_NUM_INFO_SRC( bigBox STRING, edCode STRING, mBoxAmount INT, mFaultBoxAmount INT, mFaultReserveBoxAmount INT, mReserveBoxAmount INT, mUseReserveBox INT, mUsedBoxCount INT,

Flink 1.11.2运行一段时间后，会报ResourceManager leader changed to new address null的异常

8 个slot，8个并行度，jm是2G，tm配置的是8G，其它的任务配置是 ``` SET 'execution.checkpointing.interval' = '5min'; SET 'execution.checkpointing.min-pause' = '10s'; SET 'min.idle.state.retention.time' = '1d'; SET 'max.idle.state.retention.time' = '25h'; SET 'checkpoint.with.rocksdb' = 'true'; set

回复: Re:flink 1.11.1 使用flinksql，jdbc ，后台数据库设置主键的情况下，upsert不生效

你的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 我也测试了，也不行。

回复: flink-1.11.2 的 内存溢出问题

内存不够用。 可以尝试按报错信息中提示的，把 `taskmanager.memory.task.off-heap.size` 调大看看。 只调大 TM 的总内存没有用的，不会增加 job 可用的 direct 内存。 Thank you~ Xintong Song On Mon, Nov 16, 2020 at 6:38 PM 史 正超 wrote: > flink-on-yarn . per-job模式，重启是kafka的group.id > 没变，应该是接着offset消费的，但是任务启动不起来。不知道是不是一段时间后

回复: flink-1.11.2 的 内存溢出问题

~ 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 > java.lang.OutOfMemoryError: Direct buffer memory. The direct out-of-memory > error has

flink-1.11.2 的 内存溢出问题

使用的是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

flink-1.11.2 job启动不起来，

启动命令：run -d -m yarn-cluster -p 12 -yjm 1600 -ytm 12288 -ys 12 -ynm xxx -yqu flink-critical -j /app/flink-1.11.2/executor/fcbox-streaming-sql-platform-1.11.jar --sqlid 17 --jobName realtime_app_kpi_dis_day_16 12个并行度， 12个slot，启动不了 Caused by: java.util.concurrent.CompletionException:

回复: Re:回复: flink-1.11.2 执行checkpoint失败

09:13:34，"史 正超" 写道： >这是个思路，谢谢回复，我先试下。 > >发件人: 赵一旦 >发送时间: 2020年11月13日 2:05 >收件人: user-zh@flink.apache.org >主题: Re: flink-1.11.2 执行checkpoint失败 > >如果超时也是错误的话，推进你设置容忍数量为INT最大值。因为有时候检查点失败或超时并不真正代表反压，至少我的生产中是这样。 >有部分情况，压力高，但刚刚好的情况下，会

回复: flink-1.11.2 执行checkpoint失败

这是个思路，谢谢回复，我先试下。 发件人: 赵一旦 发送时间: 2020年11月13日 2:05 收件人: user-zh@flink.apache.org 主题: Re: flink-1.11.2 执行checkpoint失败 如果超时也是错误的话，推进你设置容忍数量为INT最大值。因为有时候检查点失败或超时并不真正代表反压，至少我的生产中是这样。 有部分情况，压力高，但刚刚好的情况下，会出现部分检查点失败的case，但实际压力刚刚能顶住。没必要因此导致任务失败。 史 正超 于2020年11月13日周五 上午10

回复: flink-1.11.2 执行checkpoint失败

rable failure threshold. 顺着这个问个问题。 检查点“超时”是否计算进入checkpoint failure呢？ 史 正超 于2020年11月12日周四 下午9:23写道： > 执行checkpoint失败，报下面的错。 > 2020-11-12 21:04:56 > org.apache.flink.util.FlinkRuntimeException: Exceeded checkpoint tolerab

回复: 回复：flink-1.11.2 执行checkpoint失败

99), -99, direction, IF(boxType IS NOT NULL, boxType, -99); 发件人: 史 正超 发送时间: 2020年11月13日 1:50 收件人: user-zh@flink.apache.org 主题: 回复: 回复：flink-1.11.2 执行checkpoint失败 没有，用的是jdbc sink，先是 三张change log的 left join，然后 再分别与两张mysql维表的join，执行checkpoint超时。sql太复杂了，我越写越没信心。。。

回复: 回复：flink-1.11.2 执行checkpoint失败

sic_edinfo_16 FOR SYSTEM_TIME AS OF t1.proctime d1 ON t1.assetCode = d1.ed_code ) t2, LATERAL TABLE(RecursionDiscode(dis_code)) AS T(disCode, dislv) ) t3 JOIN edbasic.basic_district_16 FOR SYSTEM_TIME AS OF t3.proctime d2 ON t3.disCode = d2.dis_code ) GROUP BY REPLACE(SUBSTR(closeTime, 1, 10), '-', ''),

flink-1.11.2 执行checkpoint失败

执行checkpoint失败，报下面的错。 2020-11-12 21:04:56 org.apache.flink.util.FlinkRuntimeException: Exceeded checkpoint tolerable failure threshold. at org.apache.flink.runtime.checkpoint.CheckpointFailureManager.handleJobLevelCheckpointException(CheckpointFailureManager.java:66) at

回复: Flink cdc mysql 字段是datetime类型时0000-00-00 00:00:00会被flink转成1970-01-01T00:00

在flink sql 中用STRING表示datetime，这样的话后续的可操作性会比较大些。 发件人: 丁浩浩 <18579099...@163.com> 发送时间: 2020年11月11日 6:37 收件人: user-zh@flink.apache.org 主题: Flink cdc mysql 字段是datetime类型时-00-00 00:00:00会被flink转成1970-01-01T00:00 当我mysql字段时datetime并且字段值是-00-00

回复: flink1.11的cdc功能对消息顺序性的处理

Canal可以配置分区策略：配置保证相同id的记录都发到同一个分区，比如 `db.table1:id` 这样就保证了数据的有序。 发送自 Windows 10 版邮件应用 发件人: Jark Wu 发送时间: 2020年11月5日 21:28 收件人: user-zh 主题: Re: flink1.11的cdc功能对消息顺序性的处理 我理解你说的是对 pk 的更新的场景。

回复: Flink1.11.0 sql org.apache.flink.connector.jdbc.dialect.MySQLDialect发现疑似bug

应该是没有问题的， 首先你在flink sql中指定的primary key 应该要与mysql中的唯一索引或者主键对应。 其次那个方法里组装出来的语句 类似下面的语句: INSERT INTO `tablename`(`key1`, `key2`, `f1`, `f2`) VALUES (?, ?, ?, ?) ON DUPLICATE KEY UPDATE `key1`=VALUES(`key1`), `key2`=VALUES(`key2`), `f1`=VALUES(`f1`), `f2`=VALUES(`f2`) 里面已经包含了定义的key,

回复: union all 丢失部分数据

你的union all上面，也就是insert into 的第一条select 末尾有 分号 ‘;’,， 这样的话我感觉第二条select是不会执行的。还有另一个问题是你把分号去掉，我感觉数据会被 覆盖的。因为我最近也有使用union all的场景。 我觉得你应该这样组装 你的sql : ```sql Insert into xxx Select d1, d2, count(1) From ( Select * from a Union all Select * from b, ) Group by

回复: TUMBLE函数不支持 回撤流

canal-json 的format也是会有delete 和update的数据的，同样changelog-json也是。他们的都支持 INSERT UPDATE DELETE, 相关代码如下: @Override public ChangelogMode getChangelogMode() { return ChangelogMode.newBuilder() .addContainedKind(RowKind.INSERT) .addContainedKind(RowKind.UPDATE_BEFORE)

回复: Temporal table join currently only supports 'FOR SYSTEM_TIME AS OF' left table's proctime field, doesn't support 'PROCTIME()

是什么？ 在1.12前, temporal join 一个view 是不支持的, 只能直接关联一个可以lookup的table 祝好 Leonard > 在 2020年11月2日，19:40，史 正超 写道： > > Source表定义的proctime，在view中select后，不能用于Temporal table join. > sql如下： > ```sql > > CREATE TABLE SourceA ( >id STRING, >procTime AS PROCTIME()

Temporal table join currently only supports 'FOR SYSTEM_TIME AS OF' left table's proctime field, doesn't support 'PROCTIME()

Source表定义的proctime，在view中select后，不能用于Temporal table join. sql如下： ```sql CREATE TABLE SourceA ( id STRING, procTime AS PROCTIME() ) WITH ( 'connector' = 'kafka-0.11' ); CREATE TABLE DimTable ( id STRING, dim1 STRING, primary key(id) NOT

回复: 回复： flink1.11连接mysql问题

的就是flink1.11啊 在2020年11月2日 11:33，酷酷的浑蛋 写道： 你看历史的回复，用的就是flink1.11，最新的flink-1.11.2也试过了还是有这个问题，而且我是在flink sql中使用 在2020年11月2日 11:30，史 正超 写道： 你用的flink哪个版本，flink-1.11已经加了连接失效重连的功能了，应该没有这个问题了。如果不是flink-1.11.x版本的，也可以参考flink-1.11的jdbc-connector的实现，或者用SELECT 1 语句保活连接

回复: flink-cdc-mysql 使用时遇到的问题

建议你发下完整的错误信息，然后不要发图片（看不到图片），这样让更多人看到后，问题就解决快了。 发件人: yangxusun9 发送时间: 2020年11月2日 3:44 收件人: user-zh@flink.apache.org 主题: Re: flink-cdc-mysql 使用时遇到的问题 应该不是的，我用的是root账户 -- Sent from: http://apache-flink.147419.n8.nabble.com/

回复: Re:回复：flink1.11连接mysql问题

你用的flink哪个版本，flink-1.11已经加了连接失效重连的功能了，应该没有这个问题了。如果不是flink-1.11.x版本的，也可以参考flink-1.11的jdbc-connector的实现，或者用SELECT 1 语句保活连接。 发件人: 酷酷的浑蛋 发送时间: 2020年11月2日 2:28 收件人: user-zh@flink.apache.org 主题: Re:回复：flink1.11连接mysql问题 没有解决，隔一段时间就会报这个超时错误 在

单个sink表 和多个source表的统计

Hi, 我这边有这样一种场景，我的一张sink表中的 指标统计分别来源于多张source表， 比如cnt_a 来源于 count(a), cnt_b来源于 count(b)。 统计结果 sink到mysql时我现在只能用union all，如下： ```sql CREATE TABLE SourceA ( day_time STRING, a STRING ) WITH ( 'connector' = 'kafka' ) CREATE TABLE SourceB ( day_time

回复: flink任务挂掉后自动重启

可以通过使用rest api 定时拉取checkpoints的状态 rest api uri为 {cluster}/jobs/{jobid}/checkpoints 返回的是json，里面有最近一次的ck记录和 历史记录。 发件人: bradyMk 发送时间: 2020年10月30日 6:52 收件人: user-zh@flink.apache.org 主题: Re: flink任务挂掉后自动重启 谢谢您的解答~

回复: flink实时流中如何实时获取当前时间

在source表上加上 proctime AS PROCTIME()的字段 ， 下游取的时候就用proctime转，注意时差的问题。比如用FROM_UNIXTIME(CAST(proctime AS BIGINT), '-MM-dd HH:mm:ss') 函数时，在table config里设置时区 ``` java streamTableEnv.getConfig().setLocalTimeZone(ZoneOffset.ofHours(8)); `` 发件人: zjfpla...@hotmail.com

回复: Re: 关于flink-sql 维表join问题

我最近的写的业务和你差不多，不过我关联的是两张表，一张mysql的维表，一张binlog的流表。最开始我都是left join ，发现只有binlog流表有数据时才计算。 后面 我做嵌套的查询，先与mysql维表inner join（直接join）,然后再套一层query 再与流表left join，现在情况正常。就算binlog的流表没有数据也有计算到。 发件人: Jark Wu 发送时间: 2020年10月28日 7:24 收件人: user-zh 主题: Re: Re: 关于flink-sql 维表join问题

回复: flink sql CDC 同步到 hive的问题咨询

目前还不支持，因为hive不支持更新，而且filesystem connector的实现并不是DynamicTableFactory (FLIP-95 connector)，还是之前的老接口。 如果你不希望做更新操作入到hive的话，有一种思路是更改mysql-cdc的源码，让它支持的ChangeLogMode只为INSERT，然后增加一个`DebeziumDeserializationSchema`类，把每条消息当成 insert 消息去解析，把所有字段声明出来。 发件人: zhongbaoluo 发送时间:

回复: 回复：回复: flink 自定义udf注册后不能使用

你这样创建试一下，或者换个名字试试 CREATE TEMPORARY SYSTEM FUNCTION imei_encrypt AS 'com.intsig.flink.udf.IMEIEncrypt' LANGUAGE JAVA; 我刚才创建了一个 UpperCase的function，也是一样的错误，用TEMPORARY SYSTEM覆盖系统的函数(有可能存在)后，就可以了，换个字也可以 发件人: 奔跑的小飞袁 发送时间: 2020年10月16日 6:47 收件人:

回复: flink 自定义udf注册后不能使用

Hi, 从日志上看 是说 匹配不到 imei_encrypt的UDF，有可能是sql里传的字段和imei_encrypt的参数不匹配， 能看下你的具体代码和udf的声明吗 发件人: 奔跑的小飞袁 发送时间: 2020年10月16日 3:30 收件人: user-zh@flink.apache.org 主题: flink 自定义udf注册后不能使用 hello 我在使用flinkSQL注册udf时，发生了以下错误，这是我定义有问题还是flink的bug

回复: kafka topic字段 不全的统计场景

的统计场景 group by id应该就可以了吧，其他几个字段用last value或者first value[1],还有就是考虑迟到的数据怎么处理 [1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/functions/systemFunctions.html On Thu, Oct 15, 2020 at 5:01 PM 史 正超 wrote: > 大佬们，现在我有个场景： > 一个kafka 主题 有 4个字段 ， id, field2, field3, field4，

kafka topic字段 不全的统计场景

大佬们，现在我有个场景： 一个kafka 主题 有 4个字段 ， id, field2, field3, field4，其中id 是唯一标识， 但是有个问题是，并不是每个消息都会带上全量的字段消息，只有id是固有的字段。然后需要把id, field2, field3, field4 作为一个维度 统计， 比如有如下 kafka消息: {"id": 1, "field2":"b"} {"id": 1, "field3":"c", "field4":"d"} 那么 按照维度 count(1) (group by id, field2, field3, field4)

回复: 回复: flink-SQL1.11版本对map类型中value的空指针异常

.getValue()); result.put(key, value); } return new GenericMapData(result); }; } 应该是 createNullableConverter final AvroToRowDataConverter valueConverter = createNullableConverter(extractValueTypeToAvroMap(type)); ____ 发件人: 史 正超 发送时间: 2020年10月14日 5:22 收件人

回复: 回复: flink-SQL1.11版本对map类型中value的空指针异常

确定吗？我这边测试还是有问题，这应该是avro 的一个bug。 发件人: 奔跑的小飞袁 发送时间: 2020年10月14日 3:29 收件人: user-zh@flink.apache.org 主题: Re: 回复: flink-SQL1.11版本对map类型中value的空指针异常 我尝试使用MAP来定义我的类型，问题已经解决，谢谢 -- Sent from: http://apache-flink.147419.n8.nabble.com/

回复: flink-SQL1.11版本对map类型中value的空指针异常

所以我的建议是用avro的规范，你可以这样定义你的MAP类型: MAP 发件人: 史 正超 发送时间: 2020年10月14日 2:45 收件人: user-zh 主题: 回复: flink-SQL1.11版本对map类型中value的空指针异常 但是 方法上有这样的一个注释：Creates a runtime converter which assuming input object is not null. 代码这样写的前提是，不允许对象的值为null的

回复: flink-SQL1.11版本对map类型中value的空指针异常

但是 方法上有这样的一个注释：Creates a runtime converter which assuming input object is not null. 代码这样写的前提是，不允许对象的值为null的。 发件人: Benchao Li 发送时间: 2020年10月14日 2:34 收件人: user-zh 主题: Re: flink-SQL1.11版本对map类型中value的空指针异常 嗯，这应该是一个实现的bug，可以提个issue修复一下~ 史 正超 于2020年10月14日周三 上午10

回复: flink-SQL1.11版本对map类型中value的空指针异常

从你的异常来看，你用的format是 avro, 我看了下源码，他对varchar类型的covert和json不一样，avro的代码是这样的： case CHAR: case VARCHAR: return avroObject -> StringData.fromString(avroObject.toString()); 所以，你的map类型的value值为null，会报空指针异常的。 发件人: 奔跑的小飞袁 发送时间: 2020年10月14日 1:46 收件人:

回复: flink-connector-jdbc 落入mysql的时候timestamp 有时差问题

的sink表的 url后面加上 =Asia/Shanghai 发件人: dushang 发送时间: 2020年10月13日 8:38 收件人: user-zh@flink.apache.org 主题: Re: flink-connector-jdbc 落入mysql的时候timestamp 有时差问题 time_zone SYSTEM system_time_zone SYSTEM 我是通过

回复: Flink 1.11 table.executeInsert 程序退出

这个是一个已知问题，可以看看这个jira: https://issues.apache.org/jira/browse/FLINK-18545 规避这个问题的话，可以不用执行 tableEnv.execute("jobname"); 直接用 executeSql 就可以了，遇到INSERT语句就能生成job了。 发件人: HunterXHunter <1356469...@qq.com> 发送时间: 2020年9月30日 2:32 收件人: user-zh@flink.apache.org 主题: Flink 1.11

回复: 回复：关于flink sql cdc

HI, Kyle Zhang, 我刚才重现了你的问题，虽然你的mysql binlog设置是ROW格式，但是不排除其它session更改了binlog_format格式。重现步骤： 1. 登录mysql客户端(注意用cmd登录) 执行语句， SET SESSION binlog_format='MIXED'; SET SESSION tx_isolation='REPEATABLE-READ'; COMMIT; 2. 随便update或者insert一条语句。 然后就得到了和你一样的错误： 2020-09-30 10:46:37.607

一个FlinkJob消费多个kafka topic消息问题

1.平常工作中经常会有同一个统计表中会包含多个不同的统计指标,比如：post_count, send_count 2.然而这些指标来自不同的kafka 消息体 3.有没有在不用uninon all的情况下，向sink 表中写入各自topic的数据，因为union all有很多0值来填充