Hi,
方便提供一下复现的用例吗?
Best,
Jiabao
On 2023/08/21 02:19:53 guifeng huang wrote:
> (Flink1.15版本)
> 咨询求助: Least函数输入参数(Double类型)正常, 在Flink shell里测试函数无问题, 结果符合预期.
> 但是实际生产流里进行使用的时候发现返回结果有异, 以下是3种case
> - 返回结果正确, 符合预期
> - 返回0, 不符合预期, 未知原因
> - 返回结果和理论正确值有微小的gap, 找了几个case都是1位数值里的差距.
> 看看有没有其他的老师遇到过同样的问题
(Flink1.15版本)
咨询求助: Least函数输入参数(Double类型)正常, 在Flink shell里测试函数无问题, 结果符合预期.
但是实际生产流里进行使用的时候发现返回结果有异, 以下是3种case
- 返回结果正确, 符合预期
- 返回0, 不符合预期, 未知原因
- 返回结果和理论正确值有微小的gap, 找了几个case都是1位数值里的差距.
看看有没有其他的老师遇到过同样的问题
问题一：
做checkpoint时，是每个算子收到barriers之后，将状态和offset写到状态后端，并返回ack给jm之后。再做一次全量快照到jm内存或者自己设置的hdfs文件路径下啊。不理解在hdfs生成的checkpoint文件到底是2pc提交事务成功之后的checkpoint还是每个算子做完checkpoint。
是图1：
还是图二：
问题二：
做完2pc之后。出现了故障。做故障恢复。恢复的状态是上一次提交事务成功的地方的状态。还是上一个barriers所在算子做的checkpoint成功的地方开始恢复。
| |
zyzandmz
|
|
可以 savepoint 到 HDFS，然后配置 checkpoint 的地址为 对象存储。
我们就是 flink 支持对象存储和 HDFS。
Hangxiang Yu 于2023年8月2日周三 14:03写道：
> Hi, 我理解可以有两种方式：
> 1. 设定从某个存储集群上恢复并向另一个存储集群上快照，即设置[1]为 HDFS地址，[2] 为后面的对象存储地址
> 2. 还是在HDFS集群上启停作业，设置 savepoint 目录[3]到对象存储
>
> 关于 state processor api，目前 sql 作业确实操作起来比较困难，只能从日志里获取 uid
1 可控范围即可。
2 分析阶段可以分开，实际运行阶段看情况，怎样性能高就如何搞。
3 看监控，flink web ui有根据每个节点的反压情况按照不同颜色展示。
星海 <2278179...@qq.com.invalid> 于2023年8月16日周三 22:03写道：
>
> hello。大家好，请教几个问题：
> 1、flink中背压存在是合理的吗？还是在可控范围内就行？还是尽可能没有呢？
> 2、如果出现背压，如果多个operator chain 在一起不好分析，需要先将其拆开分析吗？
>
The Apache Flink community is very happy to announce the release of Apache
Flink Kubernetes Operator 1.6.0.
The Flink Kubernetes Operator allows users to manage their Apache Flink
applications and their lifecycle through native k8s tooling like kubectl.
Release highlights:
- Improved rollback
退订
请问，flink1.17在使用RichFlatMapFunction进行批计算时，如何在数据结束时将状态写入输出的数据流中？
谢谢
退订
退订
HI
我简化了我python代码
只要udft方法有外部方法,都会有递归问题, 比如
agan_add_iig(),尽管我的agan_add_iig()实现很简单,flink难道不能外部import自定义方法吗??
def agan_add_iig():
return 2
@udtf(input_types=DataTypes.STRING(),
result_types=[DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING(),
DataTypes.STRING()])
def run(data_str):
try:
各位老师好:
背景是这样的[flink1.17.1],我在window机器,本地单机调用自定义的pythonUDF,下面是我python代码
err=None
@udtf(input_types=DataTypes.STRING(),
result_types=[DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING(),
DataTypes.STRING()])
def run(data_str):
try:
logger.info("input param is ", data_str)
好滴呀 谢谢各位老师指导
| |
小昌同学
|
|
ccc0606fight...@163.com
|
回复的原邮件
| 发件人 | Yanfei Lei |
| 发送日期 | 2023年8月10日 11:50 |
| 收件人 | |
| 主题 | Re: Flink 窗口触发条件 |
hi,
hi,
感觉和[1]的问题比较像，事件时间的window在onElement和onEventTime时会触发，这两个方法又会根据watermark判断，可以看看o.a.f.table.runtime.operators.window.triggers包和o.a.f.table.runtime.operators.wmassigners包。
[1] https://juejin.cn/post/6850418110010179597
小昌同学 于2023年8月10日周四 10:52写道：
>
>
各位老师好，我这边在使用Flink的事件时间窗口的时候，关于窗口触发的条件我有一点疑问想咨询一下各位老师
我是开了一个2分钟的事件时间的窗口，但是等到两分钟后窗口并没有主动触发，等我后面再发一条数据的时候，窗口再进行了触发
所以我想请问一下窗口的触发机制不是时间点嘛，而是非要等到下一条数据发送，依赖于下一条数据携带的时间戳大于窗口的结束时间，上一个窗口才会真正的触发嘛
请各位老师指导一下
| |
小昌同学
|
|
ccc0606fight...@163.com
|
退订
好的 谢谢各位老师的指导
| |
小昌同学
|
|
ccc0606fight...@163.com
|
回复的原邮件
| 发件人 | ron |
| 发送日期 | 2023年8月10日 00:51 |
| 收件人 | |
| 主题 | Re: 回复： Flink消费MySQL |
Hi,
建议通过CDC实时读，然后用Flink的双流Join进行关联。
-原始邮件-
发件人: "小昌同学"
发送时间: 2023-08-08 11:10:19 (星期二)
收件人: user-zh
抄送: user-zh
主题: 回复：
Hi,
这个报错看起来确实很奇怪，你可以给一个完整的使用SQL吗，这样可以方便去复现一下。
> -原始邮件-
> 发件人: "junjie.m...@goupwith.com"
> 发送时间: 2023-08-03 17:05:03 (星期四)
> 收件人: user-zh
> 抄送:
> 主题: 自定义ScalarFunction函数运行报错
>
> public static class MatrixToRowFunction extends ScalarFunction {
> @DataTypeHint("ROW>")
> public
Hi,
建议通过CDC实时读，然后用Flink的双流Join进行关联。
> -原始邮件-
> 发件人: "小昌同学"
> 发送时间: 2023-08-08 11:10:19 (星期二)
> 收件人: user-zh
> 抄送: user-zh
> 主题: 回复： Flink消费MySQL
>
> 谢谢老师指导呀；
> 我目前的需求是想把两张MySQL的表数据读取出来，然后进行实时关联，我现在能想到的就是要么使用cdc实时读取，要么就是写一个循环去读MySQL中的数据
> 老师这一块有更好的建议嘛
>
>
> | |
> 小昌同学
> |
> |
>
像上面提到的，目前可能直接使用CDC是一个比较好的方案，自己读数据会有很多问题，比如update数据如何读取、如何读取增量数据、如何处理failover等，还是直接使用CDC最方便
Best,
Shammon FY
On Tue, Aug 8, 2023 at 11:30 AM Jiabao Sun
wrote:
> Hi,
>
> 可以尝试使用 flink-cdc-connectors 去实时关联。
> 使用 regular join 需要保留两张表完整的状态，表数据量较大建议使用 rocksdb backend。
> 被关联的表变化不大的话可以考虑 lookup join。
>
Hi,
可以尝试使用 flink-cdc-connectors 去实时关联。
使用 regular join 需要保留两张表完整的状态，表数据量较大建议使用 rocksdb backend。
被关联的表变化不大的话可以考虑 lookup join。
Best,
Jiabao
> 2023年8月8日 上午11:10，小昌同学 写道：
>
> 谢谢老师指导呀；
> 我目前的需求是想把两张MySQL的表数据读取出来，然后进行实时关联，我现在能想到的就是要么使用cdc实时读取，要么就是写一个循环去读MySQL中的数据
> 老师这一块有更好的建议嘛
>
>
> | |
> 小昌同学
谢谢老师指导呀；
我目前的需求是想把两张MySQL的表数据读取出来，然后进行实时关联，我现在能想到的就是要么使用cdc实时读取，要么就是写一个循环去读MySQL中的数据
老师这一块有更好的建议嘛
| |
小昌同学
|
|
ccc0606fight...@163.com
|
回复的原邮件
| 发件人 | Shammon FY |
| 发送日期 | 2023年8月8日 10:37 |
| 收件人 | |
| 主题 | Re: Flink消费MySQL |
Hi，
你代码里的ResultSet读取完成后需要将resultSet关闭掉，避免资源泄漏
Hi，
你代码里的ResultSet读取完成后需要将resultSet关闭掉，避免资源泄漏
至于你提到的Mysql数据读完程序就结束具体是指哪块？mysql是bounded
source，数据消费完成并且整个作业计算完成后，就会结束，这是正常情况
Best,
Shammon FY
On Mon, Aug 7, 2023 at 5:04 PM 小昌同学 wrote:
> 各位老师好
> ，我这边在本地使用通过继承RichSourceFunction类，实现从MySQL中读取数据，但是为啥程序将MySQL中的全部数据获取出来后，程序就自动停止了啊；
> 以下是我的代码：
> |
>
如果不是用的flink kubernetes operator或者hdfs和oss系统网络无法直接连通怎么办？
有没有办法读取hdfs的checkpoint/savepoint然后再另存为oss的checkpoint/savepoint呢？谢谢！
在 2023-08-07 10:33:25，"Ruibin Xing" 写道：
>你好，
>
>如果你们也使用的是官方的Flink Kubernetes
Hi,
看着像是版本冲突了，你有在你的flink session集群目录里放hbase的包吗？可以检查一下跟flink hbase
shaded的hbase版本是否一致
Best,
Shammon FY
On Sat, Aug 5, 2023 at 9:33 PM 杨东树 wrote:
> 各位好，
>目前使用sql-client查询hbase数据时，无法查询成功，麻烦指导下，谢谢。
>复现方法：
> 1、hbase操作:
> hbase(main):005:0> create 'flink_to_hbase','cf1'
> 0 row(s)
各位老师好
，我这边在本地使用通过继承RichSourceFunction类，实现从MySQL中读取数据，但是为啥程序将MySQL中的全部数据获取出来后，程序就自动停止了啊；
以下是我的代码：
|
public class MysqlSource2 extends RichSourceFunction {
PreparedStatement ps;
private Connection connection;
@Override
public void open(Configuration parameters) throws Exception {
你好，
如果你们也使用的是官方的Flink Kubernetes
Operator，可以参考我们迁移的经验：迁移的时候设置FlinkDeployment的initalSavepoint为HDFS上Savepoint的地址，同时配置savepoint/checkpoint目录为OSS。这样Flink启动的时候会从HDFS中的状态恢复，并将新的checkpoint保存在oss中。
On Sun, Aug 6, 2023 at 10:03 PM casel.chen wrote:
> flink on
flink on yarn作业checkpoint/savepoint保存在hdfs上面，现在想将其迁移到on
k8s上运行，使用的是对象存储oss，请问如何无感地进行作业状态迁移呢？使用的flink版本是1.15.2，谢谢！
各位好，
目前使用sql-client查询hbase数据时，无法查询成功，麻烦指导下，谢谢。
复现方法：
1、hbase操作:
hbase(main):005:0> create 'flink_to_hbase','cf1'
0 row(s) in 2.2900 seconds
hbase(main):006:0> put 'flink_to_hbase', 'rk0001', 'cf1:username', 'zhangsan'
0 row(s) in 0.0510 seconds
2、flink操作:
./start-cluster.sh
Hi,
一般需要确认一下是哪块引起的fullgc，比如metaspace还是堆内存过大导致的。如果是堆内存过大导致的，可以将内存dump下来，用一些分析工具例如mat、visualvm等具体查看一下哪些对象占比比较多，是否存在内存泄漏等原因
Best,
Shammon FY
On Fri, Aug 4, 2023 at 10:00 AM yidan zhao wrote:
> GC日志看GC原因
>
> 2278179732 <2278179...@qq.com.invalid> 于2023年8月3日周四 13:59写道：
> >
> >
Regular Join 默认把数据都存储在State中，通常会结合TTL来进行优化
guanyq 于2023年8月3日周四 15:59写道：
> 请问下多个表关联，这种flink sql如何优化呢，直接关联优点跑不动RuntimeExecutionMode.STREAMING 模式
>
> select
> date_format(a.create_time, '-MM-dd HH:mm:ss') as create_time,
> b.vehicle_code,
> a.item_name,
>
GC日志看GC原因
2278179732 <2278179...@qq.com.invalid> 于2023年8月3日周四 13:59写道：
>
> 大家好，请问下作业跑一段时间就会偶发出现背压，full gc看着很严重，有什么好的工具排查下吗？或者经验文档？谢谢！
请问下多个表关联，这种flink sql如何优化呢，直接关联优点跑不动RuntimeExecutionMode.STREAMING 模式
select
date_format(a.create_time, '-MM-dd HH:mm:ss') as create_time,
b.vehicle_code,
a.item_name,
a.item_value,
c.item_value as vehicle_score,
d.current_fault,
首先你窗口是30min，刚刚开始肯定会是涨的。
其次，后续稳定后，继续涨可能是因为流量在变化。
最后，流量不变情况下，还可能受到延迟的影响。
lxk 于2023年7月25日周二 11:22写道：
>
> 相关配置：
> Flink:1.16
>
> | Checkpointing Mode | Exactly Once |
> | Checkpoint Storage | FileSystemCheckpointStorage |
> | State Backend | EmbeddedRocksDBStateBackend |
> | Interval | 8m 0s |
>
>
>
这个取决于你是什么模型，比如python中sklearn的大多模型都可以导出成pmml格式模型，然后java用jpmml库就可以导入进行预测。
如果是tensorflow模型，也有，只不过我忘记了，你可以找找。
15904502343 <15904502...@163.com> 于2023年8月1日周二 16:48写道：
>
> 您好
> 我想知道是否有代码示例，可以在Flink程序中加载预先训练好的编码模型(用python编写)
你好，
不需要将所有的依赖都改为snapshot，仅需要将我们项目内的版本加上 snapshot 后缀。
可以在项目中统一替换版本号 1.x.x -> 1.x.x-SNAPSHOT，或者使用 mvn versions:set
-DnewVersion=1.x.x-SNAPSHOT 设置。
> 2023年8月2日 下午2:25，jinzhuguang 写道：
>
>
非常感谢你的提醒，我现在用maven工具修改了所有的版本号为snapshot，但是flink-connectors(connectors的父模块)也变成snapshot，打包的时候仓库里找不到他了，而且也没法想flink-runtime这些包手动改下版本好，这种该怎么办
> 2023年7月27日 11:05，Jiabao Sun 写道：
>
> 你好，
>
> 通常在 pom 中引入 maven-deploy-plugin，并且通过 声明私服地址，使用 mvn
> clean deploy 命令部署到nexus私服。
> 部署到 SNAPSHOT 仓库需要项目版本号包含
Hi, 我理解可以有两种方式：
1. 设定从某个存储集群上恢复并向另一个存储集群上快照，即设置[1]为 HDFS地址，[2] 为后面的对象存储地址
2. 还是在HDFS集群上启停作业，设置 savepoint 目录[3]到对象存储
关于 state processor api，目前 sql 作业确实操作起来比较困难，只能从日志里获取 uid 等信息，以及理解 sql
实际产生的状态才能使用；
[1]
/opt/flink/flink-1.17.1/bin/flink run-application -t yarn-application -yjm
1024m -ytm 1024m ./xx-1.0.jar
./config.properties以上提交命令制定的配置文件，为什么在容器内找配置文件？file
/home/yarn/nm/usercache/root/appcache/application_1690773368385_0092/container_e183_1690773368385_0092_01_01/./config.properties
does
*退订*
您好
我想知道是否有代码示例，可以在Flink程序中加载预先训练好的编码模型(用python编写)
我们要将当前在Hadoop Yarn上运行的flink
sql作业迁移到K8S上，状态存储介质要从HDFS更换到对象存储，以便作业能够从之前保存点恢复，升级对用户无感。
又因为flink作业状态文件内容中包含有绝对路径，所以不能通过物理直接复制文件的办法实现。
查了一下官网flink state processor api目前读取状态需要传参uid和flink状态类型，但问题是flink
sql作业的uid是自动生成的，状态类型我们也无法得知，请问有没有遍历目录下保存的所有状态并将其另存到另一个文件系统目录下的API ? 感觉state
processor
Hi，
如题，请教一下关于如何使用DataStream API实现有界流的join操作，我在调用join的时候必须要window，怎么避免，还是需要使用SQL
API才可以
感谢，
鱼
这个解决不了根本问题 主要是我们的任务比较多，业务上就需要保留几千个任务
| |
阿华田
|
|
a15733178...@163.com
|
签名由网易邮箱大师定制
在2023年07月28日 11:28，Shammon FY 写道：
Hi,
可以通过配置`jobstore.max-capacity`和`jobstore.expiration-time`控制保存的任务数，具体参数可以参考[1]
[1]
Hi
Flink UI 需要加载所有的 Job 信息并在 UI 渲染，在作业比较多的时候很容易导致 UI 卡死。
不只在这个页面，在一些并发比较大的任务上打开 subtask 页面也很容易导致UI 卡死。
Flink UI 需要一个分页的功能来减少数据加载和 UI 渲染的压力
Best,
Weihua
On Fri, Jul 28, 2023 at 11:29 AM Shammon FY wrote:
> Hi,
>
>
> 可以通过配置`jobstore.max-capacity`和`jobstore.expiration-time`控制保存的任务数，具体参数可以参考[1]
Hi,
可以通过配置`jobstore.max-capacity`和`jobstore.expiration-time`控制保存的任务数，具体参数可以参考[1]
[1]
https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/#full-jobmanager-options
Best,
Shammon FY
On Fri, Jul 28, 2023 at 10:17 AM 阿华田 wrote:
> 目前flink-job-history
>
目前flink-job-history 已经收录5000+任务，当点击全部任务查看时，job-history就会卡死无法访问，各位大佬有什么好的解决方式？
| |
阿华田
|
|
a15733178...@163.com
|
签名由网易邮箱大师定制
退订
On Thu, Jul 27, 2023 at 2:40 PM 许琦 wrote:
> *退订*
*退订*
你好，
通常在 pom 中引入 maven-deploy-plugin，并且通过 声明私服地址，使用 mvn
clean deploy 命令部署到nexus私服。
部署到 SNAPSHOT 仓库需要项目版本号包含 -SNAPSHOT 后缀，可以在IDE中全局替换，也可以使用 versions-maven-plugin
统一设置。
org.apache.maven.plugins
maven-deploy-plugin
2.8.2
我是基于flink 1.16.0开发的，由于版本号没有snapshot，现在又无法发布release版本的，我该怎么办？
好的，谢谢老师
在 2023-07-26 21:04:20，"Jiabao Sun" 写道：
>SqlSession 需要关闭，建议使用 SqlSessionManager，可以不用手动关闭 SqlSession。
>
>
>On 2023/07/18 02:13:16 lxk wrote:
>> 在flink内需要使用mybatis做些简化查询的工作，目前我的使用方式如下
>>
>> public class MybatisUtil {
>>
>> private static final Logger LOGGER =
>>
SqlSession 需要关闭，建议使用 SqlSessionManager，可以不用手动关闭 SqlSession。
On 2023/07/18 02:13:16 lxk wrote:
> 在flink内需要使用mybatis做些简化查询的工作，目前我的使用方式如下
>
> public class MybatisUtil {
>
> private static final Logger LOGGER =
> LogFactory.createNewLogger("MybatisUtil");
> private static ThreadLocal tl = new
退订
| |
lei-tian
|
|
totorobabyf...@163.com
|
你好，感谢回复。我使用reduce解决了问题。
祝好运。
发件人: weijie guo
发送时间: 2023年7月26日 10:50
收件人: user-zh@flink.apache.org
主题: Re: 关于DataStream API计算批数据的聚合值
你好：
Batch 模式下的 reduce 操作默认应该就是只输出最后一条数据(per-key)的。Agg 的话可能有点麻烦，可以使用
GlobalWindow + 自定义 Trigger 来 Workaround.
Best regards,
退订
wang <24248...@163.com> 于2023年7月13日周四 07:34写道：
> 退订
--
Best Regards,
*Yaohua Wang 王耀华*
School of Software Technology, Xiamen University
Tel: (+86)187-0189-5935
E-mail: wangyaohua2...@gmail.com
*退订*
你好，感谢老师回复
`insert into error_md5_info (action, serverIp,
handleSerialno,md5Num,insertTime,dateTime) values
(1,2,3,4,5,6),(1,2,3,4,9,10);`或者分成两条数据插入，我理解这两种情况的话，对于数据库来说，就是一次插入与两次插入的问题了吧，要是数据量大的话，感觉对性能还是有影响的
| |
小昌同学
|
|
ccc0606fight...@163.com
|
回复的原邮件
| 发件人 | Shammon FY |
| 发送日期 |
Hi,
目前JdbcSink会为每个Sink创建PreparedStatement，当进行batch数据处理时，会先调用PreparedStatement的addBatch()函数将数据放入缓存，到达flush条件后调用executeBatch()函数批量发送数据到jdbc
server，这样会节省网络IO。
具体到数据库侧，我理解执行 `insert into error_md5_info (action, serverIp,
handleSerialno,md5Num,insertTime,dateTime) values
你好：
Batch 模式下的 reduce 操作默认应该就是只输出最后一条数据(per-key)的。Agg 的话可能有点麻烦，可以使用
GlobalWindow + 自定义 Trigger 来 Workaround.
Best regards,
Weijie
Liu Join 于2023年7月26日周三 09:10写道：
> 例如：我使用DataStream api计算批数据也就是有界流的平均值，如何实现只输出最后一条平均值的数据，不输出中间值
>
Hi,
跟使用普通流式作业的DataStream用法一样，只需要在RuntimeMode里使用Batch模式，Flink在Batch模式下会只输出最后的结果，而不会输出中间结果。具体可以参考Flink里的WordCount例子
[1]
[1]
https://github.com/apache/flink/blob/master/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java
On Wed, Jul
例如：我使用DataStream api计算批数据也就是有界流的平均值，如何实现只输出最后一条平均值的数据，不输出中间值
各位老师好，我这边在使用Flink 的JdbcSink的时候，有一个疑问想请教一下各位老师：
我的代码如下:我代码中设定的每一个批次插入1000条，或者是每隔200ms插入一次数据，但是由于我司musql资源不行，从监控页面看插入的IO过高，我想请教一下我使用这样的insert语句，当我积累了1000条，是怎么样的格式
是
insert into error_md5_info (action, serverIp,
handleSerialno,md5Num,insertTime,dateTime) values (1,2,3,4,5,6),(1,2,3,4,9,10);
或者是
相关配置：
Flink:1.16
| Checkpointing Mode | Exactly Once |
| Checkpoint Storage | FileSystemCheckpointStorage |
| State Backend | EmbeddedRocksDBStateBackend |
| Interval | 8m 0s |
我有一个程序，主要是用来统计一些热门商品之类的数据
具体代码如下：
.keyBy(data -> data.getShopId() + data.getYh_productid())
Hi, Jiacheng:
helm也是把values.yaml和你在命令行中传入的参数，代入到一些k8s的yaml模版里，render出来一个yaml文件，然后提交到k8s的。
这些是flink kubernetes operator的helm的模版。
https://github.com/apache/flink-kubernetes-operator/tree/main/helm/flink-kubernetes-operator/templates
你完全可以自己人肉修改这些yaml模版，然后提交到k8s，当然这很麻烦，也很容易出错。
Hi,
运行的是哪个例子？从错误上看是在从MiniCluster获取结果的时候，MiniCluster被关闭了
Best,
Shammon FY
On Sat, Jul 22, 2023 at 3:25 PM guanyq wrote:
> 本地IDEA运行 MiniCluster is not yet running or has already been shut down.
> 请问是什么原因，如何处理
>
>
>
>
> 15:19:27,511 INFO
>
退订
请问没有helm的情况下能否安装flink-k8s-operator？安装operator是否可以不要clusterrole
本地IDEA运行 MiniCluster is not yet running or has already been shut down.
请问是什么原因，如何处理
15:19:27,511 INFO
org.apache.flink.runtime.resourcemanager.ResourceManagerServiceImpl [] -
Stopping resource manager service.
15:19:27,503 WARN
请问，flink sql 能否通过sql语句将mysql表加载为flink 内存表
sql语句为多表关联
退订
代码如下：
package com.didichuxing.iov.func
import com.didichuxing.iov.util.JedisTool
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.async.{ResultFuture,
RichAsyncFunction}
import org.slf4j.LoggerFactory
import redis.clients.jedis.Jedis
import
我也遇到类似的问题，我是链接失效了，最后没办法再注册了
> 2023年7月20日 14:54，李天龙 写道：
>
> 您好！
> 我想注册一个flink jira的账号，但由于提出的里有不充分给拒掉了，想再次申请，却提示邮箱已申请过，还未处理：
>
>
> There is already a pending Jira account request associated with this email
> address. Please wait for it to be processed
>
>
> 请问怎么解决这个问题，并且成功申请一个账号
>
>
>
>
您好！
我想注册一个flink jira的账号，但由于提出的里有不充分给拒掉了，想再次申请，却提示邮箱已申请过，还未处理：
There is already a pending Jira account request associated with this email
address. Please wait for it to be processed
请问怎么解决这个问题，并且成功申请一个账号
--
发自我的网易邮箱平板适配版
异常栈信息
```
2023-07-20 11:43:01,627 ERROR
org.apache.flink.runtime.taskexecutor.TaskManagerRunner [] - Terminating
TaskManagerRunner with exit code 1.
org.apache.flink.util.FlinkException: Failed to start the TaskManagerRunner.
at
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
共有 16798 项搜索結果，以下是第 601 - 700 matches
Mail list logo
