答复: 关于sink失败 不消费kafka消息的处理

感谢，目前也是通过打开checkpoint来改进的，待会测试一下看看是不是可以 -邮件原件- 发件人: Benchao Li [mailto:libenc...@apache.org] 发送时间: 2020年8月26日 星期三 12:59 收件人: user-zh 主题: Re: 关于sink失败 不消费kafka消息的处理 这种情况需要打开checkpoint来保证数据的不丢。如果sink没有两阶段提交，那就是at least once语义。 范超 于2020年8月26日周三 上午11:38写道： > 大家好，我现在有个疑问 >

Re: [ANNOUNCE] Apache Flink 1.10.2 released

Thanks Zhu for the great work and everyone who contributed to this release! Best, Xingbo Guowei Ma 于2020年8月26日周三 下午12:43写道： > Hi, > > Thanks a lot for being the release manager Zhu Zhu! > Thanks everyone contributed to this! > > Best, > Guowei > > > On Wed, Aug 26, 2020 at 11:18 AM Yun Tang

Re: [ANNOUNCE] Apache Flink 1.10.2 released

Thanks Zhu for the great work and everyone who contributed to this release! Best, Xingbo Guowei Ma 于2020年8月26日周三 下午12:43写道： > Hi, > > Thanks a lot for being the release manager Zhu Zhu! > Thanks everyone contributed to this! > > Best, > Guowei > > > On Wed, Aug 26, 2020 at 11:18 AM Yun Tang

Re: ProcessWindowFunction为何在clear方法中无法清理状态-v1.10.1

按我的理解，参考aggregate(AggregateFunction aggFunction, ProcessWindowFunction windowFunction)方法， 窗口中的状态数据是存在某个聚合函数里的，processWindowFunction只是处理窗口的结果，需要通过context获取对应的窗口state来做清理。 x <35907...@qq.com> 于2020年8月25日周二 下午6:25写道： > >

Re: 关于sink失败 不消费kafka消息的处理

这种情况需要打开checkpoint来保证数据的不丢。如果sink没有两阶段提交，那就是at least once语义。 范超 于2020年8月26日周三 上午11:38写道： > 大家好，我现在有个疑问 > 目前我使用kafka作为source，经过计算以后，将结果sink到数据库； > > 后来日志数据库发生了timeout或者宕机，kafka这边的主题，却消费掉了造成了数据丢失，那么如何设置才可以确认在sink失败的时候，不提交kafka的消费位移呢？ > > > 多谢大家了 > > 范超 > -- Best, Benchao Li

Re: [ANNOUNCE] Apache Flink 1.10.2 released

Hi, Thanks a lot for being the release manager Zhu Zhu! Thanks everyone contributed to this! Best, Guowei On Wed, Aug 26, 2020 at 11:18 AM Yun Tang wrote: > Thanks for Zhu's work to manage this release and everyone who contributed > to this! > > Best, > Yun Tang >

Re: [ANNOUNCE] Apache Flink 1.10.2 released

Hi, Thanks a lot for being the release manager Zhu Zhu! Thanks everyone contributed to this! Best, Guowei On Wed, Aug 26, 2020 at 11:18 AM Yun Tang wrote: > Thanks for Zhu's work to manage this release and everyone who contributed > to this! > > Best, > Yun Tang >

Re: flink on yarn 提交作业application模式的2个疑问

好的，谢谢大佬！ 发自我的iPhone > 在 2020年8月26日，11:48，Yang Wang 写道： > > 是的，Application模式是可以这么使用的 > > 但需要注意的是： > * 非HA模式下，如果JobManager failover了，会重新开始执行所有job。 > * HA模式，还不能支持多个job > > > Best, > Yang > > yang zhang 于2020年8月26日周三 上午11:17写道： > >> 感谢大佬回复，了解了。 >> 那是不是相当于一个main方法 >> 先写A逻辑，然后execute(), >>

Re: flink on yarn 提交作业application模式的2个疑问

是的，Application模式是可以这么使用的 但需要注意的是： * 非HA模式下，如果JobManager failover了，会重新开始执行所有job。 * HA模式，还不能支持多个job Best, Yang yang zhang 于2020年8月26日周三 上午11:17写道： > 感谢大佬回复，了解了。 > 那是不是相当于一个main方法 > 先写A逻辑，然后execute(), > 后面写B逻辑，再execute(), > 假如B逻辑必须等待A逻辑执行完毕才执行。 > 用这种方式提交直接就解决了这个场景，也不用第三方通知实现了。 > > > > >

Re: 请教一下flink链接hive的权限控制

Hi, 不好意思，我查了一下hive文档发现SQL standard authorization是在CLI/HS2端做的，那HiveCatalog目前没办法支持这种模式。HMS端可以用storage based authorization，也就是你说的通过HDFS的ACL来控制权限。这种模式对外表和内表都是有效的，但管理起来一般比较繁琐，需要人工去设置路径的ACL。 On Wed, Aug 26, 2020 at 11:08 AM faaron zheng wrote: > Hi Rui，感谢你的分享。我简单试了一下开启SQL Standard >

??????????????????checkpoint????

Hi Congxian, iteration source??barrier?? ??barrier??operator??barrier??checkpoint?? ??operator

关于sink失败 不消费kafka消息的处理

大家好，我现在有个疑问 目前我使用kafka作为source，经过计算以后，将结果sink到数据库； 后来日志数据库发生了timeout或者宕机，kafka这边的主题，却消费掉了造成了数据丢失，那么如何设置才可以确认在sink失败的时候，不提交kafka的消费位移呢？ 多谢大家了 范超

Re: flink checkpoint导致反压严重

Hi 对于已经改为at least once的checkpoint，其在checkpoint时对于作业吞吐的影响只有task 同步阶段的snapshot，这个时间段的snapshot由于与task的主线程的数据访问持有同一把锁，会影响主线程的数据处理。但是就算这样，我也很怀疑checkpoint本身并不是导致早上10点高峰期无法运行的罪魁祸首。 使用异步的，支持增量的state backend （例如RocksDBStateBackend）会大大缓解该问题。 建议排查思路： 1. 检查使用的state backend类型 2.

【闫云鹏】Flink cdc 连接mysql5.7.25报错

Hi all: 使用flink cdc连接mysql 5.7.25 使用默认的8.0.16的mysql驱动报错如下信息 Caused by: java.sql.SQLNonTransientConnectionException: CLIENT_PLUGIN_AUTH is required at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:110) at

Re: 报错 Could not resolve ResourceManager address akka.tcp://flink@hostname:16098/user/resourcemanager

另外，可以看下 TM 日志。如果有正在运行的 TM，那么说明 RM 和 TM 之间的心跳是正常的，也就证明 RM 是没有问题的。如果没有 TM 在运行，有可能是因为长时间空闲没有任务运行被释放了，可以找最近被释放的 TM 的日志看下是因为心跳超时释放的，还是空闲超市 RM 主动释放的。 Thank you~ Xintong Song On Wed, Aug 26, 2020 at 11:07 AM Xintong Song wrote: > 这个报错看起来是 RM 和 RestServer 服务都是正常的，反倒是 JobMaster 没有拿到 leader，所以 RM

Re: [ANNOUNCE] Apache Flink 1.10.2 released

Thanks for Zhu's work to manage this release and everyone who contributed to this! Best, Yun Tang From: Yangze Guo Sent: Tuesday, August 25, 2020 14:47 To: Dian Fu Cc: Zhu Zhu ; dev ; user ; user-zh Subject: Re: [ANNOUNCE] Apache Flink 1.10.2 released Thanks

Re: [ANNOUNCE] Apache Flink 1.10.2 released

Thanks for Zhu's work to manage this release and everyone who contributed to this! Best, Yun Tang From: Yangze Guo Sent: Tuesday, August 25, 2020 14:47 To: Dian Fu Cc: Zhu Zhu ; dev ; user ; user-zh Subject: Re: [ANNOUNCE] Apache Flink 1.10.2 released Thanks

Re: flink on yarn 提交作业application模式的2个疑问

感谢大佬回复，了解了。 那是不是相当于一个main方法 先写A逻辑，然后execute(), 后面写B逻辑，再execute(), 假如B逻辑必须等待A逻辑执行完毕才执行。 用这种方式提交直接就解决了这个场景，也不用第三方通知实现了。 发自我的iPhone > 在 2020年8月26日，10:44，Yang Wang 写道： > > 1. 目前Flink是允许在一个user jar的main里面，运行多次execute的，包括executeAsyc。 > 每执行一次execute，就是提交了一个任务。所以你是可以这样来实现一个Application里面 >

回复：请教一下flink链接hive的权限控制

Hi Rui，感谢你的分享。我简单试了一下开启SQL Standard Authorization，没什么效果，不知道是我用的不对还是我们hive被定制过。此外，我发现在使用kerberos的情况下，可以通过hdfs的路径来控制权限，不过这种情况主要对外表比较有效。 在2020年08月25日 21:34，Rui Li 写道： Hi， Authentication的话支持kerberos，应该正常做kinit就可以了。或者可以设置flink

Re: 报错 Could not resolve ResourceManager address akka.tcp://flink@hostname:16098/user/resourcemanager

这个报错看起来是 RM 和 RestServer 服务都是正常的，反倒是 JobMaster 没有拿到 leader，所以 RM 没有响应 JM 的请求。 你看下最早出现心跳超时 RM-JM 连接断开的时候，jobmanager 日志里面有没有 "Disconnect job manager xxx for job xxx from the resource manager." 这样的信息，描述的是 RM 主动断开了与 JM 的连接。 另外，ZK 这个报错是只出现了一次，还是对每个提交后无法调度的作业都出现了？ Thank you~ Xintong Song On Wed,

Re: flink on yarn 提交作业application模式的2个疑问

1. 目前Flink是允许在一个user jar的main里面，运行多次execute的，包括executeAsyc。 每执行一次execute，就是提交了一个任务。所以你是可以这样来实现一个Application里面 运行多个job。但这里需要注意的是，目前只有非HA的Application模式可以支持多job HA模式还不行，具体原因主要是恢复起来会很麻烦，还没有实现。 2. Flink会检查用户提交的user jar，然后从其中提取出来。具体可以看一下PackagedProgram#loadMainClass 的实现。 Best, Yang yang zhang

flink1.11 kafka sql connector

hi 我正在使用DDL语句创建kafka数据源，但是查看文档时候发现没有报漏参数指定消费者组的offset是否提交，请问这个默认情况下offset会不会提交到kafka分区

Re: Flink-1.11.1 Application-Mode提交测试

你给的这个报错目前并不能发现任何有效的信息，有可能的原因是在运行main的时候出错了 Application模式和perjob模式的很大的一个区别在于用户main运行的位置不一样 你把JobManager的log发一下，这样能看到更多详细的报错信息 Best, Yang amen...@163.com 于2020年8月25日周二 下午3:29写道： > hi, everyone > > 当我把jar包都上传至hdfs时，使用如下命令进行application mode提交， > > ./bin/flink run-application -t yarn-application

Re: 关于flink任务的日志收集到kafka，可以在logback配置文件中，加如每个job的id或者name吗？

我建议可以通env的方式传，在logback或者log4j配置中直接引用相应的env 例如，可以通过如下两个配置传递clusterId到环境变量 containerized.master.env.clusterId=my-flink-cluster containerized.taskmanager.env.clusterId=my-flink-cluster 另外，也有一些内置的环境变量可以来使用 _FLINK_CONTAINER_ID _FLINK_NODE_ID Best, Yang zilong xiao 于2020年8月25日周二 下午5:32写道： >

Re: 报错 Could not resolve ResourceManager address akka.tcp://flink@hostname:16098/user/resourcemanager

1. 我在jobmanager日志中并没有找到相关的日志信息； 2. 用的是zk，看当时的日志有如下报错： 2020-08-22 05:38:30,974 INFO org.apache.zookeeper.server.PrepRequestProcessor: Got user-level KeeperException when processing sessionid:0x26ea6955ea90581 type:create cxid:0x32311f1 zxid:0x2079a5014 txntype:-1 reqpath:n/a Error

Re: 有没有可能使用tikv作为flink 分布式的backend

了解了，非常感谢这么详细的解答。 提到的改动点和queryable state 我都去详细了解一下 -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: flink checkpoint导致反压严重

Hi zhanglachun, 你们使用 checkpoint 的方式是什么？是否有 full gc 的情况呢 Best, LakeShen 徐骁 于2020年8月26日周三 上午2:10写道： > input > .keyBy() > .timeWindow() > .reduce(new MyReduceFunction(), new MyProcessWindowFunction()); > > 可以看一下官网的 reduce 用法, 可以在数据进入窗口的时候进行处理(排序), 而没必要等到 window 结束后再进行 这样可以把时间分散到 > window

Re: flink1.11单机执行slot出错

是单机运行 standalone 模式吗？感觉像是 TM 没起来。 jps 以下看看 TM 起来了没有，如果没起来的话找下 TM 日志看看具体原因是什么。 Thank you~ Xintong Song On Wed, Aug 26, 2020 at 9:25 AM 小学生 <201782...@qq.com> wrote: > 麻烦请教一个问题，在单机的Linux下运行，任务报这个错误，怎么解决呢？ > java.util.concurrent.CompletionException: >

Re: 报错 Could not resolve ResourceManager address akka.tcp://flink@hostname:16098/user/resourcemanager

> > 1. 怎么可以确认是leader丢失呢？ > 看下是否能找到类似 "ResourceManager xxx was revoked leadership" 的日志 > 2. 通常是什么原因造成的呢？比如网络延迟？或者机器负载过高？ 网络原因是一种可能。另外也可能是 HA service 有问题，要看你集群用的是什么 HA（比如ZooKeeper），排查下 HA 的服务状态是否正常。 > 3. 有什么办法可以恢复吗？ 要根据具体原因才能知道如何恢复。 你现在是 yarn-session 上还有正在运行的作业吗？有可能试下停掉再重启 yarn-session 吗？

Re: 从kafka读取数据转化为Pojo类型的数据流注册成表，新消息来了，注册表的查询结果不打印

我知道了，我的查询sql条件的问题，已经改好了。 谢谢 发自我的iPhone > 在 2020年8月25日，16:12，yang zhang 写道： > > 从kafka读取数据转化为Pojo类型的数据流注册成表，新消息来了，注册表的查询结果不打印 > > 请问是不支持pojo流注册表吗？只能是Row类型吗？ > > 下面是相关代码 > > > >//1.创建执行环境 >StreamExecutionEnvironment streamEnv = >

flink1.11????????slot????

??Linux java.util.concurrent.CompletionException: org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: No pooled slot available and request to ResourceManager for new slot failed

Re: Loading FlinkKafkaProducer fails with LinkError

They are definitely equal, the same JAR is copied in subsequent lines in the Dockerfile. Regarding the NoSuchMethodException, I'll look it up and let you know tomorrow. On Tue, Aug 25, 2020, 22:59 Chesnay Schepler wrote: > The simplest answer is that they are in fact not equal; maybe it is a

Re: [DISCUSS] Remove Kafka 0.10.x connector (and possibly 0.11.x)

+1 to remove both the 1.10 and 1.11 connectors. The connectors have not been actively developed for some time. They are basically just sitting around causing noise by causing test instabilities and eating CI time. It would  also allow us to really simplify the module structure of the Kafka

[ANNOUNCE] Weekly Community Update 2020/31-34

Dear community, The "weekly" community update is back after a short summer break! This time I've tried to cover most of what happened during the last four weeks, but I might pick up some older topics in the next weeks' updates, too. Activity on the dev@ mailing list has picked up quite a bit as

Re: Loading FlinkKafkaProducer fails with LinkError

The simplest answer is that they are in fact not equal; maybe it is a jar of an older version of your setup? Can you give some details on the NoSuchMethodException? Specifically whether it tries to access something from the Kafka connector, or from your own user code. On 25/08/2020 21:27,

Re: Loading FlinkKafkaProducer fails with LinkError

OK, I think I figured it out. It looks like the uber-jar is also being placed under `lib`, which is probably the cause of the problem. Question is, why does it identify it as two different versions? It's exactly the same JAR. On Tue, Aug 25, 2020 at 10:22 PM Yuval Itzchakov wrote: > I'm afraid

Re: Loading FlinkKafkaProducer fails with LinkError

I'm afraid it's not being printed out due to different log levels :( Yes, I build the image myself. It takes the tar file from https://archive.apache.org/dist/flink/flink-1.9.0/ and unpacks it into the image. I've ran: find . -iname "*.jar" |

Re: flink1.10中hive module 没有plus，greaterThan等函数

Moving this to the chinese user mailing  list. On 25/08/2020 16:16, Andrey Zagrebin wrote: Hi Faaron, This mailing list is for support in English. Could you translate your question into English? You can also subscribe to the user mailing list in Chinese to get support in Chinese [1] Best,

Re: Loading FlinkKafkaProducer fails with LinkError

The NoSuchMethodException shows that the class is still on the classpath, but with a different version than your code is expecting. Otherwise you would've gotten a different error. This implies that there are 2 versions of the kafka dependencies on the classpath in your original run; it

Re: Loading FlinkKafkaProducer fails with LinkError

Will it be enough to provide you the output of `-verbose:class`? Or do you want me to add additional arguments? On Tue, Aug 25, 2020 at 6:20 PM Arvid Heise wrote: > Small correction: you'd bundle the connectors in your uber jar like you > did but you usually don't put it into flink-dist. > > So

Re: [DISCUSS] Removing deprecated methods from DataStream API

I would argue that the guarantees of @Public methods that became ineffective were broken when they became ineffective (and were deprecated). - ExecutionConfig#disable/enableSysoutLogging (deprecated in 1.10) - ExecutionConfig#set/isFailTaskOnCheckpointError (deprecated in 1.9) Removing

Re: Loading FlinkKafkaProducer fails with LinkError

Hi Arvid, I'm running Flink in a job cluster on k8s using the Lyft Operator. The flink image that I'm building does not have the flink-connector-kafka library in it's JAR, I've made sure of this using `jar -tf`. Additionally, once I removed the dependency from my uber jar, it failed with a

Re: flink checkpoint导致反压严重

input .keyBy() .timeWindow() .reduce(new MyReduceFunction(), new MyProcessWindowFunction()); 可以看一下官网的 reduce 用法, 可以在数据进入窗口的时候进行处理(排序), 而没必要等到 window 结束后再进行 这样可以把时间分散到 window 里面

Re: [DISCUSS] Remove Kafka 0.10.x connector (and possibly 0.11.x)

Hi Aljoscha, I am assuming you're asking about dropping the flink-connector-kafka-0.10/0.11 modules, right? Or are you talking about removing support for Kafka 0.10/0.11 from the universal connector? I am in favor of removing flink-connector-kafka-0.10/0.11 in the next release. These modules

Re: Flink Table API/SQL debugging, testability

Hi, What exactly are you looking for? I think the simplest stub for a test could be sth like:             final TableEnvironment env = TableEnvironment.create(...);             TableResult result = env.fromValues(...)                 .select(...)                 .execute();             try

Re: How jobmanager and task manager communicates with each other ?

Hi Sidhant, (1) If we are not using Flink's HA services then how we can dynamically > configure task manager nodes to connect to job manager? Any suggestions or > best practices? Not sure what you mean by 'dynamically'. I think you have to restart the task manager with the new configuration to

Re: Transition Flink job from Java to Scala with state migration

Hi Daksh, You need to find which type causes the problem: Long, MyCustomObject or maybe something else. You could share the logs with full exception stack trace. My guess is that your scala code uses another serializer for the failing type. See also docs to understand serialization in Flink [1]

Re: Flink OnCheckpointRollingPolicy streamingfilesink

Hi Vijay, I think it depends on your job requirements, in particular how many records are processed per second and how much resources you have to process them. If the checkpointing interval is short then the checkpointing overhead can be too high and you need more resources to efficiently keep

Re: How jobmanager and task manager communicates with each other ?

Hi Till, Thanks for the reply. (1) If we are not using Flink's HA services then how we can dynamically configure task manager nodes to connect to job manager? Any suggestions or best practices? (2) Which and how flink's HA service can be used for the service discovery of job manager ? Regards

Re: Default Flink Metrics Graphite

Thanks for inputs Nikola. I will check on graphite side. Sent from my iPhone > On Aug 23, 2020, at 9:26 PM, Nikola Hrusov wrote: > >  > Hi Vijay, > > Your steps look correct to me. > Perhaps you can double check that the graphite port you are sending is > correct? THe default carbon port

flink checkpoint导致反压严重

大佬们好,我一个flink任务,计算一分钟内的某项几项指标的中位数,总共5个指标,因为中位数计算需要全窗口数据排序,所以计算比较复杂,现在遇到的问题的是一旦开启checkpoint任务就从source端开始反压严重,但关闭checkpoint就正常运行. 目前优化的步骤有: 1.语义放弃exactlyonce 改到atleast 2.分析发现keyby过程中有数据倾斜,已改成分布聚合,在第一轮聚合中key后添加随机数,在去除key后缀进行第二轮聚合 3.计算过程中使用RoaringBitmap作为中间数据缓存容器,最大限度减少内存损耗 4.增大并行度,提交时增大-yjm -ytm

Re: Loading FlinkKafkaProducer fails with LinkError

Small correction: you'd bundle the connectors in your uber jar like you did but you usually don't put it into flink-dist. So please double-check if it's also in flink-dist and remove it there. If not, then please add the full classpath log statement. It might also be a bug related to restoring

Re: Loading FlinkKafkaProducer fails with LinkError

Hi Yuval, How do you execute Flink? Can you show us the log entry with the classpath? I'm guessing that you have Kafka bundled in your uber-jar and additionally also have the connector in flink-dist/lib. If so, you simply need to remove it in one place. In general, if you use flink-dist, you'd

Re: OOM error for heap state backend.

Hi Vishwas, If you use Flink 1.7, check the older memory model docs [1] because you referred to the new memory model of Flink 1.10 in your reference 2. Could you also share a screenshot where you get the state size of 2.5 GB? Do you mean Flink WebUI? Generally, it is quite hard to estimate the

Transition Flink job from Java to Scala with state migration

Hello, We run a Stream API based Flink application on 1.10.0, coded in Java. While moving this job to Scala (reasons unrelated to Flink/this application), we are getting the following error when trying to instantiate Scala job from a savepoint taken in the Java job:

Re: JSON to Parquet

Hi Averell, If you can describe the JSON schema I'd suggest looking into the SQL API. (And I think you do need to define your schema upfront. If I am not mistaken Parquet must know the common schema.) Then you could do sth like: CREATE TABLE json (     // define the schema of your json data )

Re: JSON to Parquet

Hi Averell, If you can describe the JSON schema I'd suggest looking into the SQL API. (And I think you do need to define your schema upfront. If I am not mistaken Parquet must know the common schema.) Then you could do sth like: CREATE TABLE json (     // define the schema of your json data )

Re: Monitor the usage of keyed state

Hi Mu, I would suggest to look into RocksDB metrics which you can enable as Flink metrics [1] Best, Andrey [1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/config.html#rocksdb-native-metrics On Fri, Aug 21, 2020 at 4:27 AM Mu Kong wrote: > Hi community, > > I have a Flink

回复：流处理任务中checkpoint失败

Hi Congxian, 测试的时候数据量是很小的，cpu使用比较低的，让我比较奇怪的一点是我杀掉任务重启的话，有时候能成功进行checkpoint，看了下日志，就是这个iteration source成功执行了snapshot，发起了barrier，进而影响到后续operator的checkpoint。失败的时候是该source无法snapshot，直至超时。 因为flink这一块，iteration是由head tail组成，是一个比较特殊的stream task，目前还没有看到jm这边是如何对此进行处理的。这个iteration source其实是由其他source

回复：flink1.11 sql问题

这个功能非常好的，因为第三方数据总是搞事情，动不动就加字段，改名字的。 -- 原始邮件 -- 发件人: Benchao Li https://issues.apache.org/jira/browse/FLINK-18002 酷酷的浑蛋

Re: 请教一下flink链接hive的权限控制

Hi， Authentication的话支持kerberos，应该正常做kinit就可以了。或者可以设置flink security相关的参数，如security.kerberos.login.keytab和security.kerberos.login.principal。具体可以参考： https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/config.html#auth-with-external-systems

Example flink run with security options? Running on k8s in my case

Hey everyone, I've been experimenting with Flink using https://github.com/GoogleCloudPlatform/flink-on-k8s-operator and I believe I've successfully deployed a JobManager and TaskManager with security enabled, and a self-signed certificate (the pods come up great).   However, I can't do much with

?????? flink 1.10.1 ???????? OutOfMemoryError: Metaspace

??ES5??pretty good?? ---- ??: "user-zh"

Re: 报错 Could not resolve ResourceManager address akka.tcp://flink@hostname:16098/user/resourcemanager

你好， 现在yarn-session上还是可以提交新作业的。只是运行时无法分配slot，报错无法解析 resourcemanager 地址。 如果是RM leadership丢失的话， 1. 怎么可以确认是leader丢失呢？ 2. 通常是什么原因造成的呢？比如网络延迟？或者机器负载过高？ 3. 有什么办法可以恢复吗？ Xintong Song 于2020年8月25日周二 下午5:26写道： > > > > 出现这个报错后就提交不了任务了 > > > 我确认一下，你之前这句话的意思，是出现这个报错之后，新的作业不能提交了，还是新的作业能提交但是提交之后页报这个找不到 RM 的错？

Re: flink1.11 sql问题

Hi, 这个功能已经在1.12支持了[1]，如果着急使用，可以cherry-pick回去试试看。 用法就是直接把这个字段声明为varchar，json format会帮你自动处理 [1] https://issues.apache.org/jira/browse/FLINK-18002 酷酷的浑蛋 于2020年8月25日周二 下午6:32写道： > > > 还没到udf那一步，直接用create table的方式，过来的数据就是获取不到值的， > CREATE TABLE test ( > a VARCHAR, > b INT > ) WITH ( >

回复： flink1.11 sql问题

还没到udf那一步，直接用create table的方式，过来的数据就是获取不到值的， CREATE TABLE test ( a VARCHAR, b INT ) WITH ( 'connector' = 'kafka', 'topic' = 'test', 'properties.bootstrap.servers' = 'xxx', 'properties.group.id' = 'groupid', 'scan.startup.mode' = 'group-offsets',

回复： flink1.11 sql问题

CREATE TABLE test ( a VARCHAR, b INT ) WITH ( 'connector'='kafka', 'topic'='test', 'properties.bootstrap.servers'='xxx', 'properties.group.id'='groupid', 'scan.startup.mode'='group-offsets', 'format'='json' );

ProcessWindowFunction??????clear??????????????????-v1.10.1

ProcessWindowFunction??clearenv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)?? .window(TumblingEventTimeWindows.of(Time.days(1)))

Re: 有没有可能使用tikv作为flink 分布式的backend

Hi 这种思路我觉得是可以尝试的，不过目前看需要改动的地方很多： 1. 需要更改RocksDB 创建checkpoint 到TiKV的代码逻辑 2. 需要改动RocksDB 从checkpoint resume的代码逻辑 3. 如果想要数据可以TiKV可以读取，那么TiKV中存储的格式要么与RocksDB内存储的一样，那这样子的话，lookup时候，需要能够反序列化Flink在RocksDB中的存储格式；要么是重新的格式，但这样子会导致RocksDB的checkpoint流程和时间都会增长。 4. TiKV中的数据的更新依赖于checkpoint

Re: [DISCUSS] FLIP-134: DataStream Semantics for Bounded Input

Thanks for creating this FLIP! I think the general direction is very good but I think there are some specifics that we should also put in there and that we may need to discuss here as well. ## About batch vs streaming scheduling I think we shouldn't call it "scheduling", because the decision

Re: 流处理任务中checkpoint失败

Hi 对于 checkpoint 超时失败的情况，需要看一下具体的原因，对于 source 没有完成的话，或许看一下相应并发（没完成 snapshot 的 source）的 CPU 占用情况，以及相应逻辑是否卡在哪里或许能看到一些线索。source 是收到 JM 的 rpc 后触发的 snapshot，所以这里相比其他的算子，不需要考虑 barrier 对齐的事情。 Best, Congxian Robert.Zhang <173603...@qq.com> 于2020年8月25日周二 上午12:58写道： > 看了日志，是由于部分checkpoint 超时未完成，web界面上

Re: 关于flink任务的日志收集到kafka，可以在logback配置文件中，加如每个job的id或者name吗？

1：想加入跟业务相关的字段，例如jobId，jobName，可以继承PatternLayout，重写doLayout即可，在方法中对日志进行填充 2：这些属性有办法可以从环境变量中获取 Jim Chen 于2020年8月25日周二 下午4:49写道： > 大家好： > 我们在做flink的日志收集到kafak时，使用的logback日志配置文件，目前的pattern是%d{-MM-dd > HH:mm:ss.SSS} [%thread] %-5level %logger{60} - >

Re: 报错 Could not resolve ResourceManager address akka.tcp://flink@hostname:16098/user/resourcemanager

> > 出现这个报错后就提交不了任务了 > 我确认一下，你之前这句话的意思，是出现这个报错之后，新的作业不能提交了，还是新的作业能提交但是提交之后页报这个找不到 RM 的错？ 从 RM 心跳超时但是整个进程还在运行这个现象来看，比较符合 RM leadership 丢失的情况，这种情况下 RM 会停止服务。 如果是新的作业干脆就无法提交了，也符合 rest server leadership 丢失的情况。 我目前怀疑是 HA 出现问题，导致 RM 和 rest server 都认为自己不再是 leader，但是又迟迟没有新的 leader 产生。所以对于

Re: 有没有可能使用tikv作为flink 分布式的backend

感谢解答 就像你说的，的确可以 用lookup方式实现一部分公用kv的共享 我的理解现有的 rocksdb backend 为：rocksdb+hdfs , 如果是变成：rocksdb+tikv ,这样在一些应用过程中产生的 kv指标数据最终会存储到 tikv之中，外部也有可能访问到，通过 lookup的方式，不知道这样是否可行 -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: 报错 Could not resolve ResourceManager address akka.tcp://flink@hostname:16098/user/resourcemanager

hi, Xintong： 我仔细查看了下日志，发现在报错"Could not resolve ResourceManager address"之前有如下日志： 2020-08-22 05:39:24,473 INFO org.apache.flink.runtime.jobmaster.JobMaster - The heartbeat of ResourceManager with id 6724e1ef8ee1c5fe5212eec6182319b6 timed out. 2020-08-22 05:39:24,473 INFO

关于flink任务的日志收集到kafka，可以在logback配置文件中，加如每个job的id或者name吗？

大家好： 我们在做flink的日志收集到kafak时，使用的logback日志配置文件，目前的pattern是%d{-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{60} - %msg，有没有什么办法在里面加入每个job的id，name或者tasknamanger的主机名之类的信息啊。在做ELK的时候，方便查询。 这个配置文件，是整个项目的，是基于Yarn的per job模式，难道每个主类打包的时候，都要改动不同的logbakc配置文件吗？

回复：请教一下flink链接hive的权限控制

我们当前用的是kerberos认证，需要额外配置什么么？ xiaoyan hua 邮箱：xiaoyanhua...@gmail.com 签名由 网易邮箱大师 定制 在2020年08月25日 15:54，faaron zheng 写道： Hi all， 我在使用flink sql-client链接hive metastore的时候，发现好像没有做任何权限控制，可以访问所有的表？这一块是没做么？有什么计划么？

Re: flink1.11 sql问题

直接CAST不可以吗？ 酷酷的浑蛋 于2020年8月25日周二 下午3:46写道： > 关键是那个值不是固定的，有时候是json，有时候是json数组，没办法固定写一个，现在我只想把value当做字符串获取到，难道没有办法吗 > > > > > 在2020年08月25日 15:34，taochanglian 写道： > flinksql，处理json ，对象的话用row，数组的话用array获取具体的值。 > > 在 2020/8/25 14:59, 酷酷的浑蛋 写道: > 还是这个问题，如果字段的值有时候是json有时候是json数组，那么我只想把它当做字符串显示，该怎么写？ >

Re: flink webui前端

flinlk-runtime-web，这个module下 罗显宴 <15927482...@163.com> 于2020年8月25日周二 下午3:43写道： > 大家好，请问flink的webui前端实现的源码在哪呀 > > > | | > 罗显宴 > | > | > 邮箱：15927482...@163.com > | > 签名由网易邮箱大师定制

Re: flink1.11 sql问题

这个需要你自定义UDF 酷酷的浑蛋 于2020年8月25日周二 下午3:46写道： > 关键是那个值不是固定的，有时候是json，有时候是json数组，没办法固定写一个，现在我只想把value当做字符串获取到，难道没有办法吗 > > > > > 在2020年08月25日 15:34，taochanglian 写道： > flinksql，处理json ，对象的话用row，数组的话用array获取具体的值。 > > 在 2020/8/25 14:59, 酷酷的浑蛋 写道: > 还是这个问题，如果字段的值有时候是json有时候是json数组，那么我只想把它当做字符串显示，该怎么写？ >

Flink??????????????????????

??A??B??AB?? ??B??ABA ??Flink??AB

从kafka读取数据转化为Pojo类型的数据流注册成表，新消息来了，注册表的查询结果不打印

从kafka读取数据转化为Pojo类型的数据流注册成表，新消息来了，注册表的查询结果不打印 请问是不支持pojo流注册表吗？只能是Row类型吗？ 下面是相关代码 //1.创建执行环境 StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment(); //全局参数设置 streamEnv.getConfig().setGlobalJobParameters(parameters2);

从kafka读取数据转化为Pojo类型的数据流注册成表，新消息来了，注册表的查询结果不打印

请教下开发者： 从kafka读取数据转化为Pojo类型的数据流注册成表，新消息来了，注册表的查询结果不打印 是不支持pojo格式流注册成表吗？只能是Row格式吗？ 下面是代码 //1.创建执行环境 StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment(); //全局参数设置 streamEnv.getConfig().setGlobalJobParameters(parameters2);

请教一下flink链接hive的权限控制

Hi all， 我在使用flink sql-client链接hive metastore的时候，发现好像没有做任何权限控制，可以访问所有的表？这一块是没做么？有什么计划么？

回复： flink1.11 sql问题

关键是那个值不是固定的，有时候是json，有时候是json数组，没办法固定写一个，现在我只想把value当做字符串获取到，难道没有办法吗 在2020年08月25日 15:34，taochanglian 写道： flinksql，处理json ，对象的话用row，数组的话用array获取具体的值。 在 2020/8/25 14:59, 酷酷的浑蛋 写道: 还是这个问题，如果字段的值有时候是json有时候是json数组，那么我只想把它当做字符串显示，该怎么写？ 在2020年08月25日 14:05，酷酷的浑蛋 写道： 我知道了 在2020年08月25日

flink webui前端

大家好，请问flink的webui前端实现的源码在哪呀 | | 罗显宴 | | 邮箱：15927482...@163.com | 签名由网易邮箱大师定制

Flink-1.11.1 Application-Mode提交测试

hi, everyone 当我把jar包都上传至hdfs时，使用如下命令进行application mode提交， ./bin/flink run-application -t yarn-application -Dyarn.provided.lib.dirs="hdfs:///user/flink/lib" -c com.yui.flink.demo.Kafka2Mysql hdfs:///user/flink/app_jars/kafka2mysql.jar 报异常如下：

Flink Table API/SQL debugging, testability

I was looking for testability, debugging practices on Flink Table API/SQL. Really difficult to find them when compared to Streaming API. Can someone please share their experiences on debugging, testability. -- A.Narasimha Swamy

Re: flink sql在实时数仓中，关联hbase维表频繁变化的问题

我们的维表，大概10个亿左右。每秒大概有4万的请求，要去查询，所以mysql扛不住。 还有就是维表数据变化后，需要秒级进行更新和关联的 china_tao 于2020年8月18日周二 下午11:13写道： > 个人觉得还是取舍的问题，我们现在用flink sql 做实时数仓，维度表暂时用mysql，与业务商定好更新事件后，配置flink sql > jdbc的lookup.cache.ttl参数来设置刷新时间，不知道你项目中，是维表数据变更后，需要秒级关联到消息中？ > > > > -- > Sent from:

Re: flink sql在实时数仓中，关联hbase维表频繁变化的问题

这个不行的话 shizk233 于2020年8月17日周一 下午11:23写道： > 有没有可能把维表数据也作为数据流从kafka输入呢 > > Jim Chen 于2020年8月17日周一 下午4:36写道： > > > 大家好： > > 我们现在在用flink > sql在做实时数仓，架构大概是kafka关联hbase维表，然后写入clickhouse。hbase维表是频繁变化的 > > 现在遇到的几个比较棘手的问题： > >

回复：flink1.11 sql问题

还是这个问题，如果字段的值有时候是json有时候是json数组，那么我只想把它当做字符串显示，该怎么写？ 在2020年08月25日 14:05，酷酷的浑蛋 写道： 我知道了 在2020年08月25日 13:58，酷酷的浑蛋 写道： flink1.11 读取json数据时format=“json”，当数据中某个字段的值是[{"a1":{"a2":"v2"}}]类似这种嵌套，flink取到的值就是空，这个怎么处理？

Re: [ANNOUNCE] Apache Flink 1.10.2 released

Thanks a lot for being the release manager Zhu Zhu! Congrats to all others who have contributed to the release! Best, Yangze Guo On Tue, Aug 25, 2020 at 2:42 PM Dian Fu wrote: > > Thanks ZhuZhu for managing this release and everyone else who contributed to > this release! > > Regards, > Dian >

Re: [ANNOUNCE] Apache Flink 1.10.2 released

Thanks a lot for being the release manager Zhu Zhu! Congrats to all others who have contributed to the release! Best, Yangze Guo On Tue, Aug 25, 2020 at 2:42 PM Dian Fu wrote: > > Thanks ZhuZhu for managing this release and everyone else who contributed to > this release! > > Regards, > Dian >

Re: [ANNOUNCE] Apache Flink 1.10.2 released

Thanks ZhuZhu for managing this release and everyone else who contributed to this release! Regards, Dian > 在 2020年8月25日，下午2:22，Till Rohrmann 写道： > > Great news. Thanks a lot for being our release manager Zhu Zhu and to all > others who have contributed to the release! > > Cheers, > Till >

Re: [ANNOUNCE] Apache Flink 1.10.2 released

Thanks ZhuZhu for managing this release and everyone else who contributed to this release! Regards, Dian > 在 2020年8月25日，下午2:22，Till Rohrmann 写道： > > Great news. Thanks a lot for being our release manager Zhu Zhu and to all > others who have contributed to the release! > > Cheers, > Till >

Re: [ANNOUNCE] Apache Flink 1.10.2 released

Great news. Thanks a lot for being our release manager Zhu Zhu and to all others who have contributed to the release! Cheers, Till On Tue, Aug 25, 2020 at 5:37 AM Zhu Zhu wrote: > The Apache Flink community is very happy to announce the release of > Apache Flink 1.10.2, which is the first

Re: [ANNOUNCE] Apache Flink 1.10.2 released

Great news. Thanks a lot for being our release manager Zhu Zhu and to all others who have contributed to the release! Cheers, Till On Tue, Aug 25, 2020 at 5:37 AM Zhu Zhu wrote: > The Apache Flink community is very happy to announce the release of > Apache Flink 1.10.2, which is the first

Re: How jobmanager and task manager communicates with each other ?

Hi Sidhant, the cluster components use tcp to communicate with each other. If you are not using Flink's HA services, then the TaskManager nodes need to be configured with the JobManager's address to connect to them. If you are using HA services, then the service discovery happens through the HA

回复：flink1.11 sql问题

我知道了 在2020年08月25日 13:58，酷酷的浑蛋 写道： flink1.11 读取json数据时format=“json”，当数据中某个字段的值是[{"a1":{"a2":"v2"}}]类似这种嵌套，flink取到的值就是空，这个怎么处理？