Re: Re: submit jobGraph error on server side

从错误堆栈上看你的请求应该是已经发到 jobmanager 上了，也就是不存在找不到端口的问题。 但是 jobmanager 在处理 submit job 的时候某个动作超时了。你这个问题是一旦把 gateway 分开就稳定复现吗？也有可能是 akka 偶然的超时。 Best, tison. 王智 于2019年8月7日周三 下午2:33写道： > 感谢您的回复与指导~ > > > 经过简单的验证（验证方案在邮件末尾），明确是网络问题。 > > > 现在我猜测是flink run 提交job graph 的时候打开了除 这四个以外的端口导致。麻烦再请教一下，flink

Re:关于event-time的定义与产生时间戳位置的问题。

Hi, 可以看下事件时间戳的生成，https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/event_timestamps_watermarks.html 下面例子里时间戳都是来自element里面的时间字段。还有一个AscendingTimestampExtractor。 /** * This generator generates watermarks assuming that elements arrive out of order, * but only to a certain degree. The

关于event-time的定义与产生时间戳位置的问题。

hi，all：   event time这个时间戳是在什么时候打到数据上面去的，看api是在flink source收到数据之后再标注的，并不是真正的数据源携带过来的（比如手机终端）？使用kafka source的话根据文档的定义kafka携带的时间戳也仅仅是kafka收到数据的时候打上的时间戳。 那么有个问题：以kafka为例，数据到队列的时候按「顺序」打上时间戳，那么如果数据是「乱序到达」的也被打上了「递增的时间戳」，后续基于event-time的处理都是基于这个时间戳来进行，那不就丧失了真实世界的定义吗？   不知道有哪里是我理解不对的地方望指教！   祝好~

Flink官网barrier疑问

Hi，老师： 老师，你好flink官网这个页面（https://ci.apache.org/projects/flink/flink-docs-release-1.8/internals/stream_checkpointing.htm）介绍barrier对齐的这里第三步 • Once the last stream has received barrier n, the operator emits all pending outgoing records, and then emits snapshot n barriers itself.

回复：Re: submit jobGraph error on server side

感谢您的回复与指导~ 经过简单的验证（验证方案在邮件末尾），明确是网络问题。 现在我猜测是flink run 提交job graph 的时候打开了除 这四个以外的端口导致。麻烦再请教一下，flink jobmanager 是否会打开新的端口进行通讯（或者还有其他端口配置我没有注意到） ports: - containerPort: 6123 protocol: TCP - containerPort: 6124 protocol: TCP - containerPort: 6125 protocol: TCP -

flink-1.8.1 yarn per job模式使用

Hi 大家好： 在使用flink过程中，运行官网给的命令 https://ci.apache.org/projects/flink/flink-docs-stable/ops/deployment/yarn_setup.html#run-a-single-flink-job-on-yarn 报错如下： ➜ flink-1.8.1 ./bin/flink run -m yarn-cluster ./examples/batch/WordCount.jar The

Re: flink-1.8.1 yarn per job模式使用

这个问题以前邮件列表有人提过...不过现在 user-zh 没有 archive 不好引用。 你看下是不是 lib 下面没有 flink-shaded-hadoop-2-uber--7.0.jar 这样一个文件。 1.8.1 之后 FLINK 把 hadoop(YARN) 的 lib 分开 release 了，你要指定自己的 HADOOP_CLASSPATH 或者下载 FLINK 官网 pre-bundle 的 hadoop。 具体可以看这个页面（https://flink.apache.org/downloads.html）第一段的内容。 Best, tison. 李玉环

Re: flink-1.8.1 yarn per job模式使用

非常感谢tison，完美的解决了我的问题，以后会多留意社区问题。 具体到自己的hadoop版本，就是在flink工程编译 flink-1.8.1/flink-shaded-hadoop/flink-shaded-hadoop2-uber/target 的jar放在lib下即可 Zili Chen 于2019年8月7日周三 下午7:33写道： > 这个问题以前邮件列表有人提过...不过现在 user-zh 没有 archive 不好引用。 > > 你看下是不是 lib 下面没有 flink-shaded-hadoop-2-uber--7.0.jar 这样一个文件。 > > 1.8.1

Re: Flink官网barrier疑问

你好，范瑞 Barrier alignment 这里并不会涉及 output/input queue，pending 的只是用于 alignment 的一小部分数据。 如果想了解 checkpoint 的原理，建议阅读文档中引用的两篇论文。[1] [2] 如果想了解 Flink 的具体实现，这里的文档是 internal 部分的，可能需要阅读一下相关代码了。[3] [4] 1. https://arxiv.org/abs/1506.08603 2.

回复：Re: Re: submit jobGraph error on server side

感谢大神， 是我配置的资源太少导致响应慢，导致akka 超时。 现在我换了一个k8s 集群，调大了资源，已经不再配到邮件中的发生的异常。 原始邮件 发件人："Zili Chen"< wander4...@gmail.com ; 发件时间：2019/8/7 15:32 收件人："王智"< ben.wa...@foxmail.com ; 抄送人："user-zh"< user-zh@flink.apache.org ; 主题：Re: Re: submit jobGraph error on server side

flink 结合canal统计订单gmv

hi 你好 需要用flink 解析mysql的binlog 统计订单表 产品维度的gmv， 但是一个insert的binlog 会出现同时购买多个订单 会出现一个集合的订单集合 但是我想统计订单里面产品维度的gmv，如下图 返回的是一个list的订单集合，但是我想取每个订单里面的产品id进行维度统计 ，请问flink 有什么算子 可以把一个list数据的流数据 变成多条流 谢谢

Re: 关于event-time的定义与产生时间戳位置的问题。

Hi, Q: event time这个时间戳是在什么时候打到数据上面去的, A: event time按字面意思理解为event发生的时间, 如果产生数据的设备提供了记录时间的字段, 并且业务逻辑也需要使用这个时间, 则可以将该时间作为event time. 更多信息可以参考 https://ci.apache.org/projects/flink/flink-docs-stable/dev/event_time.html 关于event time, processing time的描述 zhaoheng.zhaoh...@qq.com 于2019年8月8日周四 下午4:36写道：

请教 如何使用TableAPI connector将一个字段定义为rowtime属性

代码如下： StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment(); executionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(executionEnvironment); String[]

BucketingSink ????????????

??flink on yarn??1??TaskManager??4??slot??TaskManager4G??JobManager1G??BucketingSinkhdfs??3??checkpoint??1003300427427*300=128100=125KB80??80*300=24000=23KB??Flink ??

Re: env.readFile只读取新文件

恐怕没有现成的，自己写一个，继承 SourceFunction Thanks, Biao /'bɪ.aʊ/ On Wed, Jul 31, 2019 at 4:49 PM 王佩 wrote: > 如下代码: > > DataStreamSource source = env.readFile( > textInputFormat, > "/data/appData/streamingWatchFile/source", >

Re: BucketingSink 内存使用分析

把 heap dump 出来分析一下？ Thanks, Biao /'bɪ.aʊ/ On Thu, Aug 1, 2019 at 6:07 PM 九思 <1048095...@qq.com> wrote: > 请教各位老师，flink on >

flink 中文邮件订阅

flink 中文邮件订阅

rowtime/proctime

在执行 SQL 的时候遇见如下情况： streamTableEnvironment.registerDataStream（）注册的字段有一个是 over_time.rowtime 使用 sql：select tumble(over_time,interval '1' second),over_time from kafka_source group by over_time,tumble(over_time,interval '1' second) 会报错。ClassCastException 如果 将 group by 后面的子句交换位置， group by

rowtime/proctime

在执行 SQL 的时候遇见如下情况： streamTableEnvironment.registerDataStream（）注册的字段有一个是 over_time.rowtime 使用 sql：select tumble(over_time,interval '1' second),over_time from kafka_source group by over_time,tumble(over_time,interval '1' second) 会报错。ClassCastException 如果 将 group by 后面的子句交换位置， group by

請問在 Flink SQL 上能不能指定 WHERE 裡的判斷式的執行順序？

Hi, 想請問是否有辦法在 Flink SQL 上指明 WHERE 裡的判斷式的執行順序，來做到一些特定情況下的 查詢優化？ 舉例來說，在下面的 SQL，假如有個很耗時的 UDF 需要每次都去查詢資料庫。在這樣的狀況下， 如果可以確保優先執行 `!user.is_robot` 的判斷，再去執行後面的 UDF 的話，就能減少許多的資料 庫查詢。因為那些在 `user.is_robot` 裡得到 `true` 的數據就可以提早被丟棄，而不用去執行後面 較為花費時間的 UDF 了。 select * from users where !user.is_robot and

RestClusterClient

您好，flink集群关闭的情况下，运行任务会有报错信息 Could not retrieve the execution result，但是在 flink 集群正常运行的情况下，偶然会报 Could not retrieve the execution result，通过查询了解到，flink 通过 RestClusterClient类 将任务提交至 jobmanager，如果 detached 为 false，会采用 CompletableFuture 的 thenCompose 方法，在获取结果时，会有异常的捕获，如下： final CompletableFuture

AsyncIO 用Redis做缓存

需求: 事实表实时Join Kudu中的维度表，用来补全维度。 为加快查询速度，先从Kudu中查询数据，查询到数据后放入Redis缓存，下次查询先从Redis中取，取不到再从Kudu中查。 遇到的问题: 1、不用Redis缓存，checkpoint很快，效率很高。 2、用Redis缓存，用Jedis，但不用连接池，效率很低。 3、用Redis缓存，用Redis连接池，效率更低。 请教下: 1、从Kudu中取数据，不用缓存可以吗。 2、在AsyncIO中，用lettuce异步客户端，和flink netty不兼容。 3、在AsyncIO中，用Jedis连接池，flink

Re: AsyncIO 用Redis做缓存

这种Join场景，用上缓存后，理论上应该更快，但为啥会变慢呢。 王佩 于2019年8月6日周二 下午10:09写道： > 需求: 事实表实时Join Kudu中的维度表，用来补全维度。 > > 为加快查询速度，先从Kudu中查询数据，查询到数据后放入Redis缓存，下次查询先从Redis中取，取不到再从Kudu中查。 > > 遇到的问题: > 1、不用Redis缓存，checkpoint很快，效率很高。 > 2、用Redis缓存，用Jedis，但不用连接池，效率很低。 > 3、用Redis缓存，用Redis连接池，效率更低。 > > 请教下: >

Re: AsyncIO 用Redis做缓存

你好， 可以用lettuce做异步客户端，排除lettuce的netty依赖，用flink的netty，就可以了集成lettuce了 王佩 于2019年8月6日 周二22:11写道： > 这种Join场景，用上缓存后，理论上应该更快，但为啥会变慢呢。 > > 王佩 于2019年8月6日周二 下午10:09写道： > > > 需求: 事实表实时Join Kudu中的维度表，用来补全维度。 > > > > 为加快查询速度，先从Kudu中查询数据，查询到数据后放入Redis缓存，下次查询先从Redis中取，取不到再从Kudu中查。 > > > > 遇到的问题: > >

flink 动态表输出问题

各位Flink社区大佬， 你们好: 请教一下: 问题是 flink sql实现动态表之后,想基于动态表进行查询定时输出, 各位大神有什么实现的思路吗感激不尽. SQL 1: tableEnv.sqlQuery("select date_id, id, latest(status, utime) as status, latest(user_id, utime) as user_id from waybillAppendTable group by date_id, id"); 查询SQL2:tableEnv.sqlQuery("select date_id, user_id,

有一些TaskManager的slot不可用，尽管没有任务正在运行

各位大佬： 有对这种情况比较了解的吗？任务结束后，一些slot并没有释放掉。 如图所示： pengcheng...@bonc.com.cn

Re: Re: CsvTableSink 目录没有写入具体的数据

我接入了一个 RocketMQ 的流作为输入。 DataStream> ds = env.addSource(new RocketMQSource( System.out.println(res); return res; } }); tableEnv.registerDataStream("t_pick_task", ds, "pick_task_id,

Re: Re: CsvTableSink 目录没有写入具体的数据

没数据是因为没有trigger执行, 参考sample code from doc( https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/common.html ) // get a StreamTableEnvironment, works for BatchTableEnvironment equivalentlyStreamTableEnvironment tableEnv = StreamTableEnvironment.create(env); // create a

如何讓兩個 SQL 使用相同的 KafkaTableSource

Hi 我在我的 flink job 中透過 `flinkTableEnv.connect(new Kafka()...).registerTableSource(...)` 註冊了 一張 kafka table。但從文件上我才知道 SQL 只會在特定的條件下才會真正的轉為 DataStream，比 如說呼叫了Table#toRetractStream`。 因為如此，我發現當我嘗試在同一個 flink job 中使用了不同的 SQL 時，他們會同時產生各自的 kafka source operator。從 flink 的角度來說可能不是什麼大問題，各自獨立的 operator 會各自管理

Re: Re: CsvTableSink 目录没有写入具体的数据

抱歉，是我搞错了。 实际上是写入数据的。我在 windows 下做测试，刷新下文件的大小始终是 0 ， 只有编辑看下那个文件显示的文件大小才会变更。 wangl...@geekplus.com.cn Sender: Alec Chen Send Time: 2019-08-09 10:17 Receiver: user-zh Subject: Re: Re: CsvTableSink 目录没有写入具体的数据 没数据是因为没有trigger执行, 参考sample code from doc(

Re: Flink sql join问题

可以试下最新flink 1.9 blink planner的firstRow/lastRow优化[1]能否满足你的需求，目前的限制是只能基于procTime来去重。 * e.g. * 1. {{{ * SELECT a, b, c FROM ( * SELECT a, b, c, proctime, * ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime ASC) as row_num * FROM MyTable * ) WHERE row_num <= 1 * }}} will be converted

Re: 如何获取Flink table api/sql code gen 代码

Currently Flink DO NOT provides a direct way to get code gen code. But there are indirect ways to try. 1) debug in IDE Flink use Janino to compile all code gen code, and there is a single entry point [1]

Re: flink-1.8.1 yarn per job模式使用

刚发现 user-zh 是有 archive[1] 的，上面提到过的跟你类似的问题是这个 thread[2]。 Best, tison. [1] https://lists.apache.org/list.html?user-zh@flink.apache.org [2] https://lists.apache.org/thread.html/061d8e48b091b27e797975880c193838f2c37894c2a90aa6a6e83d36@%3Cuser-zh.flink.apache.org%3E Yuhuan Li 于2019年8月7日周三 下午7:57写道：

Re: 恢复savepoint，除了命令行，能通过代码获取吗?

Hi 中锋 恐怕不能通过代码来回复savepoint，目前一共只有有两个地方可以传入savepoint path，分别是 1. CliFrontendParser#createSavepointRestoreSettings [1] 2. JarRunHandler#getSavepointRestoreSettings [2] 分别对应命令行，网页（REST）提交，没办法在代码里面进行恢复请求，其实我理解REST或者网页提交应该也满足你们的需求。 [1]

回复： flink 结合canal统计订单gmv

没看到附件啊 | | 1101300123 | | hdxg1101300...@163.com | 签名由网易邮箱大师定制 在2019年8月9日 10:14，王飞 写道： hi 如上图 是一个kafka流 将一个canal解析的binlog 数据解析成一个arraybuffer的流数据 我想把arraybuffer解析成一个个order 然后再转换为table api进行产品维度的gmv统计， 原始解析代码 和 数据见附件 谢谢

Re: 有一些TaskManager的slot不可用，尽管没有任务正在运行

Hi， 邮件中的图片显示不出来。Flink邮件列表的图片附件是有点问题的，如果是截图最好上传到其他地方然后把链接贴出来。 Thank you~ Xintong Song On Fri, Aug 9, 2019 at 10:06 AM pengcheng...@bonc.com.cn < pengcheng...@bonc.com.cn> wrote: > 各位大佬： > > 有对这种情况比较了解的吗？任务结束后，一些slot并没有释放掉。 > > > 如图所示： > > > > > -- >

关于外部请求支持sql的方式。

各位commit： 我们现在遇到这么一个需求，请求帮助。 我们需要开发一个web界面支持写sql，做配置，然后提交job的功能。 我们浏览了最新版的flink1.8.1 里面有sql client的客户端，但是也仅仅支持在jobmanage上启动该界面去操作(我是这样理解的，并不支持跨机器通信) 请问下，我们需要的功能，在最新版的功能列表里，是否有相关功能模块，我们可以直接拿来用，然后做一个上层分组即可？ 请给点建议 白京京 18201047119 Tel:

恢复savepoint，除了命令行，能通过代码获取吗?

如题，restore savepoint，除了run flink -s savepointpath之外，能通过代码恢复吗，因为公司集群没法输入命令行。如果可以的话，能给个小demo，或者API也可以 谢谢。 Best， Rio Liu， 刘中锋

答复: 恢复savepoint，除了命令行，能通过代码获取吗?

你好， 可以通过flink的restFul去调用保存savepoint 发件人: liu zhongfeng 发送时间: 2019年8月9日 20:28 收件人: user-zh@flink.apache.org 主题: 恢复savepoint，除了命令行，能通过代码获取吗? 如题，restore savepoint，除了run flink -s savepointpath之外，能通过代码恢复吗，因为公司集群没法输入命令行。如果可以的话，能给个小demo，或者API也可以 谢谢。 Best， Rio Liu， 刘中锋

Flink命令提交任务时是否支持配置文件与任务jar包分离

各位社区大佬： 请问使用Flink命令提交任务时是否支持配置文件与任务jar包分离。 比如我的任务自身有个配置文件job.yaml,目前该配置是打到jar包中随任务提交的，可是有时候只是要调整下配置代码没改动也要重新打包发到环境上，感觉这样不灵活， 所以我想问下能不能单独把配置文件(可能是多个文件)放到一个目录下，然后提交任务的时候指定配置文件或者是配置目录。 类似jobManager把这些配置分发到TaskManager的classPath下这样的逻辑，这样就不用改下配置也要重新打包发到环境上了。

Re:Re: Re: Flink RocksDBStateBackend 问题

唐老师您好： 我这里指的是checkpoint时存储数据的file system，这里我用的是HDFS。 按照老师的说法，我可不可以这样理解(在Flink on yarn 以及 使用 RocksDBStateBackend 的场景下)： 1.做增量checkpoint的时候，taskmanager默认异步的将数据写入rocksdb和hdfs中（数据是相同的）。

答复: jobmanager 日志异常

你好， 谢谢！已经找到原因了 发件人: Biao Liu 发送时间: 2019年8月6日 13:55 收件人: user-zh 主题: Re: jobmanager 日志异常 你好， > org.apache.flink.runtime.entrypoint.ClusterEntrypoint - RECEIVED > SIGNAL 15: SIGTERM. Shutting down as requested. 这是收到了 signal 15 了 [1]，Wong 说得对，搜一下 yarn node manager 或者 yarn

Re: Re:Re: Re: Flink RocksDBStateBackend 问题

1. 你对增量checkpoint的理解以及taskmanager和RocksDB之间的关系理解不太对。RocksDBKeyedStateBackend使用RocksDB存储state数据，可以理解成其是taskmanager的一部分，实际上是单机的概念。增量checkpoint的时候，RocksDB会在同步阶段将其所有数据刷写到磁盘上，Flink框架选择之前没有上传的sst文件，异步上传到HDFS。如果没有开启local recovery，那么新启动的taskmanager会从hdfs上下载全量的数据文件进行恢复。 2.

submit jobGraph error on server side

向session cluster 提交job 出错，麻烦各位老师帮忙看下，给点排查提示 THX~ 环境: blink 1.8.0 用docker 方式启动的flink session cluster，flink 集群独立，我从集群外的一个docker 节点提交job（该节点的flink-conf.yaml 配置与flink 集群内的配置一致） -- 报错信息: The program finished

Re: submit jobGraph error on server side

问题是 Ask timed out on [Actor[akka://flink/user/dispatcher#-273192824]] after [1 ms]. Sender[null] sent message of type "org.apache.flink.runtime.rpc. messages.LocalFencedMessage". 也就是 submit job 的时候在请求 Dispatcher 的时候 akka ask timeout 了，可以检查一下配置的地址和端口是否正确，或者贴出你的相关配置。 Best, tison. 王智

Re: Flink RocksDBStateBackend 问题

FileSystem 我记得是存储的大小是不能超过tm的内存还是jm的内存，而rocksdb上存储的数据是可以无限的，不过相对来说， FileSystem的吞吐就会比rocksdb会高 lvwenyuan 于2019年8月6日周二 上午11:39写道： > 请教各位： >RocksDBStateBackend > 中，rocksdb上存储的内如和FileSystem上存储的数据内容是一样的？如果不一样，那么分别是什么呢？感谢回答 > > > >

Re: Re: Flink RocksDBStateBackend 问题

@lvwenyuan 首先需要明确的一点是，你这里的“FileSystem”指的究竟是checkpoint时存储数据的file system，还是FsStateBackend，建议下次提问前可以把需要咨询的内容表述清楚一些。 * 如果指的是存储checkpoint数据的远程file system，在incremental

Re: jobmanager 日志异常

你好， > org.apache.flink.runtime.entrypoint.ClusterEntrypoint - RECEIVED > SIGNAL 15: SIGTERM. Shutting down as requested. 这是收到了 signal 15 了 [1]，Wong 说得对，搜一下 yarn node manager 或者 yarn resource manager 的 log 1. https://access.redhat.com/solutions/737033 Thanks, Biao /'bɪ.aʊ/ On Tue,

Flink sql join问题

Hi all， 请问用Flink sql做双流join。如果希望两个流都只保存每个key的最新的数据，这样相当于每次join都只输出最新的一条记录。请问这种场景sql支持吗 thanks

jobmanager 日志异常

大家好： 我的flink是部署在yarn上左session，今天早上jobmanager自动退出了，然后yarn把他重新拉起了，导致里面跑的job重新启动了，但是我查看日志，看到jobmanager的日志没有任何异常，同时jobmanager也没有长时间的full gc和频繁的gc，以下是jobmanager的日志： 就是在06:44分的是偶，日志上标记了收收到停止请求，然后jobmanager直接停止了...请问是由于什么原因导致的呢？ 2019-08-06 06:43:58,891 INFO >

Re: Re: Flink RocksDBStateBackend 问题

你说的是memsystem的状态数据存在jm内存中的filesystem是存到文件系统上的 athlon...@gmail.com 发件人： 戴嘉诚 发送时间： 2019-08-06 11:42 收件人： user-zh 主题： Re: Flink RocksDBStateBackend 问题 FileSystem 我记得是存储的大小是不能超过tm的内存还是jm的内存，而rocksdb上存储的数据是可以无限的，不过相对来说， FileSystem的吞吐就会比rocksdb会高 lvwenyuan 于2019年8月6日周二 上午11:39写道： > 请教各位： >

Re: jobmanager 日志异常

Hi， 可以查看一下jobmanager所在节点的yarn log，搜索一下对应的container为什么被kill； Regards On 2019/8/6, 11:40 AM, "戴嘉诚" wrote: 大家好： 我的flink是部署在yarn上左session，今天早上jobmanager自动退出了，然后yarn把他重新拉起了，导致里面跑的job重新启动了，但是我查看日志，看到jobmanager的日志没有任何异常，同时jobmanager也没有长时间的full gc和频繁的gc，以下是jobmanager的日志：

任务异常捕获疑问

各位大佬好， 请问是否可以像下面这样捕获job异常, 如果不行应当怎么处理，或者有没有更好的方式？ 谢谢... try { env.execute("jobName"); } catch (Throwable e) { *//log..报警* }

Re: need help

你好， 异常里可以看出 AskTimeoutException, 可以调整两个参数 akka.ask.timeout 和 web.timeout 再试一下，默认值如下 akka.ask.timeout: 10 s web.timeout: 1 PS: 搜 “AskTimeoutException Flink” 可以搜到很多相关答案 Thanks, Biao /'bɪ.aʊ/ On Thu, Aug 8, 2019 at 7:33 PM 陈某 wrote: > > > -- Forwarded message - > 发件人： 陈某

Re: CsvTableSink 目录没有写入具体的数据

完整代码发一下 wangl...@geekplus.com.cn 于2019年8月8日周四 下午7:37写道： > > 我按官网上的 > https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/sql.html#specifying-a-query > 例子写的代码 > 但运行后 CsvTableSink 指定的目录只生成了空文件，没有具体的内容，这是为什么呢？ > > > > wangl...@geekplus.com.cn >

Fwd: need help

-- Forwarded message - 发件人： 陈某 Date: 2019年8月8日周四 下午7:25 Subject: need help To: 你好，我是一个刚接触flink的新手，在搭建完flink on yarn集群后，依次启动zookeeper，hadoop，yarn，flkink集群，并提交认识到yarn上时运行遇到问题，网上搜索相关问题，暂未找到解决方式，希望能得到帮助，谢谢。 使用的运行指令为： [root@flink01 logs]# flink run -m yarn-cluster

Re: flink 结合canal统计订单gmv

Hi, 截图无法显示, 不知道你是使用FlinkSQL还是DataStreamAPI实现, 前者可以参考UDTF, 后者可以参考FlatMap "Takes one element and produces zero, one, or more elements. A flatmap function that splits sentences to words" https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/ 王飞 于2019年8月8日周四 下午4:53写道： > hi 你好

CsvTableSink 目录没有写入具体的数据

我按官网上的 https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/sql.html#specifying-a-query 例子写的代码 但运行后 CsvTableSink 指定的目录只生成了空文件，没有具体的内容，这是为什么呢？ wangl...@geekplus.com.cn

Re:使用StreamingFileSink或者BucketingSink写入HDFS 问题请教

请问您第一个问题是如何解决呢 在 2019-07-18 11:15:49，"九思" <1048095...@qq.com> 写道： 请教老师，使用StreamingFileSink或者BucketingSink写入HDFS。 1、在本地，当掐掉程序的时候，文件状态还是in-process状态，不会转为正式文件，这个要怎么处理呢？ 2、重启程序开始后，编号又从0开始，而不是从之前的编号继续。看了源码，是有去获取之前的编号。但是我断点查了，没获取到，是什么原因呢？

flink 1.8 on yarn????RM????

?? flink1.8 on yarn??yarn??ResourceManagerHA??RM??2ActiveStandbystandby??8032RM PS??flink1.7 on yarnyarn??hadoop 2019-07-23

回复：flink web ui 添加 subtask index 信息

回复：Flink 的 log 文件夹下产生了 44G 日志

Re: Flink 的 log 文件夹下产生了 44G 日志

Hi Henry, 邮件列表貌似不能支持直接贴图，所以无法理解“里面不停的在产生 error 信息，但是也没有看到具体哪里报错”是什么意思。尝试贴图到第三方，然后贴链接上来？不知道还有没有更好的办法 To zhisheng2018, 你为什么老回空邮件？ zhisheng2...@gmail.com 于2019年7月18日周四 上午12:49写道： >

????StreamingFileSink????BucketingSink????HDFS ????????

??StreamingFileSinkBucketingSinkHDFS?? 1in-process?? 2??0??

Re: Flink 的 log 文件夹下产生了 44G 日志

尴尬了，之前回复的邮件难道都是空白，Henry 你可以把报错信息放到哪个博客里面，然后再这里提供个链接，邮件看不到你的截图错误信息，你可以先根据错误信息定位问题所在，把错误的问题先解决掉，然后再来合理的配置重启策略。 Biao Liu 于2019年7月18日周四 上午1:15写道： > Hi Henry, > > 邮件列表貌似不能支持直接贴图，所以无法理解“里面不停的在产生 error > 信息，但是也没有看到具体哪里报错”是什么意思。尝试贴图到第三方，然后贴链接上来？不知道还有没有更好的办法 > > To zhisheng2018, 你为什么老回空邮件？ > > >

Re:Re: Flink 的 log 文件夹下产生了 44G 日志

明白啦，谢谢哈。 我以为会显示图片呢。 我贴链接下次。 在 2019-07-18 12:03:57，"zhisheng" 写道： >尴尬了，之前回复的邮件难道都是空白，Henry >你可以把报错信息放到哪个博客里面，然后再这里提供个链接，邮件看不到你的截图错误信息，你可以先根据错误信息定位问题所在，把错误的问题先解决掉，然后再来合理的配置重启策略。 > >Biao Liu 于2019年7月18日周四 上午1:15写道： > >> Hi Henry, >> >> 邮件列表貌似不能支持直接贴图，所以无法理解“里面不停的在产生 error >>

checkpoint 文件夹Chk-no 下面文件个数是能计算出来的吗？

state_checkpoints_dir/2d93ffacbddcf363b960317816566552/chk-2903/1e95606a-8f70-4876-ad6f-95e5cc38af86 state_checkpoints_dir/2d93ffacbddcf363b960317816566552/chk-2903/2a012214-734a-4c2b-804b-d96f4f3dddf8

Re: checkpoint 文件夹Chk-no 下面文件个数是能计算出来的吗？

Hi A1: chk-x文件下面的文件个数是跟operator个数并行度是有关系的，主要是operator state的文件。对于checkpoint场景，_metadata只是元数据，真实的operator数据都是在其他文件内。 A2: 不可以将这些文件合并在一起。因为_metadata内主要记录了文件路径，如果合并的话，找不到原始路径会有问题，无法从checkpoint进行restore 祝好 唐云 From: 陈冬林 <874269...@qq.com> Sent: Thursday, July 18,

Fwd: checkpoint 文件夹Chk-no 下面文件个数是能计算出来的吗？

谢谢您的解答， 那些文件的数量是只和operator的并行度相关吗？是不是还有key 的个数等相关？有没有具体的公式呢？我没有在源码里找到这块的逻辑 还有一个最重要的问题，这些文件即然不能合并，state小文件合并指的是那些文件呢？ 祝安 Andrew > 下面是被转发的邮件： > > 发件人: Yun Tang > 主题: 回复： checkpoint 文件夹Chk-no 下面文件个数是能计算出来的吗？ > 日期: 2019年7月18日 GMT+8 下午3:24:57 > 收件人: "user-zh@flink.apache.org" >

Re: checkpoint 文件夹Chk-no 下面文件个数是能计算出来的吗？

Hi 源码部分可以参考[1] DefaultOperatorStateBackendSnapshotStrategy 执行完成的时候，每个operator state backend 都只会产生至多一个文件。 state小文件合并，你指的应该是FLINK-11937 吧，这里的所谓合并是每个rocksDB state backend创建checkpoint的时候，在一定阈值内，若干sst文件的序列化结果都写到一个文件内。由于keyed

Fwd: checkpoint 文件夹Chk-no 下面文件个数是能计算出来的吗？

好的，非常感谢您的解答。 > 下面是被转发的邮件： > > 发件人: Yun Tang > 主题: 回复： checkpoint 文件夹Chk-no 下面文件个数是能计算出来的吗？ > 日期: 2019年7月18日 GMT+8 下午4:06:59 > 收件人: "user-zh@flink.apache.org" > 回复－收件人: user-zh@flink.apache.org > > Hi > > 源码部分可以参考[1] DefaultOperatorStateBackendSnapshotStrategy 执行完成的时候，每个operator >

Re: checkpoint 文件夹Chk-no 下面文件个数是能计算出来的吗？

hi 首先先要确定是否是大量创造文件导致你的namenode RPC相应堆积多，RPC请求有很多种，例如每个task创建checkpoint目录也是会向namenode发送大量RPC请求的（参见 [https://issues.apache.org/jira/browse/FLINK-11696]）；也有可能是你的checkpoint interval太小，导致文件不断被创建和删除（subsume old checkpoint），先找到NN压力大的root cause吧。

请问这些名词，在翻译 Glossary 时，有必要翻译成中文吗？

Hi 各位, 晚上好！ 以下名词在翻译 Glossary 章节时，有必要翻译成中文吗？名词列表如下： Flink Application Cluster Flink Cluster Event ExecutionGraph Function Instance Flink Job JobGraph Flink JobManager Logical Graph Managed State Flink Master Operator Operator Chain Partition Physical Graph

Re: 请问这些名词，在翻译 Glossary 时，有必要翻译成中文吗？

没有可援引的通译出处建议专有名词不要翻译。Glossary 的解释部分可以解释得详尽一点，上面像 record task 这些有比较普遍共识的还有商讨空间，像 transformation "operator chain" 强行翻译很可能是懂的人本来就看得懂，不懂的人看了还是不懂。现在不翻译在有通译之后可以改，先根据个人喜好翻译了以后就不好改了。 一点拙见。 Best, tison. highfei2011 于2019年7月18日周四 下午11:35写道： > Hi 各位, > 晚上好！ > 以下名词在翻译 Glossary 章节时，有必要翻译成中文吗？名词列表如下：

Re:请问这些名词，在翻译 Glossary 时，有必要翻译成中文吗？

Hi，Zili Chen: 早上好，你讲的没错，谢谢。另外我发现，Glossary 英文文档中没有 Slot 和 Parallelism 的说明，建议添加。这样可以方便初学者和用户的学习和使用！ 祝好 Original Message Subject: Re: 请问这些名词，在翻译 Glossary 时，有必要翻译成中文吗？ From: Zili Chen To: user-zh@flink.apache.org CC: 没有可援引的通译出处建议专有名词不要翻译。Glossary 的解释部分可以解释得详尽一点，上面像

Re: 请问这些名词，在翻译 Glossary 时，有必要翻译成中文吗？

Hi, 欢迎有 PR 后同步到这个 thread 上 :-) Best, tison. highfei2011 于2019年7月19日周五 上午8:34写道： > Hi，Zili Chen: > 早上好，你讲的没错，谢谢。另外我发现，Glossary 英文文档中没有 Slot 和 Parallelism > 的说明，建议添加。这样可以方便初学者和用户的学习和使用！ > > 祝好 > > > > Original Message > Subject: Re: 请问这些名词，在翻译 Glossary 时，有必要翻译成中文吗？ >

Flink 的 log 文件夹下产生了 44G 日志

大家好，之前那个报错图片大家没看到，重新弄一下。 报错图片链接： https://img-blog.csdnimg.cn/20190719092540880.png https://img-blog.csdnimg.cn/20190719092848500.png 我看报错的原因是，我这里Source用的是ActiveMQ，从昨天早上9点开始运行Flink任务接收消息，到今天早上8点都很正常。然后在今天早上8点4分的时候开始猛报错flink往log文件夹下写日志。第二个图是报错开始，显示ActiveMQ好像超时，然后就是消费者关闭一直猛写log。

Re: checkpoint 文件夹Chk-no 下面文件个数是能计算出来的吗？

Hi [https://issues.apache.org/jira/browse/FLINK-11696] 里面目前的PR是我们的生产代码，你可以用。但是你现在的问题的root cause不是这个，而是创建文件和删除文件的请求太多了。可以统计一下目前你们几百个作业的checkpoint interval，一般而言3~5min的间隔就完全足够了，没必要将interval调整得太小，这是一个影响你们整个集群使用的配置，必要时需要告知用户正确的配置。

Re: Flink 的 log 文件夹下产生了 44G 日志

Hi Henry, 这个 source 看起来不像是 Flink 提供的 source，应该是 source 本身实现的问题。你可能需要修改 source 的源码让它出错后关闭或者进行其它处理... Henry 于2019年7月19日周五 上午9:31写道： > 大家好，之前那个报错图片大家没看到，重新弄一下。 > 报错图片链接： > https://img-blog.csdnimg.cn/20190719092540880.png > https://img-blog.csdnimg.cn/20190719092848500.png > > >

Re: 请问这些名词，在翻译 Glossary 时，有必要翻译成中文吗？

Hi highfei, Thanks for bringing up this discussion. I would suggest to move the discussion to the Glossary translation JIRA FLINK-13037 . Thanks, Jark On Fri, 19 Jul 2019 at 09:00, Zili Chen wrote: > Hi, > > 欢迎有 PR 后同步到这个 thread 上 :-) > >

Re: 请问这些名词，在翻译 Glossary 时，有必要翻译成中文吗？

Hi, Just find the Glossary translation PR is created [1]. Let's move the discussion there. [1]. https://github.com/apache/flink/pull/9173 On Fri, 19 Jul 2019 at 11:22, Jark Wu wrote: > Hi highfei, > > Thanks for bringing up this discussion. I would suggest to move the > discussion to the

could rest api : /jobs/:jobid/yarn-cancel trigger the savepoint?

Hi community, I have a question is that could rest api : /jobs/:jobid/yarn-cancel trigger the savepoint? I saw the fink src code, and I find it didn't trigger the savepoint, is it right? Thank you to reply .

Re:Re: 请问这些名词，在翻译 Glossary 时，有必要翻译成中文吗？

Hi, Jark Wu ,Thanks I am discussing in there . 在 2019-07-19 11:22:53，"Jark Wu" 写道： >Hi, > >Just find the Glossary translation PR is created [1]. Let's move the >discussion there. > >[1]. https://github.com/apache/flink/pull/9173 > >On Fri, 19 Jul 2019 at 11:22, Jark Wu wrote: > >> Hi

多滑动窗口问题

您好: 请问, 在进行流计算时, source相同, 处理逻辑相同, 但要计算不同的滑动时间窗口, 比如 每分钟统计最近 5m,15m,30m 以及 每15分钟计算, 1h, 3h , 12h的数据 除去每种窗口写一个程序外, 有其他更加便捷的解决方式吗 ? 谢谢

Re:Re: Flink 的 log 文件夹下产生了 44G 日志

你好，谢谢！是的，这个Source是用JMS实现的自定义Source。目前还在查原因，但是怎么能够让Flink不这样爆炸写log日志呢？20分钟就能写满磁盘，写了40G多。 在 2019-07-19 11:11:37，"Caizhi Weng" 写道： >Hi Henry, > >这个 source 看起来不像是 Flink 提供的 source，应该是 source 本身实现的问题。你可能需要修改 source >的源码让它出错后关闭或者进行其它处理... > >Henry 于2019年7月19日周五 上午9:31写道： > >>

Re: Re: Flink 的 log 文件夹下产生了 44G 日志

Hi Henry 你的意思是不想让 Flink 写 log 吗？那只能通过 `log4j.rootLogger=OFF` (log4j) 或者 ` ` (logback) 把 log 关掉，或者把 log 等级设成更高的 FATAL... 但我感觉问题还是自定义的 source 里写 log 的时候死循环了... Henry 于2019年7月19日周五 下午2:20写道： > > > > 你好，谢谢！是的，这个Source是用JMS实现的自定义Source。目前还在查原因，但是怎么能够让Flink不这样爆炸写log日志呢？20分钟就能写满磁盘，写了40G多。 > > > > >

Re: Re: Flink 的 log 文件夹下产生了 44G 日志

最根本的解法当然是去掉打日志的地方，这 source 不是 Flink 内置的，Flink 当然不能控制你们自定义 source 的行为。 你可以考虑自己改一下 log4j.properties，手动关掉这个 logger, Flink 内置的 log4j.properties 里有 example，参考着改一下 log4j.logger.org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline=ERROR, file 改成

Re: Re: Re: Flink 的 log 文件夹下产生了 44G 日志

Hi Henry, LOG.error(e.getLocalizedMessage()); running = true; 这里写错了吧，应该是 running = false; Henry 于2019年7月19日周五 下午4:04写道： > > > 谢谢你的帮助哈！ 我也是觉得 source 里的问题，但是呢，木有找到错误的地方。下面这个是我那个自定义的 source > 代码，但是里面没有写log里报的哪个错的提示。 > package com.JavaCustoms; > import

Flink 1.8 run参数不一样

之前下载的Flink 1.8，运行bin/flink run --help，会有 yarn-cluster 的一些参数，如下: Options for yarn-cluster mode: -d,--detachedIf present, runs the job in detached mode -m,--jobmanager Address of the JobManager (master) to

Re: Flink 1.8 run参数不一样

你好，可以查看下 log/ 目录下的相关日志有没有这样一段 2019-07-24 09:34:36,507 WARN org.apache.flink.client.cli.CliFrontend - Could not load CLI class org.apache.flink.yarn.cli.FlinkYarnSessionCli. java.lang.NoClassDefFoundError: org/apache/hadoop/yarn/exceptions/YarnException at

请教Flink SQL watermark遇到未来时间的处理问题

各位Flink社区大佬， 您好！ 我使用Flink SQL (Flink 1.8.0)进行一些聚合计算，消费的是Kafka数据，使用的是EventTime，但是有时候，偶然会出现rowtime字段来了一条未来时间的数据（可能是上送的数据时区导致），这样Watermark会直接推到了未来某个时间点，导致这笔错误数据到达后的数据，到未来时间点之间的数据会被丢弃。 这个问题根本确实是业务方面的问题，但是我们还是希望有一些方案应对这种异常情况。 目前，我们这边处理的方法是：

Re: Flink 1.8 run参数不一样

问题解决了，非常感谢！ 解决流程： 1、确实在log/下找到了Could not load CLI class org.apache.flink.yarn.cli.FlinkYarnSessionCli.异常 2、设置 export HADOOP_CONF_DIR=`hadoop classpath` 3、重新运行 bin/flink run --help ，出现了`Options for yarn-cluster mode` 选项 感谢大佬！❤❤❤ Zili Chen 于2019年7月24日周三 上午9:51写道： > 你好，可以查看下 log/

Re: 请教Flink SQL watermark遇到未来时间的处理问题

hi，仲尼： 通常这种时间超前的数据是由于你机器的时间有问题（未对齐），然后采集上来的数据使用的那个时间可能就会比当前时间超前了（大了），你可以有下面解决方法： 1、在 Flink 从 Kafka 中消费数据后就进行 filter 部分这种数据（可以获取到时间后和当前时间相比一下，如果超前或者超前多久就把这条数据丢掉，之前我自己项目也有遇到过这种数据问题，设置的超前 5 分钟以上的数据就丢失），就不让进入后面生成水印，这样就不会导致因为水印过大而导致你后面的问题 2、在生成水印的地方做判断，如果采集上来的数据的时间远大于当前时间（比如超过 5

Re: 获取每个Operator输入输出的数据条数

可以自己加状态统计 > 在 2019年7月16日，下午4:34，王佩 写道： > > 有如下Pipeline: > >Kafka Source=>Map=>Filter=>StreamingFileSink.forBulkFormat > > 希望实现: > > 每次写入文件时，拿到这个Pipeline当前批次每个Operator输入、输出的数据条数并写到Mysql。 > > 请教一下，可以实现吗。感谢。 > >

flink kafka相关任务执行出错 导致 taskmanager 退出

你好，flink 运行过程中出现以下异常，有何建议指导排除异常， 2019-07-23 10:52:01,420 INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version : 0.9.0.1 2019-07-23 10:52:01,420 INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId : 23c69d62a0cabf06 2019-07-23

Re: Flink checkpoint 并发问题

setMaxConcurrentCheckpoints 这个参数你设置过么? athlon...@gmail.com 发件人： 戴嘉诚 发送时间： 2019-07-25 18:07 收件人： user-zh 主题： Flink checkpoint 并发问题 大家好： 我这里有个Job，会经常不定时的挂掉重启，具体的异常日志为一下，大致是checkpoint的同步失败的

Re: Flink checkpoint 并发问题

而且,我印象中mapstate操作不是线程安全的,需要你自己加锁 athlon...@gmail.com 发件人： 戴嘉诚 发送时间： 2019-07-25 18:07 收件人： user-zh 主题： Flink checkpoint 并发问题 大家好： 我这里有个Job，会经常不定时的挂掉重启，具体的异常日志为一下，大致是checkpoint的同步失败的

Re: Flink checkpoint 并发问题

你好，这个参数我没有设置过，对于checkpoint，我只是设置了rockDB增量写入，其他都没有设置了 athlon...@gmail.com 于2019年7月25日周四 下午6:20写道： > setMaxConcurrentCheckpoints 这个参数你设置过么? > > > > athlon...@gmail.com > > 发件人： 戴嘉诚 > 发送时间： 2019-07-25 18:07 > 收件人： user-zh > 主题： Flink checkpoint 并发问题 > 大家好： > >

Re: Re: Flink checkpoint 并发问题

https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/stream/state/state.html If heap state backend is used with synchronous snapshotting, the global iterator keeps a copy of all keys while iterating because of its specific implementation which does not support concurrent

Re: Re: Flink checkpoint 并发问题

其实你可以不用自己删除.使用TTL设置短一些时间,试试 athlon...@gmail.com 发件人： 戴嘉诚 发送时间： 2019-07-25 18:24 收件人： user-zh 主题： Re: Flink checkpoint 并发问题 你好，这个参数我没有设置过，对于checkpoint，我只是设置了rockDB增量写入，其他都没有设置了 athlon...@gmail.com 于2019年7月25日周四 下午6:20写道： > setMaxConcurrentCheckpoints 这个参数你设置过么? > > > > athlon...@gmail.com

<    4   5   6   7   8   9   10   11   12   13   >