Flink与MySQL对接相关的问题

大家好！ 使用Flink与MySQL对接的过程中，遇到如下的问题： 1. MySQLCatalog目前未支持，对MySQL中的表进行读写需要先通过DDL进行声明，由于表中的字段比较多，操作起来会比较繁琐。搜查了一遍，发现目前社区对这个支持还没有列入计划。如果可以推动在1.13开发支持该Catalog，那就太好了。 https://issues.apache.org/jira/browse/FLINK-15352 https://issues.apache.org/jira/browse/FLINK-15350 2.不支持针对部分字段的插入更新;

flink sql 执行limit 很少的语句依然会暴增

请教个问题，使用flink sql 去拉取mysql数据，mysql源表有千万级别数据量，使用了 select * from sourcTable limit 10； 即使是limit几条数据也会导致内存暴增。这里的limit是从mysql原表执行后 直接在flink taskmanager进行limit取数据吗？

Re: flink1.12 on yarn per-job 运行问题

感谢回复，看日志好像没什么异常。日志已添加到附件。 Yarn Sesseion 模式运行时是没问题的， Per-job 和 Application Mode 就不行。 -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: [DISCUSS] Correct time-related function behavior in Flink SQL

Great examples to understand the problem and the proposed changes, @Kurt! Thanks Leonard for investigating this problem. The time-zone problems around time functions and windows have bothered a lot of users. It's time to fix them! The return value changes sound reasonable to me, and keeping the

Re: [DISCUSS] Correct time-related function behavior in Flink SQL

Great examples to understand the problem and the proposed changes, @Kurt! Thanks Leonard for investigating this problem. The time-zone problems around time functions and windows have bothered a lot of users. It's time to fix them! The return value changes sound reasonable to me, and keeping the

Re: flink1.12 on yarn per-job 运行问题

感谢回复。 CDH 集群 160G, 64C，平时主要夜间跑离线任务，提交程序时没指定任何资源相关的配置 -- Sent from: http://apache-flink.147419.n8.nabble.com/

flink整合hive问题

我将线上的hive-site文件复制到了flink的conf中，并且相关jar都放好。 使用flink的sql-client的-l方式将相关jar指定，启动了sql-client。 然后catalog，databases，tables等信息都能查询到了。 但是select * from xxTable where dt=''; 就会有问题。 看了flink集群报错，这个错误直接导致flink的standalonesession进程失败会。 报错如下： 2021-01-21 13:43:42,818 INFO

Re: Handling validations/errors in the flink job

Hi Team, any answer for my below question? On Wed, Jan 20, 2021 at 9:20 PM sagar wrote: > Hi Team, > > > I am creating a flink job with DataStream API and batch mode. > > It is having 5 different bounded sources and I need to perform some > business operations on it like joining , aggregating

Re: 回复：Flink 1.11 SQL可以支持kafka动态分区发现么？

1.11 文档里面没有，那么估计就不支持了，可以看下 1.12 如何实现的，然后把这个 patch 打在内部自己的 flink 版本里面 sunfulin 于2021年1月20日周三 下午2:53写道： > > 我看下这个源码，貌似是有这个参数。还不确定SQL ddl里配置会不会生效，回头验证下。 > > > > > -- > 发自我的网易邮箱手机智能版 > > > > - Original Message - > From: "Shuai Xia" > To: user-zh , sunfulin0321 > > Sent: Wed, 20 Jan 2021

Re: flink heartbeat timeout

1. 50s 的 timeout 时间通常应该是够用的。建议排查一下 timeout 当时环境中是否存在网络抖动，或者 JM/TM 进程是否存在长时间 GC 导致不响应。 2. 目前 flink 集群配置无法做到不重启热更新 Thank you~ Xintong Song On Thu, Jan 21, 2021 at 11:39 AM guoxb__...@sina.com wrote: > Hi > > *问题描述：* > > >

Re: Pyflink JVM Metaspace 内存泄漏定位

cc @Jark 看起来像是 JDBC connector 的问题。这块你熟悉吗？或者知道谁比较熟悉吗？ Thank you~ Xintong Song On Wed, Jan 20, 2021 at 8:07 PM YueKun wrote: > hi，不确定是否能看到图片，Jmap导出的数据分析看如下：< > http://apache-flink.147419.n8.nabble.com/file/t1276/WX20210120-191436.png> > > > > > -- > Sent from:

Re: Flink SQL kafka connector有办法获取到partition、offset信息嘛？

CREATE TABLE KafkaTable ( `event_time` TIMESTAMP(3) METADATA FROM 'timestamp', `partition` BIGINT METADATA VIRTUAL, `offset` BIGINT METADATA VIRTUAL, `user_id` BIGINT, `item_id` BIGINT, `behavior` STRING ) WITH ( 'connector' = 'kafka', 'topic' = 'user_behavior',

Flink ddl sql 在 Test和在Main里面执行结果不同

同一段代码，在main里面可以正常正常，在Test里面却直接结束 StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment(); EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build(); StreamTableEnvironment bsTableEnv =

Re: 设置状态存储位置后，job运行起来后找不到状态数据

你配置的是本地目录，不是 hdfs 目录，当重启后，可能新的任务运行的机器不是之前的那台机器了，那么之前作业的状态信息（在其他机器上）是不在新的机器上的，那么就会发现找不到状态文件，建议配置成 HDFS 的 Best zhisheng 刘海 于2021年1月20日周三 下午9:05写道： > Hi all > 小弟遇到个问题期望大佬解答解答： > 通过 env.setStateBackend(new > RocksDBStateBackend("file:///data/flink/checkpoints"));设置状态存储位置，job运行起来后找不到状态数据， >

Re: flink1.12 on yarn per-job 运行问题

应该要提供一下 jm 的日志，麻烦检查一下 jm 里面的日志是否有异常的日志，我们遇到过类似的问题是因为包冲突导致的作业申请资源有问题，最后一直处于 created 状态 Best zhisheng 花乞丐 于2021年1月21日周四 上午8:47写道： > 贴一下提交程序的参数，以及你机器的配置，从上面看，是资源分配不够！ > > > > -- > Sent from: http://apache-flink.147419.n8.nabble.com/

A few questions about minibatch

Hi, Our job was experiencing high write amplification on aggregates so we decided to give mini-batch a go. There's a few things I've noticed that are different from our previous job and I would like some clarification. 1) Our operators now say they have Watermarks. We never explicitly added

Re: Pyflink 提交 Batch 任务后报错 “Failed to execute sql”

结束这个问题，目前没有再出现，可能是以下两个原因的一个： 1. 更新了 Flink 版本，从 1.12.0 更新至 1.12.1 2. 扩大了Task Off-Heap 空间 -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: [DISCUSS] Correct time-related function behavior in Flink SQL

cc this to user & user-zh mailing list because this will affect lots of users, and also quite a lot of users were asking questions around this topic. Let me try to understand this from user's perspective. Your proposal will affect five functions, which are: - PROCTIME() - NOW() -

Re: [DISCUSS] Correct time-related function behavior in Flink SQL

cc this to user & user-zh mailing list because this will affect lots of users, and also quite a lot of users were asking questions around this topic. Let me try to understand this from user's perspective. Your proposal will affect five functions, which are: - PROCTIME() - NOW() -

Re: Flink Jobmanager HA deployment on k8s

Hi Amit Bhatia > What is the correct way to start three jobmanager replicas with zk? Is there any link which explains this deployment scenario and configuration? Please find more information in the last mail. Unfortunately, we do not have some documentation to guide the users how to achieve that.

Re: Flink Jobmanager HA deployment on k8s

Hi Chirag Dewan, Yes, we could have multiple replicas with ZK HA in K8 as well. Multiple JobManagers will contend for a leader and then write its rpc address to the ZooKeeper nodes. You could find more information how the HA service works here[1]. It is about the KubernetesHAService, but the

Re: Flink SQL kafka connector有办法获取到partition、offset信息嘛？

你好，可以获取 CREATE TABLE KafkaTable ( `event_time` TIMESTAMP(3) METADATA FROM 'timestamp', `partition` BIGINT METADATA VIRTUAL, `offset` BIGINT METADATA VIRTUAL, `user_id` BIGINT, `item_id` BIGINT, `behavior` STRING ) WITH ( 'connector' = 'kafka', 'topic' = 'user_behavior',

Re: flink yarn application 提交任务出错

这个报错应该是你HDFS的core-site.xml的配置有问题 defaultFS应该是hdfs://localhost:9000/才对，你可以检查一下 Best, Yang casel.chen 于2021年1月19日周二 下午6:19写道： > 今天尝试使用yarn > application模式（带yarn.provided.lib.dirs参数)，将$FLINK_LIB目录下的jar包上传到了hdfs，结果报了如下的错，是少了哪个jar包或配置文件吗？ > > >

Re: flink yarn application提交作业问题

目前user jar是可以支持远程，但是只能是hadoop compatiable的schema 因为远程的这个user jar并不会下载到Flink client本地，而是直接注册为Yarn的local resource来使用 所以你的这个报错是预期内的，还没有办法支持 Best, Yang casel.chen 于2021年1月20日周三 上午10:23写道： > ./bin/flink run-application -t yarn-application \ > >

Re: Kubernetes HA Services - artifact for KubernetesHaServicesFactory

You do not need to put flink-kubernetes_2.12-1.12.1.jar under the plugin directory. Only the S3 fs jar needs to be put there. I think I found the root cause. It seems your flink-dist is still 1.11. Do you want to use the image 1.11 and enable the K8s HA at the same time? I think it could not

flink heartbeat timeout

Hi 问题描述： 我在使用flink进行流式计算任务，我的程序造昨晚上21点启动的，当时看是正常的，数据也是正常处理的，在今早9点时候查看，任务被自动重启了，查看日志，报错如下： 从报错上来看是由于超时时间引起的，查看资料，是需要调整该参数参数：heartbeat.timeout，官网文档支出默认值是5，但是这样以来的话，就需要重启flink服务了，这在我们生产上是不允许的。 问题： 1、该错误的原因目前只是经过猜测，还没有确定具体的问题，希望有经验的朋友指点一二，万分感谢

Flink SQL kafka connector有办法获取到partition、offset信息嘛？

如题，需要获取到kafka的partition、offset进行处理 -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re:yarn-per-job 模式 savepoint执行保存点报错

./bin/flink savepoint :jobId [:targetDirectory] -yid :yarnAppId ./bin/flink savepoint fea3d87f138ef4c260ffe9324acc0e51 [:targetDirectory] application_1610788069646_0021 [:targetDirectory] hdfs:///flink/savepoints 在 2021-01-21 10:24:31，"刘海" 写道： >Hi >

Re: flink 写hive decimal类型报错

你好，有设置过table.exec.hive.fallback-mapred-writer参数么？可以把它设置成true再试试。 On Wed, Jan 20, 2021 at 4:39 PM kandy.wang wrote: > java.lang.NoSuchMethodError: > org.apache.hadoop.hive.serde2.io.HiveDecimalWritable.serialize64(I)J > > at >

yarn-per-job 模式 savepoint执行保存点报错

Hi 我目前在进行保存点相关的测试，目前执行命令报如下错误，从错误内容上看是超时，但是没有更多的信息了，有知道大致原因希望指点一下，拜谢 flink1.12 yarn-per-job 模式 jobID:fea3d87f138ef4c260ffe9324acc0e51 yarnID : application_1610788069646_0021 执行的命令如下： ./bin/flink savepoint -t yarn-per-job -D yarn.application.id=application_1610788069646_0021

flink 1.12.0版本 消费0.10版本kafka集群数据==>0.9版本kafka集群

请问下如何选择kafka connector的版本 如果选择1.12.0版本，就没有FlinkKafkaProducer09/FlinkKafkaConsumer09 org.apache.flink flink-connector-kafka_${scala.binary.version}

Re: flink1.12 on yarn per-job 运行问题

贴一下提交程序的参数，以及你机器的配置，从上面看，是资源分配不够！ -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Converting non-mini-batch to mini-batch from checkpoint or savepoint

Just tested this and I couldn't restore from a savepoint. If I do a new job from scratch, can I tune the minibatch parameters and restore from a savepoint without having to make yet another brand new job? Thanks On Wed, Jan 20, 2021 at 12:43 PM Rex Fenley wrote: > Hello, > > Is it safe to

Re: Kubernetes HA Services - artifact for KubernetesHaServicesFactory

I did move to 1.12.1 version and also ensured that docker has kubernetes jar file in the right location, i.e. /opt/flink/plugins/s3-fs-presto/flink-kubernetes_2.12-1.12.1.jar But job manager is still not able to find the class 2021-01-21 00:00:49,376 ERROR

org.apache.flink.runtime.client.JobSubmissionException: Job has already been submitted

Hello, We're running 1.9.2 on YARN, and are seeing some interesting behavior when submitting jobs in a multi-threaded fashion to an application's Flink cluster. The error we see reported in the client application logs is the following:

Converting non-mini-batch to mini-batch from checkpoint or savepoint

Hello, Is it safe to convert a non-mini-batch job to a mini-batch job when restoring from a checkpoint or a savepoint? Thanks -- Rex Fenley | Software Engineer - Mobile and Backend Remind.com | BLOG | FOLLOW US

Re: Problem with overridden hashCode/equals in keys in Flink 1.11.3 when checkpointing with RocksDB

I have an update. I have created a small project on github, https://github.com/daha/flink-key-by-problem-with-rocksdb-state, which reproduces the issue. There seems to be problem with RocksDB in all versions I have tested (from 1.7.1 and later). In Flink 1.9.x only one of the events is

Re: Restoring from a checkpoint or savepoint on a different Kafka consumer group

Thanks! On Tue, Jan 19, 2021 at 9:47 PM Piotr Nowojski wrote: > Hi Rex, > > Sorry, I might have misled you. I think you were right in your previous > email > > > So from the sounds of things, regardless of the consumer group's > offsets, it will always start from a checkpoint or savepoints

Re: question about timers

Yes, Processing time timers that should have fired will fire immediately in order. Event time timers are never *late*, they will just fire when the watermark advances. Seth On Tue, Jan 19, 2021 at 3:58 PM Marco Villalobos wrote: > If there are timers that have been checkpointed (we use

Re: Trying to create a generic aggregate UDF

Timo, I converted what I had to Java, and ended up with the exact same issue as before where it will work if I only ever use it on 1 type, but not if I use it on multiple. Maybe this is a bug? Dylan On 1/20/21, 10:06 AM, "Dylan Forciea" wrote: Oh, I think I might have a clue as to what

Re: Trying to create a generic aggregate UDF

Oh, I think I might have a clue as to what is going on. I notice that it will work properly when I only call it on Long. I think that it is using the same generated code for the Converter for whatever was called first. Since in Scala I can't declare an object as static within the class itself,

Handling validations/errors in the flink job

Hi Team, I am creating a flink job with DataStream API and batch mode. It is having 5 different bounded sources and I need to perform some business operations on it like joining , aggregating etc. I am using a CoGroup operator to join two streams as it serves as a left join. So when keys are

Re: Counter metrics for prometheus having unexepcted gaps in grafana

Not sure whether it would solve your issue, but you could maybe exclude the pod id (I assume you mean the host?) from being reported by setting this: metrics.reporter..scope.variables.excludes: On 1/20/2021 7:16 AM, Manish G wrote: Hi All, I am facing an issue with counter metrics I have

Re: Trying to create a generic aggregate UDF

As a side note, I also just tried to unify into a single function registration and used _ as the type parameter in the classOf calls there and within the TypeInference definition for the accumulator and still ended up with the exact same stack trace. Dylan On 1/20/21, 9:22 AM, "Dylan

Re: Trying to create a generic aggregate UDF

Timo, I appreciate it! I am using Flink 1.12.0 right now with the Blink planner. What you proposed is roughly what I had come up with the first time around that resulted in the stack trace with the ClassCastException I had originally included. I saw that you had used a Row instead of just the

Re: Publishing a table to Kafka

Thanks Leonard, we are working towards 1.12 upgrade and should be able to try upsert-kafka after that. > Your first workaround should have been worked, but looks like an exception > was thrown in Type conversion phase, could you share you table schema and > query that can reproduce the issue.

Re: Error querying flink state

Hi Falak, it is hard to tell what is going wrong w/o the debug logs. Could you check whether they contain anything specific? You can also share them with us. Cheers, Till On Wed, Jan 20, 2021 at 1:04 PM Falak Kansal wrote: > Hi, > > Thank you so much for the response. I am using the 1.12

设置状态存储位置后，job运行起来后找不到状态数据

Hi all 小弟遇到个问题期望大佬解答解答： 通过 env.setStateBackend(new RocksDBStateBackend("file:///data/flink/checkpoints"));设置状态存储位置，job运行起来后找不到状态数据， flink1.12 yarn pre job 模式，下面是我的配置，job运行起来后在服务器上找不到 “/data/flink/checkpoints”这个目录，像我设置了状态的存储位置是不是job一运行起来对应的存储位置就应该有状态的数据呢？ public class FlinkTestDemo {

Re: Pyflink JVM Metaspace 内存泄漏定位

hi，不确定是否能看到图片，Jmap导出的数据分析看如下： -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Pyflink JVM Metaspace 内存泄漏定位

我是Python的table API实现的，connector是jdbc，jar包是用的flink网站上提供的链接里的（https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/connectors/jdbc.html），有 flink-connector-jdbc_2.11-1.12.0.jar 和 mysql-connector-java-5.1.49.jar。 任务是用SQL写的，基本和flink提供的demo的代码一致： " ) WITH ( 'connector' = 'jdbc',

Re: Pyflink JVM Metaspace 内存泄漏定位

JDBC连接是谁创建的，能找到相关调用栈吗，是 flink 提供的 connector 还是用户代码？ Thank you~ Xintong Song On Wed, Jan 20, 2021 at 6:32 PM YueKun wrote: > 目前看泄漏是因为 mysql 的 JDBC 引起的，和 > > http://apache-flink.147419.n8.nabble.com/1-11-1-OutOfMemoryError-Metaspace-td8367.html#a8399 > 这个问题一样。这个有什么解决方法吗？需要更换

Re: Pyflink JVM Metaspace 内存泄漏定位

目前看泄漏是因为 mysql 的 JDBC 引起的，和 http://apache-flink.147419.n8.nabble.com/1-11-1-OutOfMemoryError-Metaspace-td8367.html#a8399 这个问题一样。这个有什么解决方法吗？需要更换 mysql-connector-java 版本吗？ 我目前用的 5.1.49 版本 -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: 回复：1.11.1 报OutOfMemoryError: Metaspace. 错误

你好，请问这个问题解决了吗？我目前也有遇到这个情况 -- Sent from: http://apache-flink.147419.n8.nabble.com/

flink1.12 on yarn per-job 运行问题

跑demo都能正常跑，但是跑自己的程序就出现任务一直处于created 状态。 yarn 集群资源充足。 在flink的web 界面 Exceptions 里会看到 java.util.concurrent.CompletionException: org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException:

Re: UDTAGG在SQL中可以使用么，语法是什么

你好，我这有一个使用场景，是计算前一段时间内的某字段的TopN，需要使用over窗口和udtagg的结合。 查看官网发现，udtagg不支持flinksql, tableapi的over窗口后面也只能接select语句，请问下，还有其它办法吗？ -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: flink 写hive的并行度只能为1

hi， 并行度为1的是discard sink，实际写数据的是filewriter算子 > 2021年1月20日 下午4:29，高函 写道： > > > 为什么目前flink写hive的并行度只能设置为1呢？

flink 写hive decimal类型报错

java.lang.NoSuchMethodError: org.apache.hadoop.hive.serde2.io.HiveDecimalWritable.serialize64(I)J at org.apache.orc.impl.ColumnStatisticsImpl$Decimal64StatisticsImpl.updateDecimal(ColumnStatisticsImpl.java:1010) at

Problem with overridden hashCode/equals in keys in Flink 1.11.3 when checkpointing with RocksDB

I have encountered a problem in Flink when trying to upgrade from Flink 1.9.1 to Flink 1.11.3. The problem in a combination of 2 components: * Keys implemented as case classes in Scala where we override the equals and hashCode methods. The case class has additional fields which we are not used

Re: Trying to create a generic aggregate UDF

Hi Dylan, I'm assuming your are using Flink 1.12 and the Blink planner? Beginning from 1.12 you can use the "new" aggregate functions with a better type inference. So TypeInformation will not be used in this stack. I tried to come up with an example that should explain the rough design. I

flink 写hive的并行度只能为1

为什么目前flink写hive的并行度只能设置为1呢？

Re: K8s HA Session模式下1.12.1 jobmanager 周期性 restart

拿到了吗？有什么发现吗？ -- Sent from: http://apache-flink.147419.n8.nabble.com/