Re: flink sql 数据异常导致任务失败

你kafka里面的是json么？format是json么？ String resultCreateTableSql = createKafkaSourceSQL +" WITH ( " +" 'connector' = 'kafka' ," +" 'topic' = '" + kafkaTopic +"'," +" 'properties.bootstrap.servers' = '" + kafkaBootstrapServers +"'," +" 'properties.group.id' = '" + kafkaGroupId +"'," +" 'format' = '" + k

Re: flink sql 数据异常导致任务失败

https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/formats/json.html 官网里面写的很 清楚啊 在 2020/8/18 14:45, Jingsong Li 写道: Hi, 最新的版本(1.11+)已经有这个属性可以配置了： https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/formats/json.html#format-options Best, J

task传输数据时反序列化失败

Hi all, 请教一下反序列化的问题，我有一个KeyedCoProcessFunction，输入是log流和rule流。 数据流如下： logSource .connect(ruleSource) .keyby(...) .process(My KeyedCoProcessFunction<>) .keyby(...) .print() 其中CoProcess函数中有两个MapState分别做log缓存和rule缓存。 结构为Map> logState,Map> ruleState. T在实例化函数时确定，为MyLog类型。 运行时遇到了如下错误，看样子似乎是在下游算子反序列化数据时

Re: flink sql 数据异常导致任务失败

好吧。我是1.10. taochanglian 于2020年8月18日周二 下午3:03写道： > > https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/formats/json.html > > 官网里面写的很 清楚啊 > > 在 2020/8/18 14:45, Jingsong Li 写道: > > Hi, > > > > 最新的版本(1.11+)已经有这个属性可以配置了： > > > https://ci.apache.org/projects/flink/fli

Re: flink1.11任务资源分批

目前 flink 是不支持运行时扩缩容的。 如果要做的话，只能是从外部根据 metrics 判断反压，然后停掉之前的作业，修改并发后再重新提交作业，从之前的 checkpoint 恢复。 目前社区正在准备一项 declarative resource management 的工作，让 flink 作业的并发度自动适配可用资源。这项工作完成后，可以通过从外部调整可用资源数量来调整作业的并发度。 Thank you~ Xintong Song On Mon, Aug 17, 2020 at 11:34 AM Dream-底限 wrote: > hi、 > 请问如果想要flink

回复： 1.11 kafka producer 只往一个partition里写

我看了下，你应该用的是 flink-connector-kafka_2.12 这个依赖包吧 这个的确会出现你说的情况，发到一个分区。 建议使用flink-connector-kafka-{kafka版本} 例如flink-connector-kafka-0.8 -- 原始邮件 -- 发件人:

Re: flink 1.10.1 跑批任务 OutOfMemoryError: Metaspace

按你的描述，应该是存在类加载泄露的问题。也就是说，由于某些原因，导致之前作业加载的类，没能被释放掉，致使类元数据积累越来越多，metaspace 空间不足。 具体泄露的原因还是需要根据 dump 分析，通常是作业用到的第三方依赖导致的，这种情况 flink 是没法强行清除加载类的。 Thank you~ Xintong Song On Mon, Aug 17, 2020 at 6:38 PM 胡松 wrote: > hi all >    使用flink 1.10.1 每10分钟跑一个批任务，但是跑一天后重复复现报错 > 2020-08-15 19:32:59 > org.ap

flink DataStream scala api can not overloaded method aggregate

Hi all, 我在使用scala 开发streaming应用时，使用了AggregateFunction with a ProcessWindowFunction，flink 版本是1.11.1，但是idea报错： Error:(80, 8) overloaded method value aggregate with alternatives: 很奇怪为什么不能重载下面的方法 def aggregate[ACC: TypeInformation, V: TypeInformation, R: TypeInformation](preAggregator: Aggregate

flink 1.11 order by rowtime报错

小伙伴们 ： 大家好，请教个问题，执行order by SQL文时，为什么报下面的错误： SQL文：select order_id,product_id FROM kafka_order order by rowtime Thanks & Regards

能否考虑针对检查点和保存点设置不同的超时时间

出发点是，检查点超时失败啥的其实并不是很重要，高峰时间有时候就是会超时失败，但并不会明显反压，因此没关系。但是，有时候需要重启任务，用保存点，那么高峰时期就是无法生成保存点，然后任务失败自动从上一次检查点恢复。这导致本身高峰时期，重启在停的过程失败导致回滚了近10分（检查点周期）。 有一种思路是直接将超时设置更长时间，但这也不行，因为检查点本身是占据资源的，设置短超时就是不希望检查点占据过多资源，相当于超时完成不了就不要继续了。 但是保存点却是人工介入，需要去重启任务，可能是bug或者什么原因必须重启任务。但高峰时间按照正常设置的超时可能就是无法完成保存点。

flink DataStream scala api can not overloaded method aggregate

Hi all, 我在使用scala 开发streaming应用时，使用了AggregateFunction with a ProcessWindowFunction，flink 版本是1.11.1，但是idea报错： Error:(80, 8) overloaded method value aggregate with alternatives: 很奇怪为什么不能重载下面的方法 def aggregate[ACC: TypeInformation, V: TypeInformation, R: TypeInformation](preAggregator: Aggregate

flink集成到cdh

大家好，在网上找了个制作parcel的工具，flink1.9版本打好之后可以正常通过cm安装运行,但是1.10和1.11安装之后都是启动不了，请问大家有这方面的经验可以传授下吗，感激不尽！

FlinkSQL sink 到 kafka partition 规则是怎样的？

直接用 FlinkSQL 实现抽取字段、字段转换的功能。 INSERT INTO kafka_dwd_table SELECT a, b, fun(c) FROM kafka_ods_table kafka_dwd_table topic 里面的 record 是客户端依照特定的 partition 规则发送过去的 经过 上面的 FlinkSQL 操作会怎样做 partition 呢？ 谢谢， 王磊 wangl...@geekplus.com

Re: flink 1.11 order by rowtime报错

Hi： 你的图挂了，建议直接贴代码，图片大家是看不到的 发件人： 郑斌斌 发送时间： 2020-08-18 17:45 收件人： user-zh 主题： flink 1.11 order by rowtime报错 小伙伴们 ： 大家好，请教个问题，执行order by SQL文时，为什么报下面的错误： SQL文：select order_id,product_id FROM kafka_order order by rowtime Thanks & Regards

Re: flink 1.11 order by rowtime报错

没有看到错误，把代码贴出来把，是不是eventtime没有设置或者设置不正确 -- Sent from: http://apache-flink.147419.n8.nabble.com/

【bug】flink 1.11使用连接器时schema丢失DataType指定的物理信息

hello all: 我在为flink 1.11开发新的连接器时，发现了问题。 连接器的旧版本是支持flink1.9的，最近升级了flink 1.11后，test中，发现创建连接器需要声明schema，schema需要使用TableSchema信息，TableSchema包含的DataType，DataType指定物理类型后，在TableSourceSinkFactory中，获得的schema中，丢失了前面指定的物理类型。 这个问题影响了source、sink。导致了启动时，检查类型不能通过。 例如 DataTypes.DATE().bridgedTo(java.sql.Date.cla

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

个人觉得还是取舍的问题，我们现在用flink sql 做实时数仓，维度表暂时用mysql，与业务商定好更新事件后，配置flink sql jdbc的lookup.cache.ttl参数来设置刷新时间，不知道你项目中，是维表数据变更后，需要秒级关联到消息中？ -- Sent from: http://apache-flink.147419.n8.nabble.com/

hive Streaming Reading 无法分组统计

SELECT id, count(1) FROM hive_user_parquet GROUP BY id /*+ OPTIONS('streaming-source.enable'='true', 'streaming-source.consume-start-offset'='2020-08-18') */

Re: flink 1.11 order by rowtime报错

错误呢？没看到。把代码贴出来看一下，是不是processtime没有设置或者设置不对 -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: 【bug】flink 1.11使用连接器时schema丢失DataType指定的物理信息

补充图片链接 创建连接器 http://image.zhaojianyun.com/mweb/bug1.png TableSourceSinkFactory中的创建sink http://image.zhaojianyun.com/mweb/bug2.png TableSchema的运行时物理信息 http://image.zhaojianyun.com/mweb/bug3.png 2020年8月18日 下午10:09，赵 建云 mailto:zhaojianyu...@outlook.com>> 写道： hello all: 我在为flink 1.11开发新的连接器时，发现了问题

flink本地编译源码失败

各位大佬好： 在本地编译flink源码包的时候当编译到flink-runtime-web的时候报如下错误：请问这是需要什么单独的设置嘛？还没有修改任何东西，只是clone下来之后编译,编译命令mvn clean install -DskipTests [ERROR] Failed to execute goal com.github.eirslett:frontend-maven-plugin:1.6:install-node-and-npm (install node and npm) on project flink-runtime-web_2.11: Could

Re: flink 1.11 order by rowtime报错

贴代码。 看看是不是eventtime没有设置或者设置的不正确 > 在 2020年8月18日，下午5:45，郑斌斌 写道： > > 小伙伴们 ： >大家好，请教个问题，执行order by SQL文时，为什么报下面的错误： > SQL文：select order_id,product_id FROM kafka_order order by rowtime > > > Thanks & Regards

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

@Xintong Song dump driverParentFirstClassloaderDriverManager??registeredDrivers??ParentFirstClassloadermysql??driverHikariDataSource10?

Flink SQL血缘关系

哪位大佬知道，如果要做Flink SQL血缘关系是在sqlNode中拿表之间关系好，还是在Transformation 算子中拿血缘关系好 guaishushu1...@163.com

Re: FlinkSQL sink 到 kafka partition 规则是怎样的？

按我的理解，你描述错了吧，是kafka_ods_table里面的数据，是按照客户端自定义分区发送 过来的，接下来你希望处理完以后，还要保证kafka_dwd_table里面的数据，也需要保证数据按partition有序吧。因为你是把数据insert到kafka_dwd_table吧，你的描述有问题。 如果我理解的没有问题，那你看一下https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/kafka.html#sink-partitioner

1.11????????????????????Cannot have more than one execute() or executeAsyc() call in a single environment

hello!      flink??iedaCannot have more than one execute() or executeAsyc() call in a single environmentjob manager??     org.apache.flink.client.

Re: FlinkSQL sink 到 kafka partition 规则是怎样的？

按我的理解，你描述错了吧，是kafka_ods_table里面的数据，是按照客户端自定义分区发送 过来的，接下来你希望处理完以后，还要保证kafka_dwd_table里面的数据，也需要保证数据按partition有序吧。因为你是把数据insert到kafka_dwd_table吧，你的描述有问题。 如果我理解的没有问题，那你看一下https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/kafka.html#sink-partitioner ，最关键的就是sink.partition

flink sql 如何指定之前的checkpoint执行

hi，大家好，我遇到个问题，执行一个sql，作了checkpoint，现在sql中加了where条件，但是insert select 的值都没变，想指定之前的checkpoint继续执行，但是会报错，cannot map checkpoint state for operator *** to the new programe 是不是因为sql改变了，算子id就变了找不到了，这个有什么办法吗

Re: sql client 并行度问题

通过设置 table.exec.hive.infer-source-parallelism=false, 把hive source自动并发推导关闭， 然后设置table.exec.resource.default-parallelism修改并发度 18579099...@163.com <18579099...@163.com> 于2020年8月17日周一 下午5:51写道： > sql client 读取hive表，hive表中一共有21个文件需要读取，sql > client提示需要21个并行度，但是我slot并没有这么多。有什么办法可以把并行度改小？ > > > > 1857909

Re: flink 1.11 SQL idea调试无数据也无报错

> GROUP BY TUMBLE(ts, INTERVAL '1' HOUR) 是要1个小时的window结束才会出结果。 你可以通过把window缩小或者设置early-fire来提早看到数据 table.exec.emit.early-fire.enabled=true table.exec.emit.early-fire.delay=xx > 手动拿到那个executeSql的返回的TableResult，然后去 wait job finished 这个是为了防止本地ide里执行时executeSql执行结束后进程退出导致job也强制结束 DanielGu <610

Re: ScalarFunction 访问 state

看看deduplication语法[1] 是否满足你的需求 [1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/queries.html#deduplication forideal 于2020年8月17日周一 下午12:13写道： > Hi, > > 最近我有一个使用 Flink SQL 做简单的数据去重的需求，想使用 Flink 的 `ScalarFunction`，通过阅读 > API 发现 FunctionContext context 并不支持访问 st

Re: flink本地编译源码失败

Hi, 看那个报错是说下载不下来那个jar包，你要不要试试改一下maven的settings文件使用的源 Bset, Xingbo 魏烽 于2020年8月19日周三 上午12:47写道： > 各位大佬好： > > > 在本地编译flink源码包的时候当编译到flink-runtime-web的时候报如下错误：请问这是需要什么单独的设置嘛？还没有修改任何东西，只是clone下来之后编译,编译命令mvn > clean install -DskipTests > > > [ERROR] Failed to execute goal > com.github.eirslett:fronten

Re: 1.11版本，执行任务报错：Cannot have more than one execute() or executeAsyc() call in a single environment

你的这个报错应该是使用1.11里面新增的application mode来运行的吧 因为目前application mode不能支持在HA模式下，一个Flink cluster里运行多个任务，所以会报上面的错 原因是用户可能在main方法里面增加了if...else等来实现Job之间的依赖关系，这种情况在HA模式下恢复可能会有问题 所以暂时不支持了。非HA的时候所有任务都会重跑，那就没有这个限制了 Best, Yang Asahi Lee <978466...@qq.com> 于2020年8月19日周三 上午1:37写道： > hello! >      我的flink应用在ied

Re: flink本地编译源码失败

Hi： 这个问题我遇见过，就是网络连接超时了，编译的时候会从https://nodejs.org/dist/v10.9.0/node-v10.9.0-darwin-x64.tar.gz 下载这个tar文件，因为是外网的原因下载失败了，要么你手动下载这个文件放在对应的目录下，要么设置一个nodejs.org的代理 发件人： 魏烽 发送时间： 2020-08-18 23:17 收件人： user-zh 主题： flink本地编译源码失败 各位大佬好： 在本地编译flink源码包的时候当编译到flink-runtime-web的时候报如下错误：请问这是需要什么单独的设置嘛？

Flink1.11 hive catalog使用

hi 我这面在使用hivecatalog将kafka数据落地到hive，但现在有一个问题是，我不想保存kafka source table元数据（默认自动保存），通过创建视图或临时表的方式启动flink任务，这时候直接通过kafka source table表名直接引用报错，提示hive catalog内没这张表，这种情况我改怎么引用我的kafka未持久化表哪

Re: flink1.9.1用采用-d（分离模式提交）作业报错，但是不加-d是可以正常跑的

万分感谢！ 问题已经解决，确实是包的问题，我很傻的以为不加-d可以运行，那就跟包没关系。 所以说加不加-d，应该是调用不同包的不同方法吧？ - Best Wishes -- Sent from: http://apache-flink.147419.n8.nabble.com/

flink生产环境使用

这段时间开始引入flink在生产环境，主要想将sql用起来，有木有大佬推荐一下，在生产环境下，用yarn模式好还是standalone模式，我们spark等大数据都是yarn来管理 | | 引领 | | yrx73...@163.com | 签名由网易邮箱大师定制

Re: Flink1.11 hive catalog使用

可以把kafka的表保存在内置的in-memory catalog里，名字应该是default_catalog，这样就不会持久化了。或者用create temporary table的方式应该也可以。 On Wed, Aug 19, 2020 at 10:53 AM Dream-底限 wrote: > hi > 我这面在使用hivecatalog将kafka数据落地到hive，但现在有一个问题是，我不想保存kafka source > table元数据（默认自动保存），通过创建视图或临时表的方式启动flink任务，这时候直接通过kafka source > table表名直接引用报错，

flink1.12啥时候会出呢？

如题，有大概时间嘛。

Re: Flink1.11 hive catalog使用

hi 我先前用这种方式创建的表，在sql中直接使用的时候提示hivecatalog内无这张表，请问这张表使用的时候要怎么引用哪 >或者用create temporary table的方式应该也可以。 Rui Li 于2020年8月19日周三 上午11:11写道： > 可以把kafka的表保存在内置的in-memory catalog里，名字应该是default_catalog，这样就不会持久化了。或者用create > temporary table的方式应该也可以。 > > On Wed, Aug 19, 2020 at 10:53 AM Dream-底限 wrote: > > >

Re: flink1.12啥时候会出呢？

个人估计恐怕得年底，貌似九月冻结吧，然后测试修复什么的就差不多阳历年了。。。 赵一旦 于2020年8月19日周三 上午11:16写道： > 如题，有大概时间嘛。 >

Re: flink sql 如何指定之前的checkpoint执行

hi 可以参考这篇文章https://mp.weixin.qq.com/s/Vl6_GsGeG0dK84p9H2Ld0Q 在cancel的时候触发一个savepoint 修改完SQL从savepoint恢复任务 -- Sent from: http://apache-flink.147419.n8.nabble.com/

????intervaljoin????????????????jobmanager.execution.failover-strategy??region????????????????task cancel

??flink??1.8.1??jobmanager.execution.failover-strategy??region kill??taskmanagershuffleregion??source??cancelDEPLOYcancel?? ??1300

Re: flink本地编译源码失败

应该是镜像源的问题，改下镜像源就可以了。 这是我的配置（flink-runtime-web/pom.xml） install node and npm install-node-and-npm

Flink on Yarn 启动任务，任务重启后貌似taskmanager中会丢失lib中的jar文件

如题：link on Yarn 启动任务，任务重启后貌似taskmanager中会丢失lib中的jar文件 我的任务时kafka source -> hbase sink 任务申请到新的container后，启动任务时会丢失原本存在的class文件，怀疑是重新申请的container中没有获取到lib中的资源，是否应该将lib中资源放到hdfs？该如何配置。 Best xiao cai 错误堆栈： 2020-08-19 11:23:08,099 INFO org.apache.flink.yarn.YarnResourceManager

Re: Re: flink state ttl状态清理和重新计算的疑问

1） 这个应该还是按照每个key来单独清理的 2）是按照最后的更新时间，也就是每次更新的时候超时时间重新开始计算 赵一旦 于2020年8月17日周一 上午11:51写道： > @Li Benchao > （1） > 如果不是每个key一个timer，但超时仍然应该是key级别的吧，只是说清理机制上不是每个key设置一个timer去清理。 > 比如有个全局的其他机制定期扫描清理，但超时时间应该还是key级别。 > > （2） > 关于超时时间，对于一个key，他的value每次更新超时都会重新计算，还是永远按照这个key创建时时间开始计算呢。 > > Benchao Li 于2020年

Re: flink interval join后按窗口聚组问题

Hi Tianwang，一旦， 我感觉这个场景其实可以在Flink SQL中做一个优化，我建了一个issue[1]，欢迎讨论~ [1] https://issues.apache.org/jira/browse/FLINK-18996 赵一旦 于2020年8月17日周一 上午11:52写道： > 大概看了下。这个问题我业务中涉及到过。我是DataStream API做的。 > 不过我是在任务设计阶段就考虑了所有case，然后提前考虑了这些问题的。 > watermark是可以重设的。其次我还更改了interval join的算子实现，默认1.10只支持inner join。不支持l

回复：flink 1.11 order by rowtime报错

报下面的这个错误，并行度设置为1就没有问题了，不知道为什么 java.lang.NullPointerExcpetion at org.apache.flink.runtime.state.heap.StateTable.put(StateTable.java:336) at org.apache.flink.runtime.state.heap.StateTable.put(StateTable.java:159) at org.apache.flink.runtime.state.heap.HeapMapState.put(HeapMapState.java:101)

Re: Flink on Yarn 启动任务，任务重启后貌似taskmanager中会丢失lib中的jar文件

Hi 你的 Flink 是哪个版本，期望的行为是什么样的？ 从你给的日志看，是没有 这个 class，这个 是在你放到 lib 下的某个 jar 包里面吗？另外你这个作业第一次运行的时候失败，还是中间中间 failover 之后恢复回来的时候失败呢？ Best, Congxian xiao cai 于2020年8月19日周三 下午12:50写道： > 如题：link on Yarn 启动任务，任务重启后貌似taskmanager中会丢失lib中的jar文件 > > > 我的任务时kafka source -> hbase sink > > >

Re: Flink on Yarn 启动任务，任务重启后貌似taskmanager中会丢失lib中的jar文件

Hi Flink版本是 1.11.0 期望的行为是将kafka中的数据实时写入hbase表 xxx这个class是在lib下的某个jar中的 任务初始运行都是正常的，jar包也是可以找到的，运行期间失败了，然后进入了restarting状态，就不停的在running和restarting状态切换 我提交任务的节点是20，然后container运行的节点是22，lib中的jar都在20节点上，所以猜测是任务运行过程中，重新分配新的container时 丢失了lib中jar资源。 Best, xiao cai 原始邮件 发件人: Congxian Qiu 收件人: user-zh 发

Re: flink sql 如何指定之前的checkpoint执行

Hi 如果你的算子有改变的话，想从 checkpoint/savepoint 恢复，需要添加 `--allowNonRestoredState`，这样可以忽略掉那些不在新 job 中的算子（就算逻辑一样，uid 不一样也会被忽略掉的），具体的可以参考文档[1] [1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/state/savepoints.html#what-happens-if-i-delete-an-operator-that-has-state-from-my-job Best, C

Re: flink1.9.1用采用-d（分离模式提交）作业报错，但是不加-d是可以正常跑的

Hi 1.9 上是否加 -d 应该会使用不同的模式来启动作业 (perjob 还是 session），这两个模式下的行为应该是不完全一致的，具体的可以看下这里[1] [1] https://github.com/apache/flink/blob/5125b1123dfcfff73b5070401dfccb162959080c/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java#L211 Best, Congxian bradyMk 于2020年8月19日周三 上午10:54写道：

Re: task传输数据时反序列化失败

Hi 从栈看应该是 deserialize 的时候出错了，另外 kryo 可以，Pojo 不行，能否检查下，是否满足 POJO 的一些要求[1]呢? [1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/stream/state/schema_evolution.html#pojo-types Best, Congxian shizk233 于2020年8月18日周二 下午4:09写道： > Hi all, > > 请教一下反序列化的问题，我有一个KeyedCoProcessFunctio

Re: task传输数据时反序列化失败

关于kyro和pojo，我是指通过调整lombok的@Value/@Data注解改变数据对象的模式，从而使用不同的序列化器。 在@Data注解下，满足pojo序列化器要求并使用，但会遭遇异常。 目前发现，需要在MapState中明确指定List的数据类型，pojo序列化器才能正常得到结构，怀疑是MapStateDescriptor中类型信息提取错误导致的。 但奇怪的是，kryo序列化器却没有产生该异常。 我做了个问题复现的demo[1]，有时间的话可以查看一下。 [1]https://github.com/wangwangdaxian/flink-bug-replay Congxia

flink1.11.1 SQL任务启动报错

原来任务在flink 1.10.1 跑,是正常没问题的, 跑在1.11.1上启动就报错了, 求帮助啊 2020-08-1914:08:52,172WARN org.apache.flink.runtime.taskmanager.Task [] - Source: KafkaTableSource(userId, catid, brandId, utp, utrp, extendFields, requestTime) -> SourceConversion(table=[default_catalog.default_database.user_visit_trace, s

Re: Flink任务写入kafka 开启了EOS，statebackend为rocksdb，任务反压source端日志量堆积从cp恢复失败

Hi 1 图挂了 2 你到 hdfs 上能找到 hdfs:*xx*/flink/checkpoints/ 7226f43179649162e6bae2573a952e60/chk-167/_metadata 这个文件吗？ Best, Congxian Yang Peng 于2020年8月17日周一 下午5:47写道： > 找到了 具体日志如下：2020-08-13 19:45:21,932 ERROR > org.apache.flink.runtime.entrypoint.ClusterEntrypoint - Fatal error > occur