Checkpointing

Hello, I'm curious about how Flink checkpointing would aid in recovering data if the data source is not Kafka but another system. I understand that checkpoint snapshots are taken at regular time intervals. What happens to the data that were read after the previous successful checkpoint if the

Async code inside Flink Sink

Hello, I have a use case where I need to do a cache file deletion after a successful sunk operation(writing to db). My Flink pipeline is built using Java. I am contemplating using Java completableFuture.runasync() to perform the file deletion activity. I am wondering what issues this might cause

Global connection open and close

Hello, Is there a way in Flink to instantiate or open connections (to cache/db) at global level, so that it can be reused across many process functions rather than doing it in each operator's open()?Along with opening, also wanted to know if there is a way to close them at job level stop, such

Fwd: Flink Checkpoint & Offset Commit

Hello, I am implementing proof of concepts based Flink realtime streaming solutions. I came across below lines in out-of-the-box Flink Kafka connector documents. *https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/kafka/*

Re: [E] Re: Kubernetes operator expose UI rest service as NodePort instead of default clusterIP

I remember testing the operator with the rest service exposed as NodePort. NodePort requires rbac.nodeRoules.create: true (default is false) in values.yaml. Maybe you missed that?

Flink Kubernetes Operator v1.0 ETA

Hi there, Is there an ETA on v1.0 release of operator? We are prototyping with a CI build from release-1.0 branch but would like to know the approximate ETA of official 1.0 release so that we can plan accordingly. Thanks, Jeesmon

Re: Python Job Type Support in Flink Kubernetes Operator

Hi Gyula, Any idea on this? We are exploring current limitations of using the operator for Flink deployment and if there is a plan to support Python jobs in future will help us. Thanks, Jeesmon On Fri, May 20, 2022 at 3:46 PM Jeesmon Jacob wrote: > Hi there, > > Is there a plan t

Python Job Type Support in Flink Kubernetes Operator

Hi there, Is there a plan to support Python Job Type in Flink Kubernetes Operator? If yes, any ETA? According to this previous operator overview only Java jobs are supported in operator. This page was recently modified to remove the features table.

Re: Flink checkpoint 速度很慢 问题排查

// 处理逻辑 .. resultFuture.complete(Collections.singletonList(...)); }); } } - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink checkpoint 速度很慢 问题排查

ocs-release-1.13/zh/docs/dev/datastream/operators/asyncio/> - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink checkpoint 速度很慢 问题排查

都是简单的将List数据扁平化而已，没有什么耗时操作。开始它们的并行度我设25（topic01 partition数量、消息速率是：1000~2000条/s），后面直接改成80，但并没有明显效果。 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: 回复：Flink checkpoint 速度很慢 问题排查

@lian 谢谢回复 我通过webui查询，没有背压的情况。 hbase性能这块确实存在一定问题，公司的hbase性能一直不佳，但我的程序中，是先从缓存redis中取，大部分数据都能查到，只有少部分会查hbase。 谢谢你提供的思路，我将会从代码级别考虑，看是否能优化相关逻辑 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: 回复：Flink checkpoint 速度很慢 问题排查

@lian 谢谢回复 我通过webui查询，没有背压的情况。 hbase性能这块确实存在一定问题，公司的hbase性能一直不佳，但我的程序中，是先从缓存redis中取，大部分数据都能查到，只有少部分会查hbase。 谢谢你提供的思路，我将会从代码级别考虑，看是否能优化相关逻辑 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink checkpoint 速度很慢 问题排查

（keyby、max）之后写到orc file，state大小几百兆，但耗时是秒级别。 我比较疑惑的是为什么A job的state那么小，但checkpoint却很耗时，不知道从哪个角度去优化该问题。 请各位指教 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink 1.11.2 SQL消费kafka写Hive报错

问题已解决 需要在FLink home的lib中引入kafka connector jar包 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink 1.11.2 SQL消费kafka写Hive报错

Thank you for your reply! 您所说的kafka connector 是*flink-connector-kafka_2.11* 这个依赖吗？这个是Datastream所用的依赖，我在pom中已经引入了* flink-sql-connector-kafka_2.11*依赖了。 我试了引入* flink-connector-kafka_2.11*，但还是会报错的。 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink 1.11.2 SQL消费kafka写Hive报错

Thank you for your reply! 您所说的kafka connector 是* flink-connector-kafka_2.11* 这个依赖吗？这个是Datastream所用的依赖，我在pom中已经引入了 *flink-sql-connector-kafka_2.11*依赖了。 我试了引入* flink-connector-kafka_2.11*，但还是会报错的。 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink 1.11.2 SQL消费kafka写Hive报错

.runApplicationEntryPoint(ApplicationDispatcherBootstrap.java:230) ~[flink-dist_2.11-1.11.2.jar:1.11.2] ... 10 more Caused by: org.apache.flink.table.api.ValidationException: Could not find any factory for identifier 'kafka' that implements 'org.apache.flink.table.factories.DynamicTableSourceFactory' in the classpath. - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink 流批一体的实践

StreamExecutionEnvironment ，如果用Flink SQL去运行增量和全量的Join，那就是另外的TableEnvironment了，我不太清楚怎么在同一个Job去构建这两种环境，以及，Join后要写入几张不同的hive表，必然会运行多个sql语句，我看到的一些demo 都是一次跑一个sql语句，那么这种需求是可以实现的吗？ - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Exception handling

recently encountered some setbacks in the game of whack-a-mole with pathological messages, and are hoping to mitigate the losses when these do occur. Jacob

Flink job消费kafka 失败，无法拿到offset值

(Task.java:530) at java.lang.Thread.run(Thread.java:748) Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout of 6ms expired before the position for partition Test-topic-27 could be determined 查询一圈发现基本都是说slot不够之类的原因，已经kafka broker负载等问题，这些问题已经排除。 请指教 - Thanks! Jacob -- Sent

Flink 写ORC失败

体是哪个地方的错误 看到一个相似的错误 https://stackoverflow.com/questions/55246512/error-writing-to-orcnewoutputformat-using-mapr-multipleoutputs 不太清楚这个错误时什么原因所致 已知数据不为空，不为null - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink 消费kafka ，写ORC文件

ext.timestamp(); } return null; } - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink 消费kafka ，写ORC文件

cts/flink/flink-docs-release-1.12/zh/dev/connectors/file_sink.html#%E6%A1%B6%E5%88%86%E9%85%8D - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink Job 如何集成到自己的系统，方便管理

谢谢大佬答疑。 我先尝试使用 YarnClusterDescriptor 这些类提交Job。看看后续使用情况 是否合适 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink Job 如何集成到自己的系统，方便管理

谢谢提供思路，刚通过接口编程这个思路找到了一些文章和demo。 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink Job 如何集成到自己的系统，方便管理

这些页面以超链接的形式集成到系统里面吗，在系统dashboard中点某个按钮，跳转到flink webui的某一个模块里？ - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink Job 如何集成到自己的系统，方便管理

谢 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: 回复：Flink checkpoint 速度慢问题请教

谢谢回复 看了数次checkpoint慢的情况，发现大多是async阶段耗时，如果是这样，那这应该是那个时刻网络原因导致的慢吧？ 但是我还是觉得跟磁盘有一定关系 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: 回复：Flink checkpoint 速度慢问题请教

state.checkpoints.num-retained: 3 restart-strategy: fixed-delay restart-strategy.fixed-delay.attempts: 1000 restart-strategy.fixed-delay.delay: 30 s 后面把上面配置注释掉，然后在代码中指定了checkpoint类型为内存，但速度还是很慢。 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink checkpoint 速度慢问题请教

两个hadoop集群唯一不同的是，A集群是SSD，B集群机器是机械硬盘。 job的checkpoint存储是在hdfs，是否是因为磁盘性能问题，导致B集群checkpoint速度过慢呢？是否应该使用内存作为checkpoint存储？请指教。 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink job与自己系统平台的一体化集成

的平台上提交job，那是不是应该先把flink客户端先集成到我们的系统中呢，否则job如何被启动运行呢？ 需求比较模糊，各位大佬见谅。 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Re:Flink 并行度问题

我明白了。。。感谢！！（最近很容易陷入这些低级的错误认知中）。越想越觉得这个问题很傻 23 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Re:Flink 并行度问题

谢谢回复~ 在我的理解中，并行度数量不应该超过CPU的数量的。 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink 并行度问题

？ - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink on yarn JDK 版本支持问题

home路径？ 这两个jdk对于启动一个flink job的作用是什么呀？（ 配置的env.java.home和hadoop集群的java home） - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink 并行度问题

，但在上面的理解中，好像是一个CPU就是一个线程，这是否矛盾呢？ - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink SQL 如何保证多个sql 语句按顺序执行

谢谢回复， 听起来是可以的 我先尝试一下这种方案 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink 1.10 on yarn提交任务报错

hi， 可以先做如下尝试： export HADOOP_USER_NAME=your user export HADOOP_CONF_DIR=your hadoop conf dir export HADOOP_CLASSPATH=`/opt/app/hadoop/bin/hadoop classpath` - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink sql 流批一体的应用

settings = EnvironmentSettings.newInstance().inBatchMode().build(); TableEnvironment tableEnv = TableEnvironment.create(settings); 有没有可能让这两部分合二为一呢，放在同一个job执行？ - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink SQL 如何保证多个sql 语句按顺序执行

Dear All,在Flink SQL job中,如果有多个sql语句，需要按顺序执行，即下一个sql的执行依赖上一个sql的执行结果。由于tableEnv.executeSql(sql)是*异步*提交的，那么如何保证多个sql是*顺序执行*？eg:在一个main函数中，有如下代码：String sql1 = "";tableEnv.executeSql(sql1 );String sql2 = "";tableEnv.executeSql(sql2 );问题：如何保证sql1先执行完成，再执行sql2 - Thanks! Jaco

Re: Flink SQL>查询的hive表数据全部为NULL

谢谢回复 这个问题困扰了很久 已经解决 原因是写orc时候指定的字段名是column0、column1.、column33 而hive创建表的字段是实际字段的名字，两个不匹配，因此在flink sql中读不到 数据 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink SQL查询ORC表结果全部为NULL

/null.png> Flink SQL> select count(*) from `table1` where dt='1609739880002'; <http://apache-flink.147419.n8.nabble.com/file/t1162/count.png> - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink SQL>查询的hive表数据全部为NULL

Dear All， Flink SQL>select * from table1； 在Flink客户端查询hive数据，结果每个字段全为NULL，但数据条数是对的，select count是正确的，查具体数据就是NULL，不知何故？同样的查询在hive客户端是可以查到数据的 hive表时orc文件load的数据。 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink sql执行insert into 的一些问题

Dear All， Flink SQL> insert into table1 select filed1,filed2,.filed10 from table2; 在flink sql 中执行类似上面的语句，在webui中看到很快就finished了，但数据并没有写进table1表中，查看log，也看不到什么报错。迷惑 还有，在使用select count(*) 查询表数据时，有的表能查到结果，有的就不显示结果，也没有报错。实在不知道什么原因了。。。 - Thanks! Jacob -- Sent from: http://apache-fl

Flink 1.11.2 替代MapReduce执行 Hive 语句 Full Join 非常慢

ig().addConfiguration(new Configuration().set(CoreOptions.DEFAULT_PARALLELISM, 8)); tableEnv.executeSql(HQL ); } } <http://apache-flink.147419.n8.nabble.com/file/t1162/hiveonFlink.png> - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink 1.11.2客户端 select * from table

Dear All, <http://apache-flink.147419.n8.nabble.com/file/t1162/FlinkSQL.png> 在Flink SQL客户端中使用select * from table 语句查询表数据，结果只显示表头，没有数据，是什么原因。 <http://apache-flink.147419.n8.nabble.com/file/t1162/queryTable.png> 在sql-client-defaults.yaml中已经指定catalog ----- Thanks! Jacob -- Sen

Re: Flink 操作hive 一些疑问

Thanks! 还是决定暂时用两个job执行吧 一个job执行流处理生成数据 另一个job（使用flink的hive功能）执行批处理。 主要验证一下第二个job的使用相比MapReduce节省了多少资源 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink 操作hive 一些疑问

Hi, 谢谢回复 对，也可以这样理解，总体分为两部分，先处理流消息，每隔15min写进hive表。然后再做mapreduce处理上步15min的数据。 目前的现状是： 第一步用flink处理，第二步是一个定时job去处理上一步的数据。 改善计划： 想整合这两步，都使用flin处理，flink新版本对hive有支持，就不用再使用MapReduce了，现在就是不知道怎样平滑地在同一个Job中执行。 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink 操作hive 一些疑问

e(settings); .. tableEnv.executeSql(hql); 有没有什么通用的方案，让这两个job合二为一呢？我想要的效果时，当生成完业务数据后，直接操作hive，取代mapreduce的工作。 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink 连接Hive hiveConf路径配置

ion: /etc/hive/conf/hive-site.xml (No such file or directory) ... ... .. org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: java.io.FileNotFoundException: /etc/hive/conf/hive-site.xml (No such file or directory) 提示文件不存在。 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink 连接Hive hiveConf路径配置

Dears, flink在连接hive时，需配置hiveConf所在路径 我已经下载了集群中hive-site.xml文件，不知道应该放在哪个目录 Job部署模式是 on Yarn ,请问代码中hiveConf应该放在哪个目录下，应该不是我启动job所在的机器吧？因为job提交后运行在hadoop集群，是无法找到相关目录的。 - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink 1.12 job on yarn 集成hive时如何配置 hiveConf

Dear all, 请问在flink在集成hive时候，需要配置hive的conf目录，我的job是on yarn提交的，那么如何配置这个hive conf路径呢？ String name = "myhive"; String defaultDatabase = "mydatabase"; String hiveConfDir = ""; // hive-site.xml路径 String version = "1.1.0-cdh5.8.3&qu

Flink 1.11.2 读写Hive以及对hive的版本支持

ng.ClassNotFoundException: org.apache.hadoop.mapreduce.TaskAttemptContext 是缺少MapReduce的相关包吗？ - Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Re:Re: Scala REPL YARN 运行模式报 NoSuchMethodError

感觉像是jline和Scala 某些包冲突所致，Scala我不太了解，你可以从以下方面做些尝试 1.在pom.xml或者其他相关文件中， 排除hadoop（以及其他涉及到jline的依赖）依赖中的jline子依赖，单独引入jline的依赖 我当时遇到的问题是，hadoop-common出现了版本冲突，在某个依赖中包含hadoop-common包，我在该依赖中排除了hadoop-common，然后在单独引入hadoop-common依赖，问题得以解决。 2. 改变(升级)Scala的版本 Thanks! Jacob -- Sent from: http://apache

Re: Scala REPL YARN 运行模式报 NoSuchMethodError

hi， 你的截图好像没有上传成功，通过你的描述，大概是NoSuchMethod之类的错误，我前几天在升级flink版本时候也遇到过类似问题，后来的解决方案是 导入hadoop classpath （export HADOOP_CLASSPATH=`hadoop classpath`）解决的，如果没有解决你的问题，尝试把flink-shaded-hadoop-2-uber*-*.jar放在 flink/lib下面 Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink 新版本的WEB UI taskmanager没有logs

是的，我看到项目中有logback.xml配置，在pom中也有logback-classic依赖 ch.qos.logback logback-classic 1.2.3 我移除这个依赖后，在UI中可以看到相关日志了！ 谢谢！ Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink 新版本的WEB UI taskmanager没有logs

state.checkpoints.num-retained: 3 restart-strategy: fixed-delay restart-strategy.fixed-delay.attempts: 1000 restart-strategy.fixed-delay.delay: 30 s jobmanager.execution.failover-strategy: region classloader.resolve-order: parent-first 3. job运行方式：on yarn 4. hadoop版本：2.6 Thanks! Jacob -- Sent

Flink 新版本的WEB UI taskmanager没有logs

<http://apache-flink.147419.n8.nabble.com/file/t1162/taskmanager.png> Hello everyone! 如上图所示，升级后的flink，为什么看不到taskmanager的日志了。在Stdout中能看自己代码中打的log，但flink自身的log以及springboot相关的log等，都无法看到，不知何因？升级后日志系统需要重新配置吗？ Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink 1.12.0 写ORC文件，自定义文件名

目前，Flink在写ORC文件时候，可通过OutputFileConfig类配置文件的前缀后缀：.withPartPrefix("prefix")、.withPartSuffix(".ext") 生成的文件格式为：part-- 有没有可以完全自定义生成的文件名，比如："dt=1608006781874",dt=时间戳的形式生成文件，目的是可以直接作为分区load在hive表。后期容易操作hive表。如果是flink默认的文件格式无法load在hive表。 -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: flink-shaded-hadoop-2-uber*-* 版本确定问题

谢谢回复！ 这个文档我也有查看 前几日在flink1.9-1.12各个客户端测试提交job时候发现 对于1.10+的版本，我手动导入export HADOOP_CLASSPATH=`hadoop classpath`，没有效果，各种报错，基本都是Hadoop相关类、方法不存在（NoSuchMethod之类错误），把pom文件改来改去依然无用，后来只在pom文件中导入依赖：flink-shaded-hadoop-2-uber*-*，竟然可以正常提交并运行job了。 -- Sent from: http://apache-flink.147419.n8.nabble.com/

flink-shaded-hadoop-2-uber*-* 版本确定问题

请问在升级flink版本的过程中，需要在flink/lib里面引入该包，但该包的版本号如何确定？ flink-shaded-hadoop-2-uber*-* -- Sent from: http://apache-flink.147419.n8.nabble.com/

Flink 1.10.0 on yarn 提交job失败

Hello, 请问在flink 1.10.0 on yarn提交job出现此问题是什么原因，hadoop jar包依赖吗？该程序在1.10以下的版本均可运行，在1.10.0无法提交。 谢谢！ [jacob@hadoop001 bin]$ ./yarn logs -applicationId application_1603495749855_57650 20/12/11 18:52:55 INFO client.RMProxy: Connecting

Flink 1.11.2 on yarn 提交job失败(附完整日志)

大家好，有一问题困扰许久。 *Flink1.7.2升级到1.11.2 job无法提交* 【现象】 可以提交demo的job（WordCount）,无法提交自定义的job，自己的job有操作Hadoop，读ORC，写hive等操作 提交自己的job，各种日志如下

Flink 1.11.2 on yarn 提交job失败(附完整日志)

大家好，有一问题困扰许久。*Flink1.7.2升级到1.11.2 job无法提交*【现象】可以提交demo的job（WordCount）,无法提交自定义的job，自己的job有操作Hadoop，读ORC，写hive等操作提交自己的job，各种日志如下

Re: flink 1.11.2 on yarn 可用slot始终为0，job无法提交

: Class path contains multiple SLF4J bindings. SLF4J: Found binding in [jar:file:/data/hadoop/dn/sde/yarn/nm/usercache/jacob/appcache/application_1603495749855_55197/filecache/26/test_job.jar!/org/slf4j/impl/StaticLoggerBinder.class] SLF4J: Found binding in [jar:file:/data/hadoop/dn/sdd/yarn/nm/usercache

Flink 1.11.2 on yarn 提交job失败

Time:Wed Dec 09 17:03:38 -0800 2020 LogLength:802 Log Contents: SLF4J: Class path contains multiple SLF4J bindings. SLF4J: Found binding in [jar:file:/data/hadoop/dn/sde/yarn/nm/usercache/jacob/appcache/application_1603495749855_55197/filecache/26/test_job.jar!/org/slf4j/impl/StaticLoggerBinder.class

Flink 1.11.2 on yarn 提交job失败

Dec 09 17:03:38 -0800 2020LogLength:802Log Contents:SLF4J: Class path contains multiple SLF4J bindings.SLF4J: Found binding in [jar:file:/data/hadoop/dn/sde/yarn/nm/usercache/jacob/appcache/application_1603495749855_55197/filecache/26/test_job.jar!/org/slf4j/impl/StaticLoggerBinder.class]SLF4J: Found

Re: Flink 1.11.2 on yarn报错

该问题已经fix，确实是java版本问题！ -- Sent from: http://apache-flink.147419.n8.nabble.com/

flink 1.11.2 on yarn 可用slot始终为0，job无法提交

启动命令： ./bin/flink run-application -t yarn-application -Djobmanager.memory.process.size=2048m -Dtaskmanager.memory.process.size=2048m -Dyarn.application.name="Test Job" -c com.jacob.Main /opt/app/test.jar

Flink 1.11.2 on yarn报错

*【环境】* Flink 版本：1.11.2 Hadoop 版本 ：2.6.0-cdh5.8.3 Java 版本： 1.8.0_144 - *【命令】* [jacob@localhost flink-1.11.2]$ ./bin/yarn-session.sh -jm 1024m -tm 2048m *【现象】* 2020-12-08 18:06:00,134 ERROR org.apache.flink.yarn.cli.FlinkYarnSessionCli

Re: "stepless" sliding windows?

on every event. The windows need to be of a fixed >>> size, but to have their start and end times update continuously, and I'd >>> like to trigger on every event. Is this a bad idea? I've googled and read >>> the docs extensively and haven't been able to identify built-in >>> functionality or examples that map cleanly to my requirements. >>> >>> OK, I just found DeltaTrigger, which looks promising... Does it make >>> sense to write a WindowAssigner that makes a new Window on every event, >>> allocation rates aside? >>> >>> Thanks! >>> >>> -0xe1a >>> >> -- Jacob Sevart Software Engineer, Safety

Re: Checkpoints for kafka source sometimes get 55 GB size (instead of 2 MB) and flink job fails during restoring from such checkpoint

> > this problem (when the flink job creates “abnormal” checkpoints). > > > > Configuration: > > > > We are using flink 1.8.1 on emr (emr 5.27) > > > > Kafka: confluence kafka 5.4.1 > > > > Flink kafka connector: > > org.apache.flink:flink-connector-kafka_2.11:1.8.1 (it includes > > org.apache.kafka:kafka-clients:2.0.1 dependencies) > > > > Our input kafka topic has 32 partitions and related flink source has > 32 > > parallelism > > > > We use pretty much all default flink kafka concumer setting. We only > > specified: > > > > CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, > > > > ConsumerConfig.GROUP_ID_CONFIG, > > > > CommonClientConfigs.SECURITY_PROTOCOL_CONFIG > > > > Thanks a lot in advance! > > > > Oleg > > > > > -- Jacob Sevart Software Engineer, Safety

Re: Very large _metadata file

https://github.com/apache/flink/pull/11475 On Sat, Mar 21, 2020 at 10:37 AM Jacob Sevart wrote: > Thanks, will do. > > I only want the time stamp to reset when the job comes up with no state. > Checkpoint recoveries should keep the same value. > > Jacob > > On Sat, Mar 2

Re: Very large _metadata file

Thanks, will do. I only want the time stamp to reset when the job comes up with no state. Checkpoint recoveries should keep the same value. Jacob On Sat, Mar 21, 2020 at 10:16 AM Till Rohrmann wrote: > Hi Jacob, > > if you could create patch for updating the union state

Re: Very large _metadata file

? Jacob On Tue, Mar 17, 2020 at 2:51 AM Till Rohrmann wrote: > Did I understand you correctly that you use the union state to synchronize > the per partition state across all operators in order to obtain a global > overview? If this is the case, then this will only work in case of a &

Re: Very large _metadata file

would you go about implementing something like that? On Mon, Mar 16, 2020 at 1:54 PM Till Rohrmann wrote: > Hi Jacob, > > I think you are running into some deficiencies of Flink's union state > here. The problem is that for every entry in your list state, Flink stores > a separate

Re: Very large _metadata file

Oh, I should clarify that's 43MB per partition, so with 48 partitions it explains my 2GB. On Fri, Mar 13, 2020 at 7:21 PM Jacob Sevart wrote: > Running *Checkpoints.loadCheckpointMetadata *under a debugger, I found > something: > *subtaskState.managedOperatorState[0].sateNameToPartiti

Re: Very large _metadata file

ava.time.Instant). I see a way to end up fewer items in the list, but I'm not sure how the actual size is related to the number of offsets. Can you elaborate on that? Incidentally, 42.5MB is the number I got out of https://issues.apache.org/jira/browse/FLINK-14618. So I think my two problems are clo

Re: Very large _metadata file

at the end. If it is putting state in there, under normal circumstances, does it make sense that it would be interleaved with metadata? I would expect all the metadata to come first, and then state. Jacob Jacob On Thu, Mar 5, 2020 at 10:53 AM Kostas Kloudas wrote: > Hi Jacob, > > As I said p

Re: Very large _metadata file

Kostas and Gordon, Thanks for the suggestions! I'm on RocksDB. We don't have that setting configured so it should be at the default 1024b. This is the full "state.*" section showing in the JobManager UI. [image: Screen Shot 2020-03-04 at 9.56.20 AM.png] Jacob On Wed, Mar 4, 2020 at 2

Very large _metadata file

look like HDFS paths, which leaves a lot of that file-size unexplained. What else is in there, and how exactly could this be happening? We're running 1.6. Jacob

State key serializer has not been configured in the config.

ovide some help ? Best regards Jacob private ListState<Tuple2<String, Integer>> deltaPositions; @Override public void open(org.apache.flink.configuration.Configuration parameters) throws Exception { // Create state variable ListStateDescriptor<Tuple2<