各位,请教个问题
目前使用flink往ck写入数据，使用的是datastream
api以及rocksdb状态后端，程序中了开了两个窗口，都是10秒级别。同时还使用了sql进行group by
求和，求和的操作没有加窗口,同时streamtableenv 设置了状态生存时间为10s.
在跟离线端对比数据的时候发现，一段时间内的数据跟离线差异不大，从0点-17点（数据的事件时间），但是18点（事件时间）以后的数据实时端差异量特别大。
目前在webui上发现整个管理内存的使用率已经占满，不知道是否跟这个有关系。
还有一点现象是，今天的数据我们对比了ck上实时的表（正确的），总体数据量还是要小
补充一下图片
https://s2.loli.net/2022/06/02/C5it4rPFgmJlopZ.png
https://s2.loli.net/2022/06/02/3ri2HIv1RsAawBW.png
https://s2.loli.net/2022/06/02/efVWPvXCYFwhgTp.png
https://s2.loli.net/2022/06/02/9UptbNaWvs7xwXC.png
在 2022-06-02 11:38:24，"lxk" 写道：
各位,请教个问题
目前使用flink往c
flink 版本:1.14.4
目前在使用flink interval join进行数据关联，在测试的时候发现一个问题，就是使用interval join完之后数据会丢失，但是使用sql
api，直接进行join，数据是正常的，没有丢失。
水印是直接使用kafka 自带的时间戳生成watermark
以下是代码 ---interval join
SingleOutputStreamOperator headerFullStream =
headerFilterStream.keyBy(data -> data.getId())
.intervalJoin(filterItemSt
amt, bom_type, last_updated_at,
display_qty, is_first_flag])
在 2022-06-10 11:02:56，"Shengkai Fang" 写道：
>你好，能提供下具体的 plan 供大家查看下吗？
>
>你可以直接 使用 tEnv.executeSql("Explain JSON_EXECUTION_PLAN
>").print() 打印下相关的信息。
>
>Best,
>Shengkai
>
&
考文档[1]。可以试下使用SQL interval
>join会不会丢数据（注意设置state的ttl），从而判断是数据的问题还是datastream api的问题。
>
>
>
>
>[1]
>https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/queries/joins/#interval-joins
>
>
>
>
>
>
>
>--
>
>Best！
>Xuy
义和实现上都是有区别的。所以你直接拿datastream
>api的interval join和sql上的普通join结果对比，其实是有问题的。所以我之前的建议是让你试下让sql也使用interval
>join，这样双方才有可比性。
>
>
>另外sql中设置的table.exec.state.ttl这个参数，只是代表的state会20s清空过期数据，但我看你要比较的时间窗口是-10s和20s，貌似也不大一样。
>
>
>
>
>--
>
>Best！
>Xuyang
>
>
markAssigner(rowtime=[rowtime], watermark=[(rowtime -
2000:INTERVAL SECOND)])
+- Calc(select=[id, order_id, row_num, goods_id, qty, p_sp_sub_amt,
bom_type, display_qty, CAST(SUBSTRING(last_updated_at, 0, 19)) AS rowtime])
+- TableSourceScan(table=[[default_catalog, default_data
可以把堆栈的日志打印出来看看
在 2022-06-10 18:15:53，"Summer" 写道：
使用 FinkUI 上传 Flink 任务 Jar 时，任务启动失败。 这时候JVM Metaspace就会异常增加。 这是什么原因？
me，也就是说，如果右流的 event
>>time 的更新会影响左流的数据清理。比如说右流的时间点到了 12：00，join 条件要求左流的时间不会晚于右流的时间 1h，那么左流
>>11:00之前的数据都可以被清理了。
>>
>>对于第三点，我觉得是不能的。目前的 inner join + state 清理无法覆盖 event time 的window join 的。
>>
>>best,
>>Shengkai
>>
>>lxk7...@163.com 于2022年6
; 不合理的 watermark 设置在 interval join 就会导致丢数据。设置 ttl 情况下，如果某个 key
>> 的数据频繁访问情况下，那么这个数据就不会过期。
>>
>> > 我的做法是排查了tm和jm的日志，没有看见写的表的配置的日志。
>>
>> 我记得日志是会打印相关的日志。能提一些相关的日志吗？
>>
>> best,
>> Shengkai
>>
>> lxk 于2022年6月14日周二 20:04写道
你好，图片挂了，可以尝试使用图床工具上传图片。
在 2022-06-21 09:42:54，"amber_...@qq.com.INVALID" 写道：
您好！
我使用flink1.14.4，sqlserver-cdc-2.2.1，yarn-per-job模式提交任务；
当我提交普通数据同步任务时，一切正常；
当我提交JOIN+聚合任务时，checkpoint无法正常工作，具体表现为无任何checkpoint记录，且Task Managed
Memory使用率始终是100%；
以下是我的checkpoint配置：
在使用interval join的时候有一些疑问，希望大家能够帮忙解答一下
https://pic.imgdb.cn/item/62c5015b5be16ec74ac2b23f.png
官方文档中说会从两个流中的 timestamp 中取最大值，看了下源码确实是这样
https://pic.imgdb.cn/item/62c51e8b5be16ec74ae22880.png
我的问题是:
1.这里的timestamp和watermark有什么区别？
2.interval
join中watermark是怎么计算的？两个流取最大的timestamp之后，watermark跟这个最大的time
>watermark是从source部分生成的水印个，然后向后传播。
>
>以分窗口为例，数据记录的timestamp用于决定数据划分入哪个窗口。
>watermark用于决定窗口啥时候闭合，比如窗口是0-5s，那么当watermark达到5s的时候，窗口就会闭合。
>
>考虑数据不一定能及时到达，可以让watermark=max(timestamp)-30s。30s即可容忍给的数据乱序的程度。
>
>lxk 于2022年7月6日周三 13:36写道：
>>
>> 在使用int
谢谢
我现在使用的是直连的方式，也没有关闭preparedstatement和resultset,但是没有发生过内存泄漏的问题，请问了解这背后的原因吗
在 2022-07-25 13:53:42，"Lijie Wang" 写道：
>Hi,
>根据我的经验，使用连接池时，至少需要及时关掉 statement/ResultSet，否则查询的结果会一直缓存，会有内存泄漏的问题。
>
>Best，
>Lijie
>
>lxk7...@163.com 于2022年7月23日周六 15:34写道：
>
>>
>> 目前的项目中，需要使用外部数据库进行实时的look
Flink版本：1.16.0
目前在使用Flink SQL进行多流关联，并写入Clickhouse中
具体代码如下：
select \
header.id as id, \
LAST_VALUE(header.order_status), \
LAST_VALUE(header.customer_id), \
LAST_VALUE(header.shop_id), \
LAST_VALUE(header.parent_order_id), \
LAST_VALUE(header.order_at), \
LAST_VALUE(header.pay_at), \
LAST_VALUE(head
qq.com.invalid> 于2023年1月31日周二 17:22写道：
>
>> 你可以测试不写入clickhouse是否还存在反压，如果不是因为写入瓶颈的话就从你的处理逻辑优化了
>>
>>
>> 发件人: lxk
>> 发送时间: 2023年1月31日 15:16
>> 收件人: user-zh@flink.apache.org
>> 主题: Flink SQL 如何优化以及处理反压
>>
>> Flink版本：1.16.0
>> 目前在使用Flink
Flink version:1.16
java version: jdk1.8.0_251
问题：最近上线的Flink程序，频繁young
gc，几秒一次，在调整了新生代大小之后，还是没有解决，目前整个jvm堆大小是3.57g。因此想通过程序内存情况来分析哪里问题有问题，我们通过yarn上的applicationId,使用ps
-ef|grep 1666758697316_2639108,找到对应的pid,最后执行 jmap -dump:format
b,file=user.dump 26326
命令生成dump文件，但我们测试了很多个程序，只要一开始dump,都会对线上程序产生影响
你好，可以dump下内存分析
在 2023-02-16 10:05:19，"Fei Han" 写道：
>@all
>大家好！我的Flink 版本是1.14.5。CDC版本是2.2.1。在on yarn 运行一段时间后会出现如下报错:
>org.apache.flink.runtime.jobmaster.JobMasterException: TaskManager with id
>container_e506_1673750933366_49579_01_02(hdp-server-010.yigongpin.com:8041)
> is n
Flink版本：1.16
目前公司针对Flink版本进行升级，从Flink1.14升级到Flink1.16,代码没做任何调整，但是在写入kafka的时候报错：
2023-02-17 15:03:19
org.apache.kafka.common.KafkaException: Cannot execute transactional method
because we are in an error state
at
org.apache.kafka.clients.producer.internals.TransactionManager.maybeFailWithError(Tra
gt;
>从`Caused by: org.apache.kafka.common.errors.ClusterAuthorizationException:
>Cluster authorization failed.`这个错误看起来像是权限错误，可以你检查下是否有权限问题
>
>Best,
>Shammon
>
>On Fri, Feb 17, 2023 at 6:29 PM lxk wrote:
>
>> Flink版本：1.16
>> 目前公司针对Flink版本进行升级，从Flink1.14升级到Flink1.16,代码没做任何调整，但是在写入kafka的时候报错：
>> 202
我尝试调整了参数，具体数值如下
akka.ask.timeout: 900s
但还是报同样的错
在 2023-02-17 17:32:51，"Guo Thompson" 写道：
>可能是jm 和 tm之间的心跳时间太短了， dump的过程会stop the world，tm就不响应jm的heartbeat了；
>
>lxk 于2023年2月14日周二 14:32写道：
>
>> Flink version:1.16
>> java version: jdk1.8.0_251
>&g
>>
>> [1]
>>
>> https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/#advanced-fault-tolerance-options
>>
>> Best,
>> Weihua
>>
>>
>> On Mon, Feb 20, 2023 at 1:58 PM lxk wrote:
>>
>> > 我尝试调整了参数，具体数值如下
flink版本:1.14
目前使用的是对一个数据量比较小的流进行广播，另外的主流跟这个广播流进行匹配处理。
在主程序中，我设置了状态过期策略：
SingleOutputStreamOperator baiduStream =
env.addSource(adBaiduClick).map(data -> JSON.parseObject(data,
AdvertiseClick.class)).name("BaiDuAdClick");
MapStateDescriptor baiduInfoMap = new
MapStateDescriptor<>("advertiseInf
eam/fault-tolerance/state/#state-time-to-live-ttl>对
>State TTL 的描述；
>
>On Mon, May 15, 2023 at 11:05 AM lxk wrote:
>
>> flink版本:1.14
>> 目前使用的是对一个数据量比较小的流进行广播，另外的主流跟这个广播流进行匹配处理。
>> 在主程序中，我设置了状态过期策略：
>>SingleOutputStreamOperator baiduStream =
>&g
gt;On Mon, May 15, 2023 at 1:41 PM lxk wrote:
>
>> 这么看来，广播流好像不适合在生产中使用，状态会无限止的增长。这块官方有计划增加ttl功能吗。
>> 或者使用广播流的时候有没有什么能够手动清理状态的方法？
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>&
你好，可以把相关代码贴上来吗，方便大家进行分析。如果使用sql的话还可以把执行计划贴上来。
在 2023-05-15 17:11:42，"小昌同学" 写道：
>各位老师，请教一下我在使用table API进行编程的时候，报错信息为”Caused by:
>org.apache.flink.api.common.InvalidProgramException: Table program cannot be
>compiled. This is a bug. Please file an issue. “
>flink使用版本为1.14，请问一下有相关社
thodInvocation(UnitCompiler.java:4423)
> at
> org.codehaus.janino.UnitCompiler$16.visitMethodInvocation(UnitCompiler.java:4396)
> at org.codehaus.janino.Java$MethodInvocation.accept(Java.java:5073)
> at org.codehaus.janino.UnitCompiler.compileGet(UnitCo
你好，可以先看看官方文档中关于事件时间和水印的介绍
https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/concepts/time/
如果你发了多条数据，但是都是同样数据的话，水印没有推进，窗口就不会触发
在 2023-05-25 10:00:36，"小昌同学" 写道：
>是的 我发送了很多数据，发现窗口还是没有触发
>
>
>| |
>小昌同学
>|
>|
>ccc0606fight...@163.com
>|
> 回复的原邮件
>| 发件人 | yidan
项目中使用精准一次语义写入kafka,代码和配置如下：
Properties producerProperties = MyKafkaUtil.getProducerProperties();
KafkaSink kafkaSink = KafkaSink.builder()
.setBootstrapServers(Event2Kafka.parameterTool.get("bootstrap.server"))
.setRecordSerializer(KafkaRecordSerializationSchema.builder()
.setTopic(Event2Kafka.para
上封邮件发错了，重新发一下。项目中使用精准一次语义写入kafka,代码和配置如下：
写入代码如下：
Properties producerProperties = MyKafkaUtil.getProducerProperties();
KafkaSink kafkaSink = KafkaSink.builder()
.setBootstrapServers(Event2Kafka.parameterTool.get("bootstrap.server"))
.setRecordSerializer(KafkaRecordSerializationSchema.builder()
.se
hi,
jdbc创建链接是在SimpleJdbcConnectionProvider这个类中实现的，至于真正创建链接，则是由DriverManager来处理。
关于连接数，则是跟你的并行度有关。
在 2023-05-30 13:55:57，"小昌同学" 写道：
>各位老师，请教一下关于flink jdbcsink 连接数的问题；
>我使用代码如下：在以下代码中，我查看了一下源码，并没有找到sink到MySQL时关于连接数的设定，请问这一块关于连接数的设定我应该怎么写呀；
>谢谢各位老师的指导
>
>|
>outPutInfoStream.addSink(J
你好，整个程序有反压吗
在 2023-07-10 15:32:44，"jiaot...@mail.jj.cn" 写道：
>Hello,
> 我定义了一个pattern （a->b->c->d->e->f->g）在10分钟内匹配，通过在WebUI上查看任务很快在cep节点
> busy(max)100%，我发现通过增加cep节点的并发度并不能解决问题，且checkpoint随着时间的推移状态大小越来越大，数据应该存在大量堆积。数据源同时消费4个kafka
> topic
> （setTopics），采用默认水位线间隔时间，我发现4个to
在flink内需要使用mybatis做些简化查询的工作，目前我的使用方式如下
public class MybatisUtil {
private static final Logger LOGGER = LogFactory.createNewLogger("MybatisUtil");
private static ThreadLocal tl = new ThreadLocal();
private static SqlSessionFactory factory = null;
//private static SqlSession sqlSession
相关配置：
Flink:1.16
| Checkpointing Mode | Exactly Once |
| Checkpoint Storage | FileSystemCheckpointStorage |
| State Backend | EmbeddedRocksDBStateBackend |
| Interval | 8m 0s |
我有一个程序，主要是用来统计一些热门商品之类的数据
具体代码如下：
.keyBy(data -> data.getShopId() + data.getYh_productid())
.window(TumblingEventTimeWi
好的，谢谢老师
在 2023-07-26 21:04:20，"Jiabao Sun" 写道：
>SqlSession 需要关闭，建议使用 SqlSessionManager，可以不用手动关闭 SqlSession。
>
>
>On 2023/07/18 02:13:16 lxk wrote:
>> 在flink内需要使用mybatis做些简化查询的工作，目前我的使用方式如下
>>
>> public class MybatisUtil {
&
35 matches
Mail list logo
