Re: flink 1.11 streaming 写入hive 5min表相关问题

Hi, 1.checkpoint会强制滚动 2.目前最简单的思路是加大checkpoint interval，另一个思路是在partition commit时触发hive去compaction。 3.success文件的生成依赖checkpoint interval，所以会有一定延迟。 Best, Jingsong On Thu, Jul 30, 2020 at 1:14 PM kandy.wang wrote: > 现象： > CREATE TABLE test.xxx_5min ( > > .. > > ) PARTITIONED BY (dt string , hm

flink 1.11 streaming 写入hive 5min表相关问题

现象： CREATE TABLE test.xxx_5min ( .. ) PARTITIONED BY (dt string , hm string) stored as orc TBLPROPERTIES( 'sink.partition-commit.trigger'='process-time', 'sink.partition-commit.delay'='5 min', 'sink.partition-commit.policy.kind'='metastore,success-file',

?????? flink-1.11 hive-1.2.1 ddl ????????????

sorry,idea??log4j??process-time ?? process time??log?? ---- ??:

Re: Flink jobs getting finished because of "Could not allocate the required slot within slot request timeout"

Hi Sateesh, Would you check Flink jobmanager log to see whether it has sent container requests to YARN RM? If the request is sent but not fulfilled, you will need to check the YARN RM logs or the YARN cluster resources at that time to see whether that container request is fulfillable. The

Re: 关于window过程中barrier的问题

Hi Shuwen barrier 是和 checkpoint 相关的逻辑，用来触发 checkpoint 的，你可以认为 barrier 和数据的顺序必须是严格保证的，不然没法保证 exactlyonce 的语义。 假设某个算子 B 有两个上游 A1 和 A2，那么 A1 和 A2 的 barrier 都发送的 B 之后，B 才会开始做 checkpoint，假设 A1 的 barrier 在 10:00 到了，A2 的 barrier 在 10:01 才到，那么 10:00 - 10:01 这段时间内，A1 发送到 B 的数据是否会被处理取决于是

flink????kafka??????????

flink1.11kafkagroup offset??kafka tooloffset??flink

Re: flink1.10.1/1.11.1 使用sql 进行group 和 时间窗口 操作后 状态越来越大

感谢！ flink1.11.1版本里面，我尝试了下面两种backend，目前运行了20多个小时， 能够看到状态的大小在一个区间内波动，没有发现一直增长的情况了。 StateBackend backend =new RocksDBStateBackend("hdfs:///checkpoints-data/"+yamlReader.getValueByKey("jobName").toString()+"/",false); StateBackend backend =new

?????? Sql??kafka????????????????

?? 1.10??connector type ---- ??: "user-zh"

Re: flink1.10.1/1.11.1 使用sql 进行group 和 时间窗口 操作后 状态越来越大

感谢！ flink1.11.1版本里面，我尝试了下面两种backend，目前运行了20多个小时， 能够看到状态的大小在一个区间内波动，没有发现一直增长的情况了。 StateBackend backend =new RocksDBStateBackend("hdfs:///checkpoints-data/"+yamlReader.getValueByKey("jobName").toString()+"/",false); StateBackend backend =new

Re: 近期使用flinksql(1.11.0)发现的一些问题

Q1, 可以使用 DROP TABLE IF EXISTS table; -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Sql往kafka表写聚合数据报错

抱歉哈，1.11 提供的 Canal 和 Debezium 还不支持 sink 。 预计会在 1.12 中提供。 On Wed, 29 Jul 2020 at 12:51, Benchao Li wrote: > 你可以用Canal或者Debezium format来写入kafka，那样就支持update和delete消息了。 > > op <520075...@qq.com> 于2020年7月29日周三 上午11:59写道： > > > 如下，想用sql直接往kafka写聚合结果，版本是1.11，请问能有什么办法解决，还是只能转换成datastream？？ > > > > > >

Re: 近期使用flinksql(1.11.0)发现的一些问题

Hi, 1. 这个功能应该是漏加了，我建了个 issue 去支持这个功能： https://issues.apache.org/jira/browse/FLINK-18756 2. update-mode 属性对于 kafka 来说一直都是没有用的（因为只支持 append-only）。所以自 1.10 ，这个属性就变成可选了，文档中也不再标识出来。 1.11 中新版的 connector 的实现中，也没有这个属性。 3. 目前 DDL 建表语句将表元信息存到 catalog 中，是不会走校验逻辑的。校验逻辑现在发生在 query 编译期。 这个确实和 fail-fast

Re: flink1.11 可以使用processtime开窗，但是无法使用eventtime开窗

你指定时间语义是EventTime了吗 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); > 2020年7月29日 上午9:56，111 写道： > > > > > > > > 您好，请教一个问题，谢谢： > 很简单的json， > {"num":100,"ts":1595949526874,"vin":""} > {"num":200,"ts":1595949528874,"vin":""} >

Re: flink1.10.1/1.11.1 使用sql 进行group 和 时间窗口 操作后 状态越来越大

感谢！ flink1.11.1版本里面，我尝试了下面两种backend，目前运行了20多个小时， 能够看到状态的大小在一个区间内波动，没有发现一直增长的情况了。 StateBackend backend =new RocksDBStateBackend("hdfs:///checkpoints-data/"+yamlReader.getValueByKey("jobName").toString()+"/",false); StateBackend backend =new

回复: flink1.11 sql 发布到yarn session时找不到hbase相关的类

Hi，Xintong： 是把相关的jar配到hadoop_classpath然后再export吗？ Best, Junbao Zhang 发件人: Xintong Song 发送时间: 2020年7月30日 10:28 收件人: user-zh 主题: Re: flink1.11 sql 发布到yarn session时找不到hbase相关的类 export HADOOP_CLASSPATH 就可以了 Thank you~ Xintong Song On Wed, Jul 29, 2020 at 6:43

Re: flink1.11 sql 发布到yarn session时找不到hbase相关的类

export HADOOP_CLASSPATH 就可以了 Thank you~ Xintong Song On Wed, Jul 29, 2020 at 6:43 PM wind.fly@outlook.com < wind.fly@outlook.com> wrote: > Hi,all： > 最近在升级flink1.11，sql中用到hbase connctor，发布到yarn-session时，报如下异常： > 2020-07-29 11:49:55 > org.apache.hadoop.hbase.DoNotRetryIOException: >

Re: flink1.10.1/1.11.1 使用sql 进行group 和 时间窗口 操作后 状态越来越大

感谢！ flink1.11.1版本里面，我尝试了下面两种backend，目前运行了20多个小时， 能够看到状态的大小在一个区间内波动，没有发现一直增长的情况了。 StateBackend backend =new RocksDBStateBackend("hdfs:///checkpoints-data/"+yamlReader.getValueByKey("jobName").toString()+"/",false); StateBackend backend =new

Flink Weekly | 每周社区动态更新 - 2020/07/30

大家好，本文为 Flink Weekly 的第二十四期，由王松整理，李本超Review。 本期主要内容包括：近期社区开发进展、邮件问题答疑、Flink 最新社区动态及技术文章推荐等。 社区开发进展 Release [releases] Flink 1.11.1 正式发布！ 具体信息参考： http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/ANNOUNCE-Apache-Flink-1-11-1-released-td43335.html Vote [vote] 伍翀发起Refactor

Flink Kafka EXACTLY_ONCE causing KafkaException ByteArraySerializer is not an instance of Serializer

I'm using Flink 1.10 and Kafka (AWS MSK) 2.2 and trying to do a simple app that consumes from one kafka topic and produces events into another topic. I would like to utilize the exactly_once semantic, however, I am experiencing the following error: org.apache.kafka.common.KafkaException:

Re: [Compression] Flink DataStream[GenericRecord] Avrowriters

Hi Ravi, With CustomAvroWriters (SNAPPY) when I run on a cluster, it does create files, but files are not recognized as avro files by avro tools jar, when I try to deserialize it to json. Flink Logs shows: 2020-07-29 23:54:23,270 INFO com.hadoop.compression.lzo.LzoCodec - Successfully loaded &

Re: flink on yarn 读取 hbase数据时 ，Task失败，具体描述如下

Hi，张锴 这个描述看起来没有用的信息呢，既然有任务失败，失败的日志和异常信息可以贴出来看看。或者贴一个可以复现这个失败的case. > 在 2020年7月29日，17:02，张锴 写道： > > flink获取Hbase数据并计算 > 在本地测试没问题，提交到Yarn上出现Task任务失败，无相关日志输出，task任务一直重启。任务失败的地方在数据计算部分。 > 语言：Scala，无堆栈信息输出 Best Leonard

Re: [Compression] Flink DataStream[GenericRecord] Avrowriters

Hi Ravi, Thanks for details. CustomAvrowriter was working for now. Although its failing for Snappy codec in local IDE with "java.lang.UnsatisfiedLinkError: org.apache.hadoop.util.NativeCodeLoader.buildSupportsSnappy()Z" I think I will have to try running it in an EMR/Hadoop environment to get

Re: [Compression] Flink DataStream[GenericRecord] Avrowriters

There is another alternative which you could try like this val stream:DataStream[GenericRecord] = _ val aof:AvroOutputFormat[GenericRecord] = new AvroOutputFormat(new org.apache.flink.core.fs.Path(""),classOf[GenericRecord]) aof.setSchema(schema) aof.setCodec(AvroOutputFormat.Codec.SNAPPY)

Re: [Compression] Flink DataStream[GenericRecord] Avrowriters

Hi Vijayendra, Currently AvroWriters doesn't support compression. If you want to use compression then you need to have a custom implementation of AvroWriter where you can add features of compression. Please find a sample customization for AvroWriters where you could use compression. You can use

Flink jobs getting finished because of "Could not allocate the required slot within slot request timeout"

Hi All, I have an EMR Cluster with one Master Node and 3 worker Nodes ( it has auto scaling enabled and the max no.of worker nodes can go up to 8). I have 3 Spark Jobs that are running currently on the Cluster. I submitted 3 Flink Jobs and all of them finished as the slots are not available

[Compression] Flink DataStream[GenericRecord] Avrowriters

Hi Team, Could you please provide a sample for Enabling Compression (Snappy) of Avro: DataStream[GenericRecord] AvroWriters.forGenericRecord(schema) Regards, Vijay

Matching largest event pattern without duplicates

Hey all, I'm trying to complete a small POC to see if Flink is suitable for our needs and the first step is to evaluate a stream of events and continually output the largest active group that does not contain duplicates. I'm attempting to do this with the CEP pattern matching. For example, for

Re: 使用datagen connector生成无界数据，一秒时间窗口的聚合操作，一直没有数据输出打印

Hi > >bsTableEnv.executeSql("SELECT f_random, count(1) " + >"FROM datagen " + >"GROUP BY TUMBLE(ts, INTERVAL '1' second), f_random").print(); TableResult.print() 方法目前只支持了 exactly-once 语义，在 streaming 模式下必须设置checkpoint才能work， 你配置下checkpoint之后再试下，支持 At

Re: Compression Streamingfilesink ROW-encoded format

Thank You. On Wed, Jul 29, 2020 at 2:07 AM Ravi Bhushan Ratnakar < ravibhushanratna...@gmail.com> wrote: > Yes, flink-compress module is supported from 1.10.0 and onward. > > Regards, > Ravi > > On Tue 28 Jul, 2020, 23:11 Vijayendra Yadav, > wrote: > >> Thank You Ravi for Quick help. One Last

Re: flink-1.11 hive-1.2.1 ddl 无法写入数据

Hi, kcz 看connector的properties还是1.10的格式，你换成1.11试试[1]. > 在 2020年7月29日，15:07，kcz <573693...@qq.com> 写道： > > tableEnv.executeSql("CREATE TABLE situation.source_table (\n" + >"\thost STRING,\n" + >"\turl STRING,\n" + >"\tpublic_date STRING\n" + >

Re:flink-1.11 hive-1.2.1 ddl 无法写入数据

有什么异常信息吗 在 2020-07-29 14:07:26，"kcz" <573693...@qq.com> 写道： >确认数据源有数据，全部代码如下，但是hive就是没有数据 > >package com.hive; > >import org.apache.flink.runtime.state.filesystem.FsStateBackend; >import org.apache.flink.streaming.api.CheckpointingMode; >import

unsubscribe

Re:使用datagen connector生成无界数据，一秒时间窗口的聚合操作，一直没有数据输出打印

Hi Asahi Lee: 我在 master 上的 flink-sql-client 模块中建了一个类，复制你的代码控制台是有输出的，你使用的版本是什么的？ Best, Hailong Wang 在 2020-07-29 15:35:30，"Asahi Lee" <978466...@qq.com> 写道： >以下程序运行，控制台一直没有数据输出1. 程序package kafka; > >import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; >import

Re: Making sense of Meter metrics graph on Grafana

Thanks. Got it. With regards On Wed, Jul 29, 2020 at 8:18 PM Chesnay Schepler wrote: > Yes; a rate of 1 means that 1 event occurred per second, which in your > case means one call to markEvent() per second. > > Note that the default Meter implementation calculates the rate per > second over

Re: Making sense of Meter metrics graph on Grafana

Yes; a rate of 1 means that 1 event occurred per second, which in your case means one call to markEvent() per second. Note that the default Meter implementation calculates the rate per second over the last minute (basically, rate(T) = (count(T) - count(T-60)) / 60; so short spikes tend to be

Re: Flink state reconciliation

Hi Kostas Thanks for a possible help! пт, 24 июл. 2020 г., 19:08 Kostas Kloudas : > Hi Alex, > > Maybe Seth (cc'ed) may have an opinion on this. > > Cheers, > Kostas > > On Thu, Jul 23, 2020 at 12:08 PM Александр Сергеенко > wrote: > > > > Hi, > > > > We use so-called "control stream" pattern

Re: Flink 1.11.0 UDAF fails with "No match found for function signature fun()"

Hi Timo, Thank you for your time and your help! Described approach works. On 29/07/2020, 10:22, "Timo Walther" wrote: Hi Dmytro, I would not recommend to use internal functions from `org.apache.flink.table.planner.functions.aggfunctions`. They are called by a slightly

Colocating Compute

Hello, I am using Flink v1.10 in a distributed environment to run SQL queries on batch and streaming data. In my setup, data is sharded and distributed across the cluster. Each shard receives streaming updates from some external source. I wish to minimize data movement during query evaluation

[DISCUSS] FLIP-131: Consolidate the user-facing Dataflow SDKs/APIs (and deprecate the DataSet API)

Hi Everyone, my colleagues (in cc) and I would like to propose this FLIP for discussion. In short, we want to reduce the number of APIs that we have by deprecating the DataSet API. This is a big step for Flink, that's why I'm also cross-posting this to the User Mailing List. FLIP-131:

flink1.11 sql 发布到yarn session时找不到hbase相关的类

Hi,all： 最近在升级flink1.11，sql中用到hbase connctor，发布到yarn-session时，报如下异常： 2020-07-29 11:49:55 org.apache.hadoop.hbase.DoNotRetryIOException: java.lang.NoClassDefFoundError: Could not initialize class org.apache.hadoop.hbase.util.ByteStringer at

flink1.9.1 在WebUI中查看committedOffsets指标为负值

flink1.9.1 在WebUI中查看Source__Custom_Source.KafkaConsumer.topic.geek-event-target.partition.3.committedOffsets指标为负值，查看官网释义：对于每个分区，最后一次成功提交到Kafka的偏移量。 但我这里为什么是负值呢？ 希望能得到指导，万分感谢~ -- Sent from:

State Restoration issue with flink 1.10.1

Hi, Presently I have a flink application running on version 1.8.2 I have taken a savepoint on the running app which is stored in s3 , Now I have changed my flink version to 1.10.1 , Now when I running the new application on version flink-1.10.1 from the savepoint taken on flink 1.8.2 it is

flink on yarn 读取 hbase数据时 ，Task失败，具体描述如下

flink获取Hbase数据并计算 在本地测试没问题，提交到Yarn上出现Task任务失败，无相关日志输出，task任务一直重启。任务失败的地方在数据计算部分。 语言：Scala，无堆栈信息输出

Re: PyFlink DDL UDTF join error

Hi Wei, Thank you for the clarification and workaround. Regards, Manas On Wed, Jul 29, 2020 at 12:55 PM Wei Zhong wrote: > Hi Manas, > > It seems a bug of the create view operation. I have created a JIRA for it: > https://issues.apache.org/jira/browse/FLINK-18750 > > Before repairing, please

Re: Compression Streamingfilesink ROW-encoded format

Yes, flink-compress module is supported from 1.10.0 and onward. Regards, Ravi On Tue 28 Jul, 2020, 23:11 Vijayendra Yadav, wrote: > Thank You Ravi for Quick help. One Last Question is this compression > supported with Flink Version 1.10.0 ? > > Regards, > Vijay > > On Tue, Jul 28, 2020 at

关于window过程中barrier的问题

大家好，想请教一个关于barrier的问题 如果我有如下算子 .window() .reduce() 假设barrier和元素顺序是 tuple 和 barrier 当tuple流经窗口期，barrier是会阻塞等待tuple从窗口流出之后再达到下一个算子，还是不在窗口阻塞，直接流到下一个算子呢？ Best Wishes, Shuwen Zhou

回复： Flink使用Kafka作为source时checkpoint成功提交offset的机制

你是说emit之后的offset commit么？可以看下 `Kafka09Fetcher`的runFetchLoop方法 在2020年07月29日 15:09，shuwen zhou 写道： 比如读到一条offset值为100的消息，有Kafka Consumer source，一个Process算子和一个Sink算子 那么在这条消息到达Process算子，还未到达Sink算子时，提交offset是100吗？ On Wed, 29 Jul 2020 at 14:51, venn wrote: checkpoint成功时就会把它的offset提交，可以看下这个类:

近期使用flinksql(1.11.0)发现的一些问题

1. create table语句不支持create talbe if not exists： 不支持if not exists语法在实际使用时特别麻烦，每次重新执行SQL都需要先删除上次执行创建的table。 Q1： CREATE TABLE IF NOT EXITS语法个人理解实现并不特别麻烦，社区为什么还没实现？ 2. flink1.11创建kafka sink表时不再支持update-mode属性： 创建kafka sink表时报不支持udpate-mode属性的语法检验错误。但查看flink1.11源码中仍存在多个测试类在使用update-mode属性。

RE: Flink使用Kafka作为source时checkpoint成功提交offset的机制

可以这样理解，实际上souce 算子只知道这条数据发出去了，不知道这条数据执行到哪里的 -Original Message- From: user-zh-return-5981-wxchunjhyy=163@flink.apache.org On Behalf Of shuwen zhou Sent: 2020年7月29日 15:10 To: user-zh@flink.apache.org Subject: Re: Flink使用Kafka作为source时checkpoint成功提交offset的机制

????datagen connector??????????????????????????????????????????????????????????

1. package kafka; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.EnvironmentSettings; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; public class DataGenTest {

Re: Count of records coming into the Flink App from KDS and at each step through Execution Graph

I'd recommend to do the aggregation over 5 seconds in graphite/prometheus etc., and expose a counter in Flink for each attribute/event_name. User variables are a good choice for encoding the attribute/event_name values. As for your remaining questions: Flink does not support aggregating

Re: Flink 1.11.0 UDAF fails with "No match found for function signature fun()"

Hi Dmytro, I would not recommend to use internal functions from `org.apache.flink.table.planner.functions.aggfunctions`. They are called by a slightly different stack that might cause this exception. Instead you can use the testing functions in

Re: PyFlink DDL UDTF join error

Hi Manas, It seems a bug of the create view operation. I have created a JIRA for it: https://issues.apache.org/jira/browse/FLINK-18750 Before repairing, please do not use create view operation for udtf call. Best, Wei > 在

Re: Flink使用Kafka作为source时checkpoint成功提交offset的机制

比如读到一条offset值为100的消息，有Kafka Consumer source，一个Process算子和一个Sink算子 那么在这条消息到达Process算子，还未到达Sink算子时，提交offset是100吗？ On Wed, 29 Jul 2020 at 14:51, venn wrote: > checkpoint成功时就会把它的offset提交，可以看下这个类: FlinkKafkaConsumerBase 的 这个方法： > notifyCheckpointComplete > > -Original Message- > From:

Re: flink 1.11 rest api saveppoint接口 异常

好哒，我的自动更新逻辑依赖了这个 api，不过现在我用其他方式先处理了。 感谢相应，祝好~ 原始邮件 发件人: Congxian Qiu 收件人: user-zh 发送时间: 2020年7月29日(周三) 13:34 主题: Re: flink 1.11 rest api saveppoint接口 异常 Hi 创建了一个 Issue[1] 来跟进这个问题 [1] https://issues.apache.org/jira/browse/FLINK-18748 Best, Congxian Congxian Qiu 于2020年7月29日周三 下午1:24写道： > Hi

flink-1.11 hive-1.2.1 ddl ????????????

hive package com.hive; import org.apache.flink.runtime.state.filesystem.FsStateBackend; import org.apache.flink.streaming.api.CheckpointingMode; import org.apache.flink.streaming.api.TimeCharacteristic; import

Re: Flink Deployment on Kubernetes session Cluster

Hi Yang, Thank you for your reply. Yes, we have evaluated job specific clusters (as we used to deploy the same in YARN) , the main issue is Job monitoring of multiple jobs as we won't be having a single endpoint like YARN does . We will evaluate K8's operator you have suggested Thanks and

RE: Flink使用Kafka作为source时checkpoint成功提交offset的机制

checkpoint成功时就会把它的offset提交，可以看下这个类: FlinkKafkaConsumerBase 的 这个方法： notifyCheckpointComplete -Original Message- From: user-zh-return-5976-wxchunjhyy=163@flink.apache.org On Behalf Of shuwen zhou Sent: 2020年7月29日 14:24 To: user-zh@flink.apache.org Subject:

FLink1.11.1整合hadoop3.0.0

以 -m yarn-clsuter运行flink1.11.1的examples/streaming/SocketWindowWordCount.jar 任务正常。ncat -l 输入数据可以在taskmanager.out日志中看到。 但是，yarn 中ApplicationMaster（也就是Tracking UI）打不开flink集群的页面 提示：{"errors":["Unable to load requested file /index.html"]} 大佬遇到过吗？

Flink使用Kafka作为source时checkpoint成功提交offset的机制

大家好，请教一个问题， 当Flink在使用Kafka作为source，并且checkpoint开启时，提交的offset是在消息被链路最后一个算子处理之后才会把这条offset提交吗？ 还是说这个消息只要从Kafka source读取后，不论链路处理到哪个算子， checkpoint成功时就会把它的offset提交呢？ 另外有大神指路这段代码具体在哪个文件吗？ 谢谢！ -- Best Wishes, Shuwen Zhou

Re: Flink 1.11.1 - job manager exists with exit code 0

Thanks for reporting back. Glad you found the issue. This reminds me of a ticket about this topic some time ago :) https://issues.apache.org/jira/browse/FLINK-15156 On Wed, Jul 29, 2020 at 7:51 AM Alexey Trenikhun wrote: > Hi Robert, > I found the cause, it was due to bug in job itself - code

Re: Customization of execution environment

Hi Flavio, I think the recommended approach is as follows: (then you don't need to create to environments) final Configuration conf = new Configuration(); conf.setLong(...) env = new LocalEnvironment(conf); I agree that in theory it would be nicer if the configuration returned was editable, but

Re: JobManager refusing connections when running many jobs in parallel?

Hi Andreas, Thanks for reaching out .. this should not happen ... Maybe your operating system has configured low limits for the number of concurrent connections / sockets. Maybe this thread is helpful: