flink 1.17.1 on Yarn实时任务运行了几天出现了Yarn
token过期问题,在1.12未出现。这块具体有什么变化嘛，我是否还需要再配置其他参数。
具体配置:
```
security.kerberos.access.hadoopFileSystems: viewfs://AutoLfCluster;hdfs://ns1
security.kerberos.login.keytab: /xxx/krb5.keytab
security.kerberos.login.principal: flink/xxx
Hi casel
之前有类似的讨论， 不过暴露 ROWKIND 之后可能可以会造成 SQL 语义上的不明确，你可以在 dev 邮件在发起讨论看看，看看大家的想法。
https://issues.apache.org/jira/browse/FLINK-24547
Best,
Feng
On Wed, Jul 19, 2023 at 12:06 AM casel.chen wrote:
> 社区无人响应吗？
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
> 在 2023-07-15 12:19:46，"casel.chen" 写道：
>
社区无人响应吗？
在 2023-07-15 12:19:46，"casel.chen" 写道：
>Flink社区能否考虑将ROWKIND做成元数据metadata，类似于kafka
>connector中的offset和partition等，用户可以使用这些ROWKIND
>metadata进行处理，例如对特定ROWKIND过滤或者答输出数据中添加ROWKIND字段
退订
在flink内需要使用mybatis做些简化查询的工作，目前我的使用方式如下
public class MybatisUtil {
private static final Logger LOGGER = LogFactory.createNewLogger("MybatisUtil");
private static ThreadLocal tl = new ThreadLocal();
private static SqlSessionFactory factory = null;
//private static SqlSession
Hi, 在1.15之前，一般是通过history server[1]去拿到最终状态，在1.15之后可以设置这两个Experimental参数
execution.shutdown-on-application-finish[2]
execution.submit-failed-job-on-application-error[3]
设置两个参数的前提条件是必须开启了JobManager的HA[4]
[1]:
history server是不是有延迟性 做不到实时获取任务的状态
| |
阿华田
|
|
a15733178...@163.com
|
签名由网易邮箱大师定制
在2023年07月15日 12:14，casel.chen 写道：
可以查看history server
在 2023-07-14 18:36:42，"阿华田" 写道：
hello 各位大佬， flink on K8S ，运行模式是native 模式，任务状态实时监控遇到一个问题： 就是pod退出了
无法判断flink任务是正常Finished
Hi,
根据上面的异常栈信息，你可以检查一下是否配置了cluster id，在yarn里配置项是`yarn.application.id`
Best,
Shammon FY
On Sat, Jul 15, 2023 at 6:50 PM 杨东树 wrote:
> 您好，
>针对sql-client运行在yarn-session模式报错，现补充相关日志报错信息：
> 2023-07-15 18:43:21,503 INFO org.apache.flink.table.client.cli.CliClient
> [] -
您好，
针对sql-client运行在yarn-session模式报错，现补充相关日志报错信息：
2023-07-15 18:43:21,503 INFO org.apache.flink.table.client.cli.CliClient
[] - Command history file path: /root/.flink-sql-history
2023-07-15 18:43:28,225 INFO org.apache.flink.table.catalog.CatalogManager
Flink社区能否考虑将ROWKIND做成元数据metadata，类似于kafka
connector中的offset和partition等，用户可以使用这些ROWKIND
metadata进行处理，例如对特定ROWKIND过滤或者答输出数据中添加ROWKIND字段
可以查看history server
在 2023-07-14 18:36:42，"阿华田" 写道：
>
>
>hello 各位大佬， flink on K8S ，运行模式是native 模式，任务状态实时监控遇到一个问题： 就是pod退出了
>无法判断flink任务是正常Finished 还是异常失败了，各位大佬有什么建议吗
>| |
>阿华田
>|
>|
>a15733178...@163.com
>|
>签名由网易邮箱大师定制
>
hello 各位大佬， flink on K8S ，运行模式是native 模式，任务状态实时监控遇到一个问题： 就是pod退出了
无法判断flink任务是正常Finished 还是异常失败了，各位大佬有什么建议吗
| |
阿华田
|
|
a15733178...@163.com
|
签名由网易邮箱大师定制
Hi,
我觉得增加到3分钟可能不是一个合适的方法，这会增加作业恢复时间。建议还是追查一下为什么上游task这么长时间没有部署启动成功比较好。
Best,
Shammon FY
On Fri, Jul 14, 2023 at 2:25 PM zhan...@eastcom-sw.com <
zhan...@eastcom-sw.com> wrote:
> hi, 上次将`taskmanager.network.request-backoff.max` 从默认的10s增加到30s后 跑了5天还是出现
> PartitionNotFoundException循环重启
>
Hi, Community. There was an issue that happened to one of our Flink Streaming
jobs using 1.14.3 and that job didn't enable JobManager HA. The issue is after
the only jobManager pod's flink-main-container restarted, some of the
taskManager pods keep throwing the below exception:
INFO
退订
嗨你好，用于sort的中间数据是存储在状态后端当中吗，数据量很大的情况下。
> 2023年7月12日 19:48，weijie guo 写道：
>
> 你好，
> 首先，Batch Shuffle 的中间数据都是会落盘的。其次，对于 Sort 这个操作来说，上面给出的解法和Dataset一致，都不会落盘。
>
> Best regards,
>
> Weijie
>
>
> jinzhuguang 于2023年7月12日周三 17:28写道：
>
>> 如果我的数据量很大，内存装不下，flink在batch
>>
你好，
首先，Batch Shuffle 的中间数据都是会落盘的。其次，对于 Sort 这个操作来说，上面给出的解法和Dataset一致，都不会落盘。
Best regards,
Weijie
jinzhuguang 于2023年7月12日周三 17:28写道：
> 如果我的数据量很大，内存装不下，flink在batch
> mode下的行为是否会像传统的批处理系统，例如hive那样，会进行shuffe、中间数据落盘等操作。
>
> > 2023年7月12日 17:05，weijie guo 写道：
> >
> >
>
如果我的数据量很大，内存装不下，flink在batch mode下的行为是否会像传统的批处理系统，例如hive那样，会进行shuffe、中间数据落盘等操作。
> 2023年7月12日 17:05，weijie guo 写道：
>
> 你好，对于DataSet中不按照key进行全量聚合/排序的API（例如，sortPartition/mapPartition），DataStream上目前没有直接提供相同的API，但可以通过组合DataStream上现有的API实现相同的功能。
> 以mapPartition为例，可以通过以下三个步骤实现相同的功能：
> 1.
你好，对于DataSet中不按照key进行全量聚合/排序的API（例如，sortPartition/mapPartition），DataStream上目前没有直接提供相同的API，但可以通过组合DataStream上现有的API实现相同的功能。
以mapPartition为例，可以通过以下三个步骤实现相同的功能：
1. dataStream.map(record -> (subtaskIndex，
record))，为每个Record增加处理该record时子任务编号。
2.
Hello:
请教2个问题。
1、flink 使用sql-client.sh -f xx.sql 怎么传递参数修改sql里面的文件。比如MySQL,Kafka的连接地址。
2、flink sql消费Kafka
设置group-offset,group.id之前没提交过，会直接报错。怎么设置成没提交过从earliest消费等等。
感谢大家
Flink 社区在这个 thread 讨论了这个问题，之后会出一个 guideline 来帮助用户迁移 DataSet API
[1] https://lists.apache.org/thread/r0y9syc6k5nmcxvnd0hj33htdpdj9k6m
Best regards,
Yuxia
- 原始邮件 -
发件人: "jinzhuguang"
收件人: "user-zh"
发送时间: 星期二, 2023年 7 月 11日 下午 7:16:06
主题: 如果DataSet API 被彻底废掉了，那我如何用DataStream实现分区、排序这个需求？
作业已重启 其他日志暂时没有了
在 2023-07-12 11:06:31，"Shammon FY" 写道：
>Hi
>
>你可以贴一下完整的异常栈信息，这可以帮助定位具体问题
>
>Best,
>Shammon FY
>
>
>On Wed, Jul 12, 2023 at 10:52 AM chenyu_opensource <
>chenyu_opensou...@163.com> wrote:
>
>> 目前是用flink1.12版本，从kafka中读取数据到hdfs，前期运行正常，过段时间报错：
>> Caused by:
Hi
你可以贴一下完整的异常栈信息，这可以帮助定位具体问题
Best,
Shammon FY
On Wed, Jul 12, 2023 at 10:52 AM chenyu_opensource <
chenyu_opensou...@163.com> wrote:
> 目前是用flink1.12版本，从kafka中读取数据到hdfs，前期运行正常，过段时间报错：
> Caused by: org.apache.flink.streaming.runtime.tasks.TimerException:
>
目前是用flink1.12版本，从kafka中读取数据到hdfs，前期运行正常，过段时间报错：
Caused by: org.apache.flink.streaming.runtime.tasks.TimerException:
org.apache.hadoop.ipc.RemoteException(java.io.IOException):
BP-504689274-10.204.4.58-1507792652938:blk_3265799450_2192171234 does not exist
or is not under Constructionnull
flink
Flink社区能否考虑将ROWKIND做成元数据metadata，类似于kafka connector中的offset,
partition，用户可以引用这些metadata进行过滤操作？
在 2023-07-10 23:39:00，"yh z" 写道：
>Hi, shi franke. 你可以尝试自己实现一个 DynamicTableSink，在里面添加参数 “sink.ignore-delete”。
>你可以参考 github 上的一些实现，例如 clickhouse：
目前，我在flink 1.16还没有发现DataStream有实现分区排序的接口，现在是否有方法在有界数据集上实现分区、排序这个需求呢？
目前，我在flink 1.16还没有发现DataStream有实现分区排序的接口，现在是否有方法在有界数据集上实现分区、排序这个需求呢？
你好，整个程序有反压吗
在 2023-07-10 15:32:44，"jiaot...@mail.jj.cn" 写道：
>Hello,
> 我定义了一个pattern （a->b->c->d->e->f->g）在10分钟内匹配，通过在WebUI上查看任务很快在cep节点
> busy(max)100%，我发现通过增加cep节点的并发度并不能解决问题，且checkpoint随着时间的推移状态大小越来越大，数据应该存在大量堆积。数据源同时消费4个kafka
> topic
>
Hi, shi franke. 你可以尝试自己实现一个 DynamicTableSink，在里面添加参数 “sink.ignore-delete”。
你可以参考 github 上的一些实现，例如 clickhouse：
https://github.com/liekkassmile/flink-connector-clickhouse-1.13
shi franke 于2023年7月7日周五 19:24写道：
>
>
退订
Hi,
邮件里的图片看不到
Best,
Shammon FY
On Sun, Jul 9, 2023 at 7:30 PM 杨东树 wrote:
> 各位好，
>目前我在使用flink1.14.5版本的sql-client on
> yarn-session模式时，发现无法正常执行sql任务，日志报如下错误，希望能得到指导，谢谢：
>背景信息：
>1、当flink配置execution.target:
> yarn-per-job时，随后进入sql-client执行sql任务，可正常执行。
>
各位好，
目前我在使用flink1.14.5版本的sql-client on
yarn-session模式时，发现无法正常执行sql任务，日志报如下错误，希望能得到指导，谢谢：
背景信息：
1、当flink配置execution.target: yarn-per-job时，随后进入sql-client执行sql任务，可正常执行。
2、当flink配置execution.target: yarn-session，并启动flink
yarn-session集群，随后进入sql-client执行同样的sql任务，报上图中的错误。
感谢您的回复，这样自定义是可以实现的，我们目前使用的是1.15的flink版本。想看一下社区是不是有在框架层面实现这个配置的支持，理解这应该也是一个相对common的配置
junjie.m...@goupwith.com 于2023年7月7日周五 17:57写道：
> 可以自己用DataStream API通过RowKind进行过滤。
> 如下示例代码：import org.apache.flink.api.common.functions.RichFlatMapFunction;
> import org.apache.flink.types.Row;
> import
咨询下各位大佬，请问下connector现在有支持忽略delete消息的选项配置
吗？比如上游的数据是一个upsert/retract流，在connector这里是否有选项去忽略delete
message,当作append流只去戳里insert消息。我看现在代码没有类似的功能，不确定是否有相关的jira或者实现
您好：
我在尝试使用coGroup对两条流进行连接，使用的依赖版本是flink-streaming-java:1.16.1，流连接的代码如下：
DataStream dataStream3=
dataStream1.coGroup(dataStream2).where(Data1::getKey).equalTo(Data2::getKey)
.window(TumblingEventTimeWindows.of(Time.seconds(5*60)))//窗口大小5min
好的，非常感谢，我试试用 Hint 来控制 join 类型。
发件人: yh z
发送时间: 2023年7月6日 12:10
收件人: user-zh@flink.apache.org
主题: Re: Flink 1.16 流表 join 的 FilterPushDown 及并行
Hi, Chai Kelun, 你的 filter condition 里面包含了你自定义的 UDF，是不满足 filter push down
的条件的，因为对于优化器来说 UDF 是不确定的，优化器不能从里面提取到可以下推的条件,
Hi, aiden. 看起来是类冲突，按照官方的文档，使用 kafka 时，你应该是不需要引入 flink-core 和
flink-connector-base 的（
https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/connectors/table/kafka/）。如果是因为其他原因需要使用这两个
jar， 你可以使用 mvn dependency::tree 查看一下
"org/apache/kafka/clients/consumer/ConsumerRecord"
Hi, Chai Kelun, 你的 filter condition 里面包含了你自定义的 UDF，是不满足 filter push down
的条件的，因为对于优化器来说 UDF 是不确定的，优化器不能从里面提取到可以下推的条件, 如果你想实现下推，可以尝试抽取下确定性的
condition，如 product.id > 10 etc.。另外，Flink 是支持 broadcast hash join
的，如果你想控制某两个表的 join type，你可以通过 join hint 来指定 join 类型为 broadcast。()
Chai Kelun 于2023年7月3日周一
Hi,
我们的做法是启动Flink集群后，在其他节点(pod或者独立启动)启动Sql-Gateway，通过Flink的地址远程连接Flink集群，这样Sql-Gateway的部署和Flink集群完全分开
Best,
Shammon FY
On Tue, Jul 4, 2023 at 10:52 AM chaojianok wrote:
> 大家好，请教个问题。
>
> 用native kubernetes方式在k8s集群上部署好了flink，现在需要在这个flink集群里使用flink sql
> gateway，大家有什么好的方案吗？
> 目前的做法是，进入pod里启动sql
Hi,
如果要增加request
partition的重试时间，可以调整配置项`taskmanager.network.request-backoff.max`，默认是10秒，具体配置可以参阅[1]
[1]
https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/#full-taskmanageroptions
Best,
Shammon FY
On Tue, Jul 4, 2023 at 11:38 AM zhan...@eastcom-sw.com <
Hi,
PartitionNotFoundException异常原因通常是下游task向上游task发送partition
request请求，但是上游task还没有部署成功。一般情况下，下游task会重试，超时后会报出异常。你可以查看下有没有其他的异常日志，查一下上游task为什么没有部署成功。
Best,
Shammon FY
On Tue, Jul 4, 2023 at 9:30 AM zhan...@eastcom-sw.com <
zhan...@eastcom-sw.com> wrote:
>
> 异常日志内容
>
> 2023-07-03 20:30:15,164
退订
有一张 kafka 流表 logClient(id int, name string, price double)，一张实现了
SupportsFilterPushDown 的 customConnector 维表 product(id int, name string, value
double)，实现了自定义函数 MyUDF(double d1, double d2) 用于自定义逻辑计算并支持该算子的下推。
在 Stream-Table Join 的场景下，下列 SQL 并没有将算子进行下推，而是通过 TableScan 将所有算子提到 Join
Hi
可以的，DataStream有很多内置的数据类型，也支持自定义数据类型和数据的序列化反序列化，然后在DataStream的计算内对数据执行计算，可以参考DataStream的官方文档[1]和数据类型文档[2]
[1]
https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/overview/
[2]
请教下，如果用flink进行批计算，使用DataStream API有没有什么优化的地方，是否可以直接将数据作为矩阵在算子之间传递进行计算
mysql库中设置的是utf8mb4编码，单独sql查询mysql表没有出现中文乱码
使用flink datastream作业通过cdc消费mysql
binlog并写到下游doris表时遇到字符串长度超长问题，我们是按mysql表schema创建的doris
schema，就很奇怪为什么总是报字符串超长错误。于是将异常时的原始数据打印出来，才发现数据中只要包含了中文字符都会显示成乱码，要么都是???，要么都是其他莫名字符。
没有人遇到过这个问题吗？
在 2023-06-19 10:41:30，"casel.chen" 写道：
>Flink作业消费数据源来自mysql业务表，后者使用了`-00-00 00:00:00`这个dummy date来存时间，直接用Flink
>CDC消费的话会被解析成`1970-01-01 00:00:00` (mysql中是datetime类型）或者`1970-01-01 08:00:00`
>(mysql中是timestamp类型）。
>问题1：可否给个Flink
感谢姜老师回复，请问如何使用OnlineLogiticRegression生成连续模型表，有样例程序吗
At 2023-06-10 19:18:37, "姜鑫" wrote:
>Hi Xun,
>
>The OnlineLogisticRegression model data is represented as Flink Tables. A
>Flink table can be unbounded and contains continuous model data, so the way to
>update model is just insert new model
我排查了一下，因为任务其实是跑在本地模式上，而我一直没有配置本地模式的slot数量导致slot不足，而这个1024G其实是一个默认值所以出现了需求1T内存这种奇怪的报错。
以往没有出现这种问题是因为以前本地模式会自动分配足够的slot，但flink doris
connecter由于未知的原因没有被计入slot需求中，这就导致缺少一个slot无法达到需求。
> 2023年6月19日 16:18，郭欣瑞 写道：
>
> 我在ide里测试一个任务的时候，任务一直处于created状态，过了很久之后报了以下的错
>
>
退订
在 2023-06-20 11:16:18，"Yanfei Lei" 写道：
>Hi,
>
>从 ResourceProfile{taskHeapMemory=1024.000gb (1099511627776 bytes),
>taskOffHeapMemory=1024.000gb (1099511627776 bytes),
>managedMemory=128.000mb (134217728 bytes), networkMemory=64.000mb
>(67108864 bytes)}, numberOfRequiredSlots=1}]
Hi,
从 ResourceProfile{taskHeapMemory=1024.000gb (1099511627776 bytes),
taskOffHeapMemory=1024.000gb (1099511627776 bytes),
managedMemory=128.000mb (134217728 bytes), networkMemory=64.000mb
(67108864 bytes)}, numberOfRequiredSlots=1}] 来看，sink节点想申请 1T的 heap
memory 和 1T的 off heap
Hi,
这个doris的sink是你自己实现的还是flink或者doris官方提供的？从错误来看，像是sink节点申请了超大的内存资源，你可以确认一下是否有问题，或者是否有配置项可以配置
Best,
Shammon FY
On Mon, Jun 19, 2023 at 4:19 PM 郭欣瑞 wrote:
> 我在ide里测试一个任务的时候，任务一直处于created状态，过了很久之后报了以下的错
>
> DeclarativeSlotPoolBridge.java:351 - Could not acquire the minimum
> required
我在ide里测试一个任务的时候，任务一直处于created状态，过了很久之后报了以下的错
DeclarativeSlotPoolBridge.java:351 - Could not acquire the minimum required
resources, failing slot requests. Acquired:
[ResourceRequirement{resourceProfile=ResourceProfile{taskHeapMemory=1024.000gb
(1099511627776 bytes),
Flink作业消费数据源来自mysql业务表，后者使用了`-00-00 00:00:00`这个dummy date来存时间，直接用Flink
CDC消费的话会被解析成`1970-01-01 00:00:00` (mysql中是datetime类型）或者`1970-01-01 08:00:00`
(mysql中是timestamp类型）。
问题1：可否给个Flink CDC选项，遇到这种dummy时间转成NULL？存量query和增量消费binlog处理这种dummy时间结果一致么？
问题2：如果是mysql -> mysql同步场景，使用Flink
超时输出可以输出这种不满足条件的，但是必须是超时后才能输出
回复的原邮件
| 发件人 | jiaot...@mail.jj.cn |
| 发送日期 | 2023年06月16日 18:04 |
| 收件人 | user-zh |
| 主题 | CEP获取不满足模式事件 |
hello
当我在Flink中定义了一个Patten 例如（a -> b -> c ->
d）四个事件，可以很轻松的获取到满足该模式的所有事件，但是当我想获取该模式的子集缺很复杂（在不重新定义的情况下），
例如我想知道满足（a -> b -> c）模式 、（a ->
多谢多谢
回复的原邮件
| 发件人 | Weihua Hu |
| 日期 | 2023年06月14日 12:32 |
| 收件人 | user-zh@flink.apache.org |
| 抄送至 | |
| 主题 | Re: (无主题) |
>
> 这个状态变量是否需要用transient来修饰
ValueState 再 Rich fuction 的 open 方法中被初始化，不应该被序列化和反序列化，建议使用 transient 来修饰。
但实际上自定义函数的序列化、反序列化只在任务部署阶段执行，而且初始状态下 ValueState 的值是
退订
On Thu, Jun 15, 2023 at 7:23 PM im huzi wrote:
> 退订
>
> On Tue, Jun 13, 2023 at 08:51 casel.chen wrote:
>
> > 线上跑了200多个flink
> >
> sql作业，接了prometheus指标（prometheus定期来获取作业指标）监控后没跑一会儿就将prometheus内存打爆（开了64GB内存），查了一下是因为指标名称过长导致的。
> > flink
> >
>
退订
On Tue, Jun 13, 2023 at 08:51 casel.chen wrote:
> 线上跑了200多个flink
> sql作业，接了prometheus指标（prometheus定期来获取作业指标）监控后没跑一会儿就将prometheus内存打爆（开了64GB内存），查了一下是因为指标名称过长导致的。
> flink
> sql作业的指标名称一般是作业名称+算子名称组成的，而算子名称是由sql内容拼出来的，在select字段比较多或sql较复杂的情况下容易生成过长的名称，
> 请问这个问题有什么好的办法解决吗？
是这个原因，学习了，感谢！
在 2023-06-15 16:25:30，"yuanfeng hu" 写道：
>消费者要设置事务隔离级别
>
>> 2023年6月15日 16:23，163 写道：
>>
>> 据我了解，kafka支持事务，开启checkpoint及exactly-once后仅当checkpoint执行完毕后才能将数据写入kafka中。测试：flink读取kafka的topic
>> a写入topic b，开启checkpoint及exactly-once，flink未执行完新一次的checkpoint，但topic
>>
消费者要设置事务隔离级别
> 2023年6月15日 16:23，163 写道：
>
> 据我了解，kafka支持事务，开启checkpoint及exactly-once后仅当checkpoint执行完毕后才能将数据写入kafka中。测试：flink读取kafka的topic
> a写入topic b，开启checkpoint及exactly-once，flink未执行完新一次的checkpoint，但topic
> b已经可以消费到新数据，这是什么原因？请大家指教！
据我了解，kafka支持事务，开启checkpoint及exactly-once后仅当checkpoint执行完毕后才能将数据写入kafka中。测试：flink读取kafka的topic
a写入topic b，开启checkpoint及exactly-once，flink未执行完新一次的checkpoint，但topic
b已经可以消费到新数据，这是什么原因？请大家指教！
配置参数之后， task name 也会简化.
Best,
Feng
On Wed, Jun 14, 2023 at 11:23 AM casel.chen wrote:
>
>
>
>
>
>
>
>
>
>
>
>
> 谢谢，除了operator name，我看了flink sql作业生成的task name也很长，目前有办法可以简化下吗？例如
>
>
>
>
> 这个状态变量是否需要用transient来修饰
ValueState 再 Rich fuction 的 open 方法中被初始化，不应该被序列化和反序列化，建议使用 transient 来修饰。
但实际上自定义函数的序列化、反序列化只在任务部署阶段执行，而且初始状态下 ValueState 的值是 null，所以不使用 transient
关键字也不会有太大的影响。
以及什么情况下flink代码中需要用transient来修饰变量，什么情况下不用transient来修饰
理解自定义函数的序列化、反序列化是在任务部署阶段执行之后，这个问题就比较好回答了。 如果你的变量在是函数的
谢谢，除了operator name，我看了flink sql作业生成的task name也很长，目前有办法可以简化下吗？例如
在flink处理函数中定义一个状态变量，比如private ValueState
vs；这个状态变量是否需要用transient来修饰，为什么呢？以及什么情况下flink代码中需要用transient来修饰变量，什么情况下不用transient来修饰？请大家指教
hi casel
1. 可以考虑使用 Flink1.15, 使用精简的 operator name
https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/config/#table-exec-simplify-operator-name-enabled
2. Flink 也提供了 restful 接口直接获取瞬时的 metric，如果不需要历史的 metric
线上跑了200多个flink
sql作业，接了prometheus指标（prometheus定期来获取作业指标）监控后没跑一会儿就将prometheus内存打爆（开了64GB内存），查了一下是因为指标名称过长导致的。
flink
sql作业的指标名称一般是作业名称+算子名称组成的，而算子名称是由sql内容拼出来的，在select字段比较多或sql较复杂的情况下容易生成过长的名称，
请问这个问题有什么好的办法解决吗？
谢谢解答，如果是flink sql作业要如何获取到作业中每个算子的latency指标呢？
而且通过prometheus获取作业指标的话，因为flink sql作业中每个算子的名称是按sql内容拼出来的，会出现名称很长，
这样的算子指标直接打到prometheus的话会直接将prometheus内存打爆，这个问题有什么好的办法解决吗？
在 2023-06-12 18:01:11，"Hangxiang Yu" 写道：
>[.[.]]..latency
>这个应该可以满足需求？也可以设置不同的粒度。
Hi!
I think you forgot to upgrade the operator CRD (which contains the updates
enum values).
Please see:
https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/docs/operations/upgrade/#1-upgrading-the-crd
Cheers
Gyula
On Mon, 12 Jun 2023 at 13:38, Liting Liu (litiliu)
wrote:
Hi, I was trying to submit a flink 1.17 job with the flink-kubernetes-operator
version v1.5.0.
But encountered the below exception:
The FlinkDeployment "test-scale-z6t4cd" is invalid: spec.flinkVersion:
Unsupported value: "v1_17": supported values: "v1_13", "v1_14", "v1_15", "v1_16"
I think
[.[.]]..latency
这个应该可以满足需求？也可以设置不同的粒度。
https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/ops/metrics/#io
On Mon, Jun 12, 2023 at 5:05 PM casel.chen wrote:
> 想统计数据经过flink作业各个算子的延迟指标，目前社区开源版能实现吗？
--
Best,
Hangxiang.
想统计数据经过flink作业各个算子的延迟指标，目前社区开源版能实现吗？
我的是flink sql作业，要如何实现你说的方案呢？我看到阿里云实时计算平台VVR是支持展示作业时延指标的，想知道它是如何实现的
在 2023-06-08 16:46:34，"17610775726" <17610775...@163.com> 写道：
>Hi
>
>
>你提到的所有的监控都是可以通过 metric 来监控报警的，至于你提到的 LatencyMarker 因为它不参与算子内部的计算逻辑的时间，所以这个
>metric 并不是准确的，但是如果任务有反压的情况下 LatencyMarker
Hi Xun,
The OnlineLogisticRegression model data is represented as Flink Tables. A Flink
table can be unbounded and contains continuous model data, so the way to update
model is just insert new model data into the Table. Actually,
OnlineLogisticRegression can fit models continuously from
thanks, 设置成GBK是可以的
| |
赵兵杰
|
回复的原邮件
| 发件人 | Leo |
| 日期 | 2023年06月08日 15:16 |
| 收件人 | user-zh@flink.apache.org、Shammon FY |
| 抄送至 | |
| 主题 | Re: pyflink1.17 中文乱码 |
Hi,
这个问题我在1.16版本测试了一下， Pycharm 和 Windows命令行都测试了，结论如下：
1) Windows命令行没有任何问题，因为默认的终端窗口的字符集编码就是GBK;
2)
Hi
你提到的所有的监控都是可以通过 metric 来监控报警的，至于你提到的 LatencyMarker 因为它不参与算子内部的计算逻辑的时间，所以这个
metric 并不是准确的，但是如果任务有反压的情况下 LatencyMarker
也会被阻塞，所以大体上还是可以反应出任务的延迟情况，如果想要准确的计算出端到端的延迟，可以在 消费 kafka 的时候获取一个 start time 时间戳
在 sink 的时候获取一个 end time 时间戳，然后自定义一个 metric 把这个结果上报 基于这个 metric 做端到端的延迟监控。
Best
JasonLee
我想知道当前flink作业延迟了多久现在能通过什么指标可以获取到吗？想通过设置作业延迟告警来反馈作业健康状况，是否产生背压，是否需要增加资源等。
以mysql表实时同步到doris表为例：mysql binlog -> kafka -> flink -> doris
延迟指标包括：
1. 业务延迟：业务延迟=当前系统时间 - 当前系统处理的最后一条数据的事件时间（Event time）
例如：kafka消息写入doris的时间 - kafka消息数据本身产生时间（例如更新mysql记录的时间）
2. 数据滞留延迟：数据滞留时间=数据进入实时计算的时间 - 数据事件时间（Event
Hi,
你是怎么运行的？是不是中文的文件编码格式不对？
Best,
Shammon FY
On Thu, Jun 8, 2023 at 10:07 AM yidan zhao wrote:
> 可以描述再详细点
>
> 1 于2023年6月7日周三 19:55写道：
> >
> > 老师们好，pyflink运行官网例子 wordcount 。把单词改成中文 乱码
> >
> >
> >
> >
> >
>
可以描述再详细点
1 于2023年6月7日周三 19:55写道：
>
> 老师们好，pyflink运行官网例子 wordcount 。把单词改成中文 乱码
>
>
>
>
>
Hi, 目前对RocksDB使用的内存是没有严格限制住的，可以参考这个 ticket：
https://issues.apache.org/jira/browse/FLINK-15532
如果要定位到内存使用情况，可以先看一些粗的Metrics：
https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/#rocksdb-native-metrics
如果要再细致定位到单 instance 内部 RocksDB 的详细内存使用情况，可能需要用 malloc
老师们好，pyflink运行官网例子 wordcount 。把单词改成中文 乱码
Hi All,Im trying OnlineLogisticRegressionExample of flink-ml.How can i update
OnlineLogisticRegression Model continuously?
退订
回复的原邮件
| 发件人 | smq<374060...@qq.com.invalid> |
| 发送日期 | 2023年05月30日 12:22 |
| 收件人 | user-zh |
| 主题 | python 自定义sink |
java中可以继承richainkfunction和checkpointedfunction 两个类实现自定义sink。在python中如何实现这种功能呢
退订请发送任意邮件到 user-zh-unsubscr...@flink.apache.org，可以参考
https://flink.apache.org/community.html#how-to-subscribe-to-a-mailing-list
On Sun, Jun 4, 2023 at 11:13 PM 张保淇 wrote:
> 退订
>
>
> 麻烦尽快帮忙处理
>
>
> Best wishes
> 张保淇
> 电话：+8618878478770
> 邮件：hzuzhangba...@163.com
退订
麻烦尽快帮忙处理
Best wishes
张保淇
电话：+8618878478770
邮件：hzuzhangba...@163.com
一、背景
1、在standalone部署模式下，分别有1个jobmanager和1个taskmanager
2、网络环境要求按端口实际使用开放进出（非NAT）
3、依据flink最新文档要求，有关address和port的设置都已作固定设置
如：jobmanager（cdc-master 192.168.10.11）：18081（web
ui端口）；16123（rpc端口）；50101（metrics端口）；51101（blob端口）
taskmanager（cdc-worker
这个问题比较奇怪，我尝试加大了task.offheap的内存，主要是为了加大 direct memory 的限制，部分作业加大后ok了。
今天又发现一个作业，direct mem 达到9G了还不行（5G
task.offheap额外配置，另外4G是network部分配置的，所以jvm参数部分对direct mem的限制是9G）。
Yuxin Tan 于2023年5月26日周五 18:15写道：
>
> hi, yidan
>
> 除 jvm 参数外，flink 其他配置完全一样吗？比如 state backend 是否有变化？
>
> 另外， jdk11
sql示例：
create table kafka_source() with ('connector'='kafka');
insert into sink_table1 select * from kafka_source;
insert into sink_table2 select * from kafka_source;
报错内容如下：
javax.management.InstanceAlreadyExistsException:
你好，老师，感谢你的回复；
我在MySQL中设置的主键不是自增主键，是一个业务主键的，所以我理解是不是这一块只能是设置并发度为1进行数据的插入
| |
小昌同学
|
|
ccc0606fight...@163.com
|
回复的原邮件
| 发件人 | yidan zhao |
| 发送日期 | 2023年5月31日 12:30 |
| 收件人 | |
| 主题 | Re: flink jdbcsink 连接数的问题 |
你好，这个问题和flink无关，看你主键实现机制吧，如果是自增，那就是mysql级别自动实现的自增，跟flink搭不上关系的。
小昌同学
你好，老师，感谢你的回复；
您说的打印到日志文件，是需要配置flink 的logback.xml 嘛，这一块的配置有一个小demo可以参考嘛
| |
小昌同学
|
|
ccc0606fight...@163.com
|
回复的原邮件
| 发件人 | Shammon FY |
| 发送日期 | 2023年6月1日 10:33 |
| 收件人 | |
| 主题 | Re: flink 输出异常数据 |
Hi
可以看一下报空指针的具体异常栈，如果是你的业务代码，可以在你的处理逻辑里加上一些判断信息并打印到日志文件；如果不是你的业务代码，可以贴一下具体的异常栈信息。
Hi
可以看一下报空指针的具体异常栈，如果是你的业务代码，可以在你的处理逻辑里加上一些判断信息并打印到日志文件；如果不是你的业务代码，可以贴一下具体的异常栈信息。
On Wed, May 31, 2023 at 12:31 PM yidan zhao wrote:
> 这个得靠你自己打日志吧，在可能出NPE的地方 try catch 到，然后打印原始记录。
>
> 小昌同学 于2023年5月29日周一 18:30写道：
> >
> > 你好，数据源是kafka，使用的是stream api
> >
> >
> > | |
> > 小昌同学
> > |
> > |
> >
你在hive的catalog中定义表的时候就可以定义好event time，以及watermark呀。
ZhaoShuKang 于2023年5月25日周四 08:53写道：
>
> 各位老师好，我最近在做Flink查询Hive的功能，需要用到窗口处理数据，在编写代码过程中无法设置水印，我看官网看到Table API & SQL
> 设置事件时间有三种方式：
> 1、在 DDL 中定义
> 2、在 DataStream 到 Table 转换时定义
> 3、使用 TableSource 定义
>
这个得靠你自己打日志吧，在可能出NPE的地方 try catch 到，然后打印原始记录。
小昌同学 于2023年5月29日周一 18:30写道：
>
> 你好，数据源是kafka，使用的是stream api
>
>
> | |
> 小昌同学
> |
> |
> ccc0606fight...@163.com
> |
> 回复的原邮件
> | 发件人 | Weihua Hu |
> | 发送日期 | 2023年5月29日 15:29 |
> | 收件人 | |
> | 主题 | Re: flink 输出异常数据 |
> Hi,
>
>
你好，这个问题和flink无关，看你主键实现机制吧，如果是自增，那就是mysql级别自动实现的自增，跟flink搭不上关系的。
小昌同学 于2023年5月31日周三 09:41写道：
>
> 老师，你好，再请教一下，连接数与并行度有关系的话，如果插入数据的MySQL是有主键的话，是不是连接数据也就是并行度只能为1啦呀，如果是多个并行度的话，可能会造成主键冲突；
> 感谢各位老师的指导
>
>
> | |
> 小昌同学
> |
> |
> ccc0606fight...@163.com
> |
> 回复的原邮件
> | 发件人 | lxk |
> | 发送日期 |
没发现你web ui哪里显示了watermark呢？
小昌同学 于2023年5月31日周三 10:22写道：
>
> 你好，老师，感谢回复，我这边将截图放在了腾讯文档中，请查收；
> 感谢各位老师的指导
> 【腾讯文档】flink web ui
> https://docs.qq.com/sheet/DYkZ0Q0prRWJxcER4?tab=BB08J2
>
>
> | |
> 小昌同学
> |
> |
> ccc0606fight...@163.com
> |
> 回复的原邮件
> | 发件人 | Shammon FY |
> | 发送日期 |
你好，老师，感谢回复，我这边将截图放在了腾讯文档中，请查收；
感谢各位老师的指导
【腾讯文档】flink web ui
https://docs.qq.com/sheet/DYkZ0Q0prRWJxcER4?tab=BB08J2
| |
小昌同学
|
|
ccc0606fight...@163.com
|
回复的原邮件
| 发件人 | Shammon FY |
| 发送日期 | 2023年5月31日 09:59 |
| 收件人 | |
| 主题 | Re: flink web ui显示问题 |
Hi,
好像没有收到附件或者文档，你可以检查确认一下
Best,
Hi,
好像没有收到附件或者文档，你可以检查确认一下
Best,
Shammon FY
On Wed, May 31, 2023 at 9:52 AM 小昌同学 wrote:
> 各位老师好，请教一个关于flink web ui的显示问题；
> 具体的显示异常截图的我以附件的形式放在文档中，我的疑惑是web
> ui上面已经显示watermark，但是看detail的时候显示不是watermark；
> 感谢各位老师指导
>
> 小昌同学
> ccc0606fight...@163.com
>
>
各位老师好，请教一个关于flink web ui的显示问题；
具体的显示异常截图的我以附件的形式放在文档中，我的疑惑是web ui上面已经显示watermark，但是看detail的时候显示不是watermark；
感谢各位老师指导
| |
小昌同学
|
|
ccc0606fight...@163.com
|
老师，你好，再请教一下，连接数与并行度有关系的话，如果插入数据的MySQL是有主键的话，是不是连接数据也就是并行度只能为1啦呀，如果是多个并行度的话，可能会造成主键冲突；
感谢各位老师的指导
| |
小昌同学
|
|
ccc0606fight...@163.com
|
回复的原邮件
| 发件人 | lxk |
| 发送日期 | 2023年5月30日 14:30 |
| 收件人 | |
| 主题 | Re:flink jdbcsink 连接数的问题 |
hi,
共有 16722 项搜索結果，以下是第 601 - 700 matches
Mail list logo
