Re: flink on yarn 异常停电问题咨询

Hi 我觉得Flink 作业恢复失败时，作业本身很难确定失败是checkpoint的文件块损坏之类的原因。如果你的作业做过savepoint，可以尝试从指定的savepoint恢复作业 Best, Shammon On Thu, Mar 9, 2023 at 10:06 PM guanyq wrote: > 前提 > 1.flink配置了高可用 > 2.flink配置checkpoint数为10 > 3.yarn集群配置了任务恢复 > 疑问 > yarn集群停电重启后，恢复flink任务时，如果最近的checkpoint由于停电导致块损坏，是否会尝试从其他checkpoint启动

Re: flink问题咨询

Hi 我个人觉得可以将你现在的process计算分为两部分，你提到每隔20s触发的delta计算部分放到stream2部分，类似于这种形式 stream1.keyBy().connect(stream2.keyBy().process(处理增量，每20秒触发输出)).process(根据增量更新ListState) 这样不需要从ListState中去查找哪些数据被更新了 Best, Shammon On Thu, Mar 9, 2023 at 10:48 AM 陈隽尧 wrote: > 您好， > > >

flink on yarn 异常停电问题咨询

前提 1.flink配置了高可用 2.flink配置checkpoint数为10 3.yarn集群配置了任务恢复 疑问 yarn集群停电重启后，恢复flink任务时，如果最近的checkpoint由于停电导致块损坏，是否会尝试从其他checkpoint启动

flink问题咨询

您好， 我是flink一名新用户，最近在项目中需要用到flink完成一项业务功能，但目前遇到了一些一些困难，想咨询你一下是否有合适的解决方案，期待您的回信 问题背景：我们需要基于股票交易流水和股票行情去计算股票账户层面的一些指标（为简化场景，假定账户指标只有持仓量，买入均价，市值），页面前端20s刷新一次，指标计算想基于flink的dataStream Api实现，但遇到一个问题，目前初步想法如下，请flink大神帮忙指导 初步方案设想：假定stream1: 股票交易流水， stream2：股票行情流水

Re: Re:Re: Re: 使用flinksql将kafka的数据同步mysql，在源kafka进行删除操作，mysql收到op为d的数据进行删除操作但是与此同时会新增一条当前数据以前的旧数据。

从 plan 上看起来在 sink 节点这里因为推导不出 upsert key 加上了 SinkUpsertMaterializer[1], 这里会按照 sink 表定义的主键进行 keyby shuffle[2], 只能保证最终一致性. 另外你的操作描述中 schema 为三列, 但 DDL 是四列, 且格式乱了. 一些可能的建议如下 1. 如果上游数据有主键并且也是 rowid 的话, 建议在 Flink source 表上声明 PK, 避免额外生成 materializer 节点; 同时注意在声明 Flink source 表时不要带上 metadata 列 (比如 op),

Re: Flink异步Hbase导致Too many open files异常

+1 有遇到过类似 fd 泄露的问题。注意 close 的时候buffer 数据刷盘, 然后资源关闭，future cancel。 Best Regards, Ran Tao Weihua Hu 于2023年3月8日周三 16:52写道： > Hi, > > 通过代码看作业在Failover 时的确会有 HBaseClient 的资源泄露。 > > 在 HbaseDimensionAsyncFunc 中重写一下 close 方法，释放掉 HBaseClient。 > > Best, > Weihua > > > On Wed, Mar 8, 2023 at 4:19 PM

Re: Flink异步Hbase导致Too many open files异常

Hi, 通过代码看作业在Failover 时的确会有 HBaseClient 的资源泄露。 在 HbaseDimensionAsyncFunc 中重写一下 close 方法，释放掉 HBaseClient。 Best, Weihua On Wed, Mar 8, 2023 at 4:19 PM aiden <18765295...@163.com> wrote: > Hi > 我在使用Async Hbase时频繁遇到too many open file异常，程序自动重启后会立即报错，具体报错日志如下： > 2023-03-08 16:15:39 >

Re: Re: flink on K8S(operator) 如何获取 Accumulator

Hi 像上面提到的，jobClient.get().getAccumulators()会从flink集群获取作业相关信息，如果是application模式，作业结束后flink集群也会退出。你可以通过其他方式，包括session集群运行或者启动history server等方式，也可以通过自定义metrics等输出到其他系统 Best, Shammon On Tue, Mar 7, 2023 at 11:27 PM 李银苗 wrote: > 退订

Re:Re: flink on K8S(operator) 如何获取 Accumulator

退订

Re: flink on K8S(operator) 如何获取 Accumulator

Hi, 按照你的描述，我猜测你使用的是 Application 模式吧？这种模式下 user code 会在 JobManager 侧执行，Job 执行结束后会直接 shutdown cluster。 可以尝试使用 session mode[1] 部署 cluster [1] https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/resource-providers/native_kubernetes/#session-mode Best, Weihua On Mon, Mar 6, 2023

FlinkCEP GroupPattern 匹配结果异常问题

各位好， 我发现 FlinkCEP 中涉及 GroupPattern 的一些查询的匹配结果和直觉不太符合。 以如下查询为例： Pattern.begin("pl").where( new SimpleCondition() { @Override public boolean filter(Event value) { return value.getName() == 2; } }

Re: Flink作业tm Connection timed out异常问题

Hi 很多原因都可能会导致连接失败问题，包括机器故障、系统问题或者服务器负载，如果是怀疑负载问题你可以找几台服务器和这台有疑问的服务器组成个小集群，提交一些作业，让这台服务器负载不要太高，观察一下作业运行情况 Best, Shammon On Mon, Mar 6, 2023 at 8:49 PM crazy <2463829...@qq.com.invalid> wrote: > 报错日志下面这个一样，是同一个问题么 > https://issues.apache.org/jira/browse/FLINK-19925 > > > 其中描述到服务器 "high cpu usage

Re: flinkSQL无法实现两个timestamp(3) 相减

Hi 如果没有现成的系统函数，你可以写个自定义udf来实现 https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/functions/udfs/ Best, Shammon On Mon, Mar 6, 2023 at 7:46 PM 唐世伟 wrote: > > 我们需要对两个时间相减，精度为毫秒。但是无论是TIMESTAMPDIFF函数，还是先转成UNIX_TIMESTAMP，都只支持秒的精度。请问还有其他方法吗？

flink cdc connector计划支持hudi change data capture吗？

flink cdc connector计划支持hudi change data capture吗？

flink on K8S(operator) 如何获取 Accumulator

Hi,大佬们 如何在任务结束后获取Accumulator 数据？ 参考代码：（但是无法获取） ableResult execute = statementSet.execute(); Optional jobClient = execute.getJobClient(); jobClient.get().getAccumulators().get() PS: 最初的需求是： 对任务同步的数据量做统计。希望在批任务结束后，准确的获取Accumulator 中值，但是在K8S 中无法获取？ 大佬求助！！先磕为敬

flinkSQL无法实现两个timestamp(3) 相减

我们需要对两个时间相减，精度为毫秒。但是无论是TIMESTAMPDIFF函数，还是先转成UNIX_TIMESTAMP，都只支持秒的精度。请问还有其他方法吗？

Re: Flink作业tm Connection timed out异常问题

不建议这样做，因为这样会掩盖问题。 但如果一定要配置"重试次数"或"超时时长" 这些参数，会涉及到很多参数，比如 akka.tcp.timeout, taskmanager.network.netty.client.connectTimeoutSec, taskmanager.network.retries等等，具体可以参考[1]。 [1] https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/deployment/config/ Best, Yuxin crazy

Re: Flink作业tm Connection timed out异常问题

"如果进程没被调度到这台机器上，任务正常"，从给出的描述来看，确实很可能是 A 这台机器有问题。 可以检查机器 A 的网络、内存、CPU 指标或者监控是否正常，与其他机器是否存在不同。比如网络参数的配置、机器内存是否存在损坏、机器是否存在异常进程或负载等等。 如果硬件问题，系统日志有可能有一些报错。也可以使用一些机器检查工具, dmesg/vmstat等。 Best, Yuxin crazy <2463829...@qq.com.invalid> 于2023年3月6日周一 14:23写道： > 各位大佬好，有个线上作业频繁failover，异常日志如下： > >

Flink????tm Connection timed out????????

failover 2023-03-05 11:41:07,847 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Process (287/300) (b3ef27fec49fe3777f830802ef3501e9) switched from RUNNING to FAILED on container_e26_1646120234560_82135_01_97 @

Re:Re:Re: Re: 使用flinksql将kafka的数据同步mysql，在源kafka进行删除操作，mysql收到op为d的数据进行删除操作但是与此同时会新增一条当前数据以前的旧数据。

刚做了一下测试 目前假定有3行数据需要同步(全量)： | 编号 | 电话 | 座机 | | 1 | 1311313 | 123 | | 2 | 1311313 | 456 | | 3 | 1311313 | 789 | 这个时候我修改第四行数据的两个字段(增量): | 1 | 电话 | 座机 | | 1 | 1311313 | 123 | | 2 | 1311313 | 456 | | 3 | 13113133110 | 888 | 修改完后我删除字段2这个时候去mysql看结果2是正确被删除，且无新增的（操作正确）.

Re:Re: Re: 使用flinksql将kafka的数据同步mysql，在源kafka进行删除操作，mysql收到op为d的数据进行删除操作但是与此同时会新增一条当前数据以前的旧数据。

hi 早上好 我将flink升级到了1.16.1的版本去执行kafka同步到mysql的任务，发现还是存在一样的问题，我本机执行了explain的执行过程给的输出如下 == Abstract Syntax Tree == LogicalSink(table=[default_catalog.default_database.电话_1], fields=[rowID, 名称, 手机, 座机]) +- LogicalProject(rowID=[CAST($0):VARCHAR(255) CHARACTER SET "UTF-16LE"], 名称=[$1],

Re: Re: 使用flinksql将kafka的数据同步mysql，在源kafka进行删除操作，mysql收到op为d的数据进行删除操作但是与此同时会新增一条当前数据以前的旧数据。

Hi, 抱歉, 这里 typo 了, 应该是 1.16.1. 我在 1.16.1 上验证了你之前发的 query, 是可以正常删除的. 可以在 1.16.1 上尝试下, 也可以试试在 1.15.2 上使用 EXPLAIN CHANGELOG_MODE INSERT INTO...[1] 将 plan 打印出来看看. [1] https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/sql/explain/ 祝好! Jane On Sun, Mar 5, 2023 at 2:36 PM 陈佳豪

Re:Re: 使用flinksql将kafka的数据同步mysql，在源kafka进行删除操作，mysql收到op为d的数据进行删除操作但是与此同时会新增一条当前数据以前的旧数据。

hi 你好 目前没有1.16.2版本的吧？ 我看flink官网都是1.16.0 或者是1.16.1的 在 2023-03-02 11:52:41，"Jane Chan" 写道： >Hi, > >可以尝试用 EXPLAIN CHANGELOG_MODE[1] 把 plan 打出来看看, 或者尝试升级到 FLINK 1.16 版本, 这个 >query 在 1.16.2 上验证没有问题 > >[1]

Re: flink sql

好滴 谢谢大佬呀 | | 小昌同学 | | ccc0606fight...@163.com | Replied Message | From | 17610775726<17610775...@163.com> | | Date | 3/3/2023 15:55 | | To | user-zh@flink.apache.org | | Cc | user-zh | | Subject | Re:flink sql | Hi 可以通过设置 pipeline.operator-chaining = false 来实现。 Best JasonLee

Re:flink sql

Hi 可以通过设置 pipeline.operator-chaining = false 来实现。 Best JasonLee Replied Message | From | 小昌同学 | | Date | 03/3/2023 15:50 | | To | user-zh | | Subject | flink sql | 各位大佬，请教一下如何使用flink sql实现DataStreaming的disableOperatorChaining功能 | | 小昌同学 | | ccc0606fight...@163.com |

flink sql

各位大佬，请教一下如何使用flink sql实现DataStreaming的disableOperatorChaining功能 | | 小昌同学 | | ccc0606fight...@163.com |

回复： Flink内存问题

感谢，我看下 | | 吴先生 | | 15951914...@163.com | 回复的原邮件 | 发件人 | Weihua Hu | | 发送日期 | 2023年3月3日 10:37 | | 收件人 | | | 主题 | Re: Flink内存问题 | Hi, 针对问题 2， 可以增加下列环境变量来排除 Glibc 的问题，详情可以参考[1] containerized.master.env.MALLOC_ARENA_MAX: 1 containerized.taskmanager.env.MALLOC_ARENA_MAX: 1 [1]

Re:flink sql jdbc connector是否支持多流拼接？

这种情况下有两种方式可以处理 1> 注册表-使用join方式直接拼接成大宽表写入 2> 每个任务-直接写入下游数据 ，每个任务只更新自己的字段即可(因为主键相同) 在 2023-03-02 20:59:59，"casel.chen" 写道： >flink sql jdbc connector是否支持多流拼接？ >业务上存在基于主键打宽场景，即多个流有相同的主键字段，除此之前其他字段没有重叠，现在想打成一张大宽表写入mysql/oracle这些关系数据库。 >每条流更新大宽表的一部分字段。

Re: flink sql接cdc数据源关联维表写入下游数据库发现漏数据

听上去像是数据乱序了。可以看看这个文档对应的解决下[1] [1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/concepts/determinism/ Best, Shengkai casel.chen 于2023年3月1日周三 16:18写道： > flink sql上游接kafka canal json topic消费mysql同步过来的变更，接着关联几张维表，最后写入下游数据库发现漏数据。 > >

Re: Flink内存问题

Hi, 针对问题 2， 可以增加下列环境变量来排除 Glibc 的问题，详情可以参考[1] containerized.master.env.MALLOC_ARENA_MAX: 1 containerized.taskmanager.env.MALLOC_ARENA_MAX: 1 [1] https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/memory/mem_trouble/ Best, Weihua On Thu, Mar 2, 2023 at 8:10 PM 吴先生

Re: flink sql jdbc connector是否支持多流拼接？

hi. 手动使用 join 将多个流拼接起来？ Best, Shengkai casel.chen 于2023年3月2日周四 21:01写道： > flink sql jdbc connector是否支持多流拼接？ > 业务上存在基于主键打宽场景，即多个流有相同的主键字段，除此之前其他字段没有重叠，现在想打成一张大宽表写入mysql/oracle这些关系数据库。 > 每条流更新大宽表的一部分字段。

Re: 退订

Hi，退订请发送邮件到 user-zh-unsubscr...@flink.apache.org Best, Vincent Woo > 2023年3月3日 08:41，zhangjunjie 写道： > > 退订 > >

退订

退订

flink sql jdbc connector是否支持多流拼接？

flink sql jdbc connector是否支持多流拼接？ 业务上存在基于主键打宽场景，即多个流有相同的主键字段，除此之前其他字段没有重叠，现在想打成一张大宽表写入mysql/oracle这些关系数据库。 每条流更新大宽表的一部分字段。

回复： Flink内存问题

Hi, 目前分析问题应该在堆外，大概率是managed和overhead这两部分，这两部分的内存分配比例都是默认配置，通过网上的相关资料来看有两种解决方案： 1、调大managed和overhead这两块的内存比例， 问题：调整多大合适？是否调整之后还会持续增长 2、还有另一种说法是glibc内存分配器有个64M的问题引起（这里可有深入研究），替换为jemalloc可避免 问题：有具体的知道方案吗 | | 吴先生 | | 15951914...@163.com | 回复的原邮件 | 发件人 | Shammon FY | | 发送日期 | 2023年3月2日

Re: Flink内存问题

Hi 如果有搜集metrics，可以根据metrics查看一下是哪部分内存上涨导致container被kill掉；然后将上涨比较快的container内存dump一下，查看具体是哪些对象占用内存比较多 Best, Shammon On Thu, Mar 2, 2023 at 7:14 PM 吴先生 <15951914...@163.com> wrote: > Hi， > Flink版本：1.12 > 部署模式：on yarn per-job > 开发方式：DataStream Api > 状态后端：RocksDB >

Flink内存问题

Hi， Flink版本：1.12 部署模式：on yarn per-job 开发方式：DataStream Api 状态后端：RocksDB Job逻辑为一个15分钟的窗口计算，任务在运行一段时间后会出现内存使用超限，container被yarn kill的现象，目前有不少任务都会存在类似问题： Closing TaskExecutor connection container_e02_1654567136606_1034_01_12 because: [2023-03-02 08:12:44.794]Container

Re: 使用flinksql将kafka的数据同步mysql，在源kafka进行删除操作，mysql收到op为d的数据进行删除操作但是与此同时会新增一条当前数据以前的旧数据。

Hi, 可以尝试用 EXPLAIN CHANGELOG_MODE[1] 把 plan 打出来看看, 或者尝试升级到 FLINK 1.16 版本, 这个 query 在 1.16.2 上验证没有问题 [1] https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/sql/explain/ Best, Jane On Wed, Mar 1, 2023 at 6:22 PM 陈佳豪 wrote: > flink ，kafka连接 jdbc连接版本都是1.15.2的 > > > > > > >

Re: Re: Flink SQL 如何优化以及处理反压

可以看一下反压算子是否出现在同一台机器（排除单点故障）。比如使用了 rocksdb + hdd 盘；单机负载过高；磁盘打满等。 如果不是单点故障，可以打 jstack 查看对应的线程具体在执行什么样的操作，再进行相应的逻辑优化。 On Tue, Jan 31, 2023 at 6:01 PM lxk wrote: > 现在从web ui上看，瓶颈主要在于group by 聚合函数之后去重这个逻辑。 > 而且SQL这个并行度是全局设置的，没法针对某一个特定的算子设置并行度，并行度多了之后，资源又感觉有点吃紧。 > > > > > > > > > > > > > > > > > > 在

Re:使用flinksql将kafka的数据同步mysql，在源kafka进行删除操作，mysql收到op为d的数据进行删除操作但是与此同时会新增一条当前数据以前的旧数据。

flink ，kafka连接 jdbc连接版本都是1.15.2的 在 2023-03-01 18:14:35，"陈佳豪" 写道： >问题如标题所示，就是删除操作的时候mysql的表数据不对，每次都会新增当前主键的旧数据。 >String kafka = "CREATE TABLE `电话` (`rowid` >VARCHAR(2147483647),`63fd65fb36521f81a2cfab90` >VARCHAR(2147483647),`63fd660536521f81a2cfabad`

使用flinksql将kafka的数据同步mysql，在源kafka进行删除操作，mysql收到op为d的数据进行删除操作但是与此同时会新增一条当前数据以前的旧数据。

问题如标题所示，就是删除操作的时候mysql的表数据不对，每次都会新增当前主键的旧数据。 String kafka = "CREATE TABLE `电话` (`rowid` VARCHAR(2147483647),`63fd65fb36521f81a2cfab90` VARCHAR(2147483647),`63fd660536521f81a2cfabad` VARCHAR(65535),`63fd660536521f81a2cfabae` VARCHAR(65535) ) WITH ( 'connector' = 'kafka', 'topic' =

metrics.latency.interval指标应该如何查看？

我使用的flink版本是1.14.0，在flink-conf.yaml里添加了latency的配置 但是我在web-ui中没有找到这个指标 请问下是哪里出问题了。 flink任务是从kafka中读数据写入mysql中 public class FlinkSqlTask { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env =

flink sql接cdc数据源关联维表写入下游数据库发现漏数据

flink sql上游接kafka canal json topic消费mysql同步过来的变更，接着关联几张维表，最后写入下游数据库发现漏数据。 随后在写目标数据库中加了一些日志后发现同一主键的变更记录(前后发生间隔时间很短）被发送到了不同的TaskManager处理，导致新数据被旧数据覆盖，造成漏数据现象。 请问： 1. cdc数据源关联维表后会被分散到不同TaskManager吗？什么情况下会发生？ 2. 如何解决这个问题？是需要在写目标表之前加一层窗口去重[1]吗？ [1]

退订

退订

Re: managed memory占用100%的问题

Hi 根据邮件里的异常信息看了下代码，这里的RecordArea会从managed memory申请内存分片 你可以根据作业流量尝试调整下窗口大小或者内存分配 Best, Shammon On Tue, Feb 28, 2023 at 6:47 PM Junrui Lee wrote: > Hi, > > 图片挂掉了，能不能直接用文字描述配置文件？ > > Best, > Junrui > > 生于八十年代 <623730...@qq.com.invalid> 于2023年2月28日周二 18:31写道： > > > 社区的各位大佬们有个问题咨询一下大家： > > 1. > >

Re: managed memory占用100%的问题

Hi, 图片挂掉了，能不能直接用文字描述配置文件？ Best, Junrui 生于八十年代 <623730...@qq.com.invalid> 于2023年2月28日周二 18:31写道： > 社区的各位大佬们有个问题咨询一下大家： > 1. > 问题背景：我们在使用flink读取消费kafka中的hdfs路径消息，然后从hdfs中读取文件，做完处理后写入hive，整个过程都是以流式的过程完成，而不是批处理的过程； > 目前遇到的问题是任务运行一段时间之后，kafka就开始出现hdfs路径消息积压，目前发现managed >

managed memory????100%??????

??1. flinkkafkahdfshdfs??hive ??kafka??hdfs??managed

Re:Flink Kafka Sink时间戳异常

hi， 这个问题是因为经过窗口算子后StreamRecord中指定的时间时间戳被改成了window.maxTimestamp()，可以查看[1]中WindowOperator或EvictingWindowOperator中的emitWindowContents方法。 如果想要更改时间戳，可以实现一个ProcessFuncton TimestampedCollector collector = (TimestampedCollector) out; collector.setAbsoluteTimestamp( ); collector.collect(value);

Re: 退订

退订请发送邮件到 user-zh-unsubscr...@flink.apache.org Best, Weihua On Tue, Feb 28, 2023 at 12:13 AM zhangjunjie wrote: > 退订 > > >

退订

退订

Re: 【Windowing TVF】 GROUP BY window_start, window_end 没有输出

Hi, 可以详细描述下你的使用 case 吗？用的 SQL 语句是什么样子的 Best, Weihua On Mon, Feb 27, 2023 at 12:51 PM wei_yuze wrote: > 您好！ > > > > > 我在使用Windowing table-valued functions (Windowing TVFs) 的时候，GROUP BY 中一旦加上 > window_start, window_end 就没有输出，但也不报错。请问有哪位大佬知道是什么原因吗？ > > Lucas

退订

????

| ?? | fei<704669...@qq.com.INVALID> | | | 2023??02??26?? 09:02 | | ?? | user-zh | | ?? | | | | |

Re:

退订请发送邮件至 user-zh-unsubscr...@flink.apache.org Best, Jane On Fri, Feb 24, 2023 at 7:43 PM LITA LITA wrote: > 退订 > > <704669...@qq.com.invalid> 于2023年2月24日周五 07:58写道： > > > 退订 > > > > >

Re: 使用flink sql 将kafka的数据同步到mysql无法删除。

Hi, 原问题中 String 变量 kafka 和 mysql 赋值反了, 以及能提供下所使用的 flink 版本吗, 我使用 1.16.1 没有复现此问题 payload { "before": { "rowid": "f251af39-1a95-4d6f-b4cb-cdf93d5d1b6d", "63f73b332e77497da91286f0": "Jerry", "63f73b3f2e77497da91286fb": "mobile number", "63f73b3f2e77497da91286fc": "telephone

Re:

退订 <704669...@qq.com.invalid> 于2023年2月24日周五 07:58写道： > 退订 > >

Re: flink avro schema 升级变动，job如何平滑过渡

Hi 你可以贴一下错误看下具体原因 Best, Shammon On Fri, Feb 24, 2023 at 6:10 PM Peihui He wrote: > Hi, all > > 请教大家有没有遇到这样的情况，flink 使用avro > 消费kafka中数据，后来在schema结构中加入新的字段。在灰度过程中会混杂着新老数据，这样的flink 消费程序就会挂掉。 > > 大家一般是怎么处理的呢 > > Best Wishes. >

flink avro schema 升级变动，job如何平滑过渡

Hi, all 请教大家有没有遇到这样的情况，flink 使用avro 消费kafka中数据，后来在schema结构中加入新的字段。在灰度过程中会混杂着新老数据，这样的flink 消费程序就会挂掉。 大家一般是怎么处理的呢 Best Wishes.

Re: 在计算Window Top-N时，Flink SQL 时间语义不生效

更乱了哦...可以尝试加个附件或推到 github, 贴个链接 On Fri, Feb 24, 2023 at 4:59 PM wei_yuze wrote: > >

在计算Window Top-N时，Flink SQL 时间语义不生效

使用flink sql 将kafka的数据同步到mysql无法删除。

-建表语法如下 String kafka = "CREATE TABLE `电话` " + "(`rowID` VARCHAR(255),`名称` STRING,`手机` VARCHAR(255),`座机` VARCHAR(255), " + " PRIMARY KEY (`rowID`) NOT ENFORCED ) " + " WITH " + "('connector' = 'jdbc', " + " 'driver' = 'com.mysql.cj.jdbc.Driver', " + " 'url' =

Re: flink taskmanger重启失败的问题

从 region 改为 full 会扩容单个 Task 故障的影响范围，可以参考社区文档： https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/ops/state/task_failure_recovery/ Best, Weihua On Fri, Feb 24, 2023 at 2:12 PM 唐世伟 wrote: > 谢谢回复，我看日志已经超出来yarn保存的期限被删了。另外Failover从region改为full。是不是能避免这个问题啊？ > > > 2023年2月23日

Re: flink taskmanger重启失败的问题

谢谢回复，我看日志已经超出来yarn保存的期限被删了。另外Failover从region改为full。是不是能避免这个问题啊？ > 2023年2月23日 上午11:36，Weihua Hu 写道： > > Hi, > > 在 Cancel 其他 task 时会先将 task 状态置为 cancelling，这时 task 失败是不会二次触发 Failover 的。 > 可以检查下是不是作业划分了多个 region，多个 region 的异常是统一计数的。 > > 或者可以贴一下日志吗？ > > > Best, > Weihua > > > On Thu, Feb 23,

Re: Re: Re: [急] flink sql写数据到带主键表时能确保相同主键的记录都被同一个taskmanager处理吗？

> 你说的这个参数我看了默认值不是auto吗？需要我显式地指定为force? Sink upsert materialize would be applied in the following circumstances: 1. `TABLE_EXEC_SINK_UPSERT_MATERIALIZE` set to FORCE and sink's primary key nonempty. 2. `TABLE_EXEC_SINK_UPSERT_MATERIALIZE` set to AUTO and sink's primary key doesn't contain upsert

Re:Re: Re: [急] flink sql写数据到带主键表时能确保相同主键的记录都被同一个taskmanager处理吗？

你说的这个参数我看了默认值不是auto吗？需要我显式地指定为force? Because of the disorder of ChangeLog data caused by Shuffle in distributed system, the data received by Sink may not be the order of global upsert. So add upsert materialize operator before upsert sink. It receives the upstream changelog records and

答复:

退订

[ANNOUNCE] Apache Flink Kubernetes Operator 1.4.0 released

The Apache Flink community is very happy to announce the release of Apache Flink Kubernetes Operator 1.4.0. The Flink Kubernetes Operator allows users to manage their Apache Flink applications and their lifecycle through native k8s tooling like kubectl. Release highlights: - Flink Job

Re: 退订

退订请发送邮件到 user-zh-unsubscr...@flink.apache.org Best, Shammon On Thu, Feb 23, 2023 at 11:03 AM 宋品如 wrote: > 退订

Re: flink taskmanger重启失败的问题

Hi, 在 Cancel 其他 task 时会先将 task 状态置为 cancelling，这时 task 失败是不会二次触发 Failover 的。 可以检查下是不是作业划分了多个 region，多个 region 的异常是统一计数的。 或者可以贴一下日志吗？ Best, Weihua On Thu, Feb 23, 2023 at 11:16 AM 唐世伟 wrote: > 我们有一个flink任务，同时写10几张doris表，每次doris出问题的时候任务就挂，flink的重启策略没有效果。 > flink的重启配置入下： > restart-strategy:

flink taskmanger重启失败的问题

我们有一个flink任务，同时写10几张doris表，每次doris出问题的时候任务就挂，flink的重启策略没有效果。 flink的重启配置入下： restart-strategy: failure-rate restart-strategy.failure-rate.delay: 60 s restart-strategy.failure-rate.failure-rate-interval: 10 min restart-strategy.failure-rate.max-failures-per-interval: 3

退订

退订

退订

退订

Re: FlinkSql如何实现水位线对齐

Hi 目前SQL还不支持watermark对齐，目前有FLIP正在讨论中 https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=240884405 Best, Shammon On Wed, Feb 22, 2023 at 3:15 PM haishui wrote: > Hi, all > 以并行度4读取kafka的topic1和topic2形成两个流，然后IntervalJoin。在kafka堆积大量数据的情况下，我分别用SQL和DataStream > API实现了上述功能。 > > >

Re: Re: [急] flink sql写数据到带主键表时能确保相同主键的记录都被同一个taskmanager处理吗？

Hi, Re *"如何确保同一主键变更记录(上游接的是cdc数据源)的时间先后顺序性呢？", *checking out ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE for details about solution of disordering problems in KeyBy shuffling. Best, Shuo On Wed, Feb 22, 2023 at 10:23 AM casel.chen wrote: > >

FlinkSql如何实现水位线对齐

Hi, all 以并行度4读取kafka的topic1和topic2形成两个流，然后IntervalJoin。在kafka堆积大量数据的情况下，我分别用SQL和DataStream API实现了上述功能。 使用SQL实现的作业中IntervalJoin算子的状态会逐渐增大，直到checkpoint失败。原因是在8个Source分区中输出水位线差距很大。 使用API实现的作业，在使用Flink15版本的水位线对齐后可以保证正常读取topic内的所有数据。

Re: Re: [急] flink sql写数据到带主键表时能确保相同主键的记录都被同一个taskmanager处理吗？

如果想保证每次写入 mysql 的事件是最新的，需要在 Flink 内部针对事件时间排序取 TOP 1, 可以参考[1]。 但是需要注意这需要使用 state，你可以需要指定合适的 TTL[2] 来保证 state 不会过大 [1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/queries/topn/ [2]

Re: 退订

退订请发送邮件到 user-zh-unsubscr...@flink.apache.org Best regards, Weijie 宋品如 于2023年2月22日周三 11:37写道： > 退订 > > > > > > > > > > > -- > > 祝工作顺利，生活愉快！ > 发件人：宋品如 > 岗位：大数据开发

Re: 退订

退订请发送邮件到 user-zh-unsubscr...@flink.apache.org Best regards, Weijie 646208563 于2023年2月22日周三 11:39写道： > 退订

Re:Re: [急] flink sql写数据到带主键表时能确保相同主键的记录都被同一个taskmanager处理吗？

如果像楼上所说[1]主动keyby将同一主键记录汇聚到同一个TM处理的话，Flink如何确保同一主键变更记录(上游接的是cdc数据源)的时间先后顺序性呢？ 在 2023-02-20 09:50:50，"Shengkai Fang" 写道： >我理解如果sink 表上带有 pk，就会主动 keyby 的[1]。 > >Best, >Shengkai > >[1]

Re:Re: Re: Flink程序内存Dump不了

非常感谢，切换账户之后，有些job是能dump下来，但是有些job还是会报相同的错误，并且已经确认过flink作业的user和运行linux命令的user是一致的。 在 2023-02-21 11:26:07，"Biao Geng" 写道： >Hi, >这个报错 sun.jvm.hotspot.debugger.UnmappedAddressException: 7f74efa5d410 >可能和flink关系不大。 >我们之前在生产环境中dump内存的时候遇到过类似问题，后续定位发现是运行jmap命令的Linux

Re: Whether Flink SQL window operations support "Allow Lateness and SideOutput"?

Hi, Maybe you can use CURRENT_WATERMARK()[1] to handle some late data. [1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/functions/systemfunctions/ Best, Weihua On Tue, Feb 21, 2023 at 1:46 PM wang <24248...@163.com> wrote: > Hi dear engineers, > > One question as

Whether Flink SQL window operations support "Allow Lateness and SideOutput"?

Hi dear engineers, One question as title: Whether Flink SQL window operations support "Allow Lateness and SideOutput"? Just as supported in Datastream api (allowedLateness and sideOutputLateData) like: SingleOutputStreamOperator<>sumStream = dataStream.keyBy().timeWindow()

Re: 广播流与非广播流 数据先后问题

Hi, 可以看到你的问题，你需要订阅 user-zh 的邮件列表才能收到相关的回复，可以参考： https://flink.apache.org/community.html Best, Weihua On Tue, Feb 21, 2023 at 12:17 PM 知而不惑 wrote: > 有收到我的问题吗 > > > > > --原始邮件-- > 发件人: > "user-zh" > >

Re: Re: Flink程序内存Dump不了

Hi, 这个报错 sun.jvm.hotspot.debugger.UnmappedAddressException: 7f74efa5d410 可能和flink关系不大。 我们之前在生产环境中dump内存的时候遇到过类似问题，后续定位发现是运行jmap命令的Linux user与运行flink作业的Linux user不一致导致的。 不知道和你们的问题一不一致，你可以试试t用op -p 查到USER，然后su 一下，再进行jmap -dump:format=b,file=/tmp/dump.hprof 试试。 Best, Biao Geng Weihua Hu

Re: 广播流与非广播流 数据先后问题

Hi, 可以把具体的报错信息贴一下，另外代码中没有看到使用 listState 缓存元素的部分 Best, Weihua On Tue, Feb 21, 2023 at 9:38 AM 知而不惑 wrote: > 各位大佬好 > 我使用广播流与非广播流进行connet，随后继承实现了BroadcastProcessFunction，运行时发现在自定义实现的BroadcastProcessFunction > 中，广播流数据会先到，导致processElement() 中获取广播流数据为空，请问有什么写法或机制解决该问题？我尝试在谷歌和chatgpt > 找寻答案，给到的回复是

flink cep A B C 事件一段时间不分先后顺序匹配

hi, all 如题，看了https://mp.weixin.qq.com/s/PT8ImeOOheXR295gQRsN8w 这篇文章后，发现第四个问题没有讲到解决方案。 请教大家有什么好的方案没呢？ Best Wishes!

Re: Re: flink canal json格式忽略不识别的type

Hi, 可以尝试使用： json.ignore-parse-errors[1] 来忽略解析的报错，需要注意这个参数会忽略所有的解析错误 [1] https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/formats/json/#json-ignore-parse-errors Best, Weihua On Mon, Feb 20, 2023 at 10:14 AM casel.chen wrote: > 日志中就是报这个 "type":"INIT_DDL"

Re: Re: Flink程序内存Dump不了

Hi, 可以把心跳超时(heartbeat.timeout)[1]也调大再尝试 dump 内存。 [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: > 我尝试调整了参数，具体数值如下 > > > akka.ask.timeout: 900s > > > > 但还是报同样的错 > > > >

Re:Re: Flink程序内存Dump不了

我尝试调整了参数，具体数值如下 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 >> 问题：最近上线的Flink程序，频繁young >>

Re:Re: Flink1.16写入kafka 报错：Cluster authorization failed.

我们改了权限确实解决了这个问题。但我现在想了解的是为什么Flink在1.16的时候需要创建producerID的权限，以及这个权限是不是针对新老Kafka API都需要的。针对新的Kafka API在精准一次的时候需要管理ProducerID在源码中有体现，但是老的API没看见相关的，只使用了一个ProducerID也需要由Flink内部自己管理吗？ 在 2023-02-20 08:45:18，"Shammon FY" 写道： >Hi > >从`Caused by:

Re:Re: [急] flink sql写数据到带主键表时能确保相同主键的记录都被同一个taskmanager处理吗？

Flink SQL作业示意如下： create table user_source_table ( id BIGINT NOT NULL PRIMARY KEY NOT ENFORCED, name STRING, dept_id BIGINT NOT NULL, proctime AS PROCTIME() ) with ( 'connector' = 'kafka', 'format' = 'canal-json', ... ); create table department_dim_table ( id BIGINT NOT NULL

Re:Re: flink canal json格式忽略不识别的type

日志中就是报这个 "type":"INIT_DDL" 不能识别呀，然后作业就退出了 在 2023-02-20 09:58:56，"Shengkai Fang" 写道： >Hi. 能同时分享下复现这个 case 的sql 以及相关的报错栈吗？ > >Best, >Shengkai > >casel.chen 于2023年2月9日周四 12:03写道： > >> 不同云厂商的数据同步工具对于全量+增量同步mysql数据到kafka canal json格式时行为不一致 >> 有的厂商会将DDL语句同步到topic导致下游flink

Re:Re:[急] flink sql写数据到带主键表时能确保相同主键的记录都被同一个taskmanager处理吗？

你说的这个在写入之前进行shuffle（先执行一个group by主键）这个操作我认为应该是Flink框架层面的事情，不应该在作业层面显式添加。 Flink框架应该在执行sink的时候判断目标表是否有主键，如果有主键的话应该插入一个group by算子将相同主键的记录发到同一个TaskManager处理。 我听说 Flink新版本1.15还是1.16不记得了已经改进了这个问题，有谁知道吗？有相关issue或PR链接没？ 在 2023-02-19 13:43:29，"RS" 写道： >Hi，

Re: flink canal json格式忽略不识别的type

Hi. 能同时分享下复现这个 case 的sql 以及相关的报错栈吗？ Best, Shengkai casel.chen 于2023年2月9日周四 12:03写道： > 不同云厂商的数据同步工具对于全量+增量同步mysql数据到kafka canal json格式时行为不一致 > 有的厂商会将DDL语句同步到topic导致下游flink sql作业不能识别抛错，建议flink canal > json格式解析时直接忽略不识别的type，例如 > 例1： >

Re: [急] flink sql写数据到带主键表时能确保相同主键的记录都被同一个taskmanager处理吗？

我理解如果sink 表上带有 pk，就会主动 keyby 的[1]。 Best, Shengkai [1] https://github.com/apache/flink/blob/master/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java#L188 Shammon FY 于2023年2月20日周一 08:41写道： > Hi > >

Re: Flink1.16写入kafka 报错：Cluster authorization failed.

Hi 从`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的时候报错： > 2023-02-17

Re: [急] flink sql写数据到带主键表时能确保相同主键的记录都被同一个taskmanager处理吗？

Hi 如果join计算的关联字段里没有主键，在将join结果直接写入sink表时，像上面RS提到的，需要自己再增加一次shuffle操作 Best, Shammon On Sun, Feb 19, 2023 at 1:43 PM RS wrote: > Hi， > connector里面配置的主键控制是写入存储的，有些存储在写入数据的时候，能根据主键自动更新去重 > 所以我感觉，你这里的应该是想在计算的时候shuffle（写入之前），你应该需要先执行一个 group by 主键，然后再执行insert into > > > Thanks > > > > 在 2023-02-17

Re:[急] flink sql写数据到带主键表时能确保相同主键的记录都被同一个taskmanager处理吗？

Hi， connector里面配置的主键控制是写入存储的，有些存储在写入数据的时候，能根据主键自动更新去重 所以我感觉，你这里的应该是想在计算的时候shuffle（写入之前），你应该需要先执行一个 group by 主键，然后再执行insert into Thanks 在 2023-02-17 15:56:51，"casel.chen" 写道： >作业场景是kafka cdc数据源关联几张redis维表再和其他流表进行双流regular inner >join，最后将打宽表写入mongodb。使用的是flink 1.13.2 sql模式。开了debug日志。

Flink1.16写入kafka 报错：Cluster authorization failed.

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

Re: Flink程序内存Dump不了

可能是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 > 问题：最近上线的Flink程序，频繁young > gc，几秒一次，在调整了新生代大小之后，还是没有解决，目前整个jvm堆大小是3.57g。因此想通过程序内存情况来分析哪里问题有问题，我们通过yarn上的applicationId,使用ps > -ef|grep

<    7   8   9   10   11   12   13   14   15   16   >