回复： CheckPoint Dir 路径下引发的一些问题

Hi 麻烦使用第三方图床，把图片链接过来，直接贴图片的话显示不出来 --原始邮件-- 发件人:"Weihua Hu"

Re: CheckPoint Dir 路径下引发的一些问题

HI, Px New 1. Checkpoint 保存数量可以通过参数： state.checkpoints.num-retained 来控制，默认是 1 2. _metadata 只是一些元数据，保存了state 的句柄，其他文件是 state 数据，由各 Task 在触发 checkpoint 的时候上传。相反在恢复 checkpoint 的时候JM 读取_metadata 将相应句柄下发到 Task，Task 通过远端 HDFS 拉取对应的 state。 Best Weihua Hu > 2020年6月5日 13:36，Px New

Re: flink sql ddl能设置ttl吗？

Hi, 第一个问题，即将发布的1.11可以设定primary key, 不用再推断primary key，除pk外的字段也没有限制，当前你可以加firt_value,last_value等函数取非group by的其他字段, jdbc的文档在撰写中了[1] 第二个问题，目前应该还不支持通过SET在DDL里设置ttl的. Best, Leonard Xu [1] https://issues.apache.org/jira/browse/FLINK-17829 [2]

CheckPoint Dir 路径下引发的一些问题

Hi everyOne 有一个关于CheckPoint相关的一个问题: 1.我在项目中使用的状态后端为:Fsstatebackend 2.我在jobManager的log输出找到相应的job ID后 去对应的HDFS 找到了对应的chk目录 3.但我有两个疑问: 3.1.没有设置 chk的存储数默认是多保留多少份呢(我这边看到保留了近20次的chk)? 3.2 当我点进具体的chk-id 后 发现有很多文件[见2图] 我清楚的是当任务发生异常后tesk 会从hdfs 将_metadata 下载后进行任务恢复操作的,那其他的哪些文件是如何产生的?以及有什么作用呢? 期待回复:

Re: flink sql ddl能设置ttl吗？

Hi， 1. 如果我没记错的话，mysql的写入应该就是replace的方式写入的，应该不需要group by来实现upsert写入吧，除非你想要的是聚合结果。 2. 这个看你用的是哪种方式，如果是table api，可以直接用你说的这种方式来设置；如果用的是sql-client，可以直接在配置文件里面配置。 xu yihan 于2020年6月4日周四 下午6:07写道： > 各位好， > > 我有一个需求是要维表关联mysql数据，再upsert写入mysql，就是个丰富字段的简单需求。 > > 现在有两个问题： >

Re: Batch 模式 Table API 增加cache算子

这个内容会放在 FLIP-36 [1]中讨论，估计会在 1.12 的计划中。 cc @Becket Qin who is the driver of this FLIP. Best, Jark [1] FLIP-36: https://cwiki.apache.org/confluence/display/FLINK/FLIP-36%3A+Support+Interactive+Programming+in+Flink [2] 讨论邮件:

Re: flink1.10整合hbase测试遇到的问题

FYI Leonard's reply in another thread: = Flink hbase connector官网有的[1],1.10以前的包名是flink-hbase, 1.11后社区统一了所有connector的包名，包名为flink-connector-hbase[1]，1.11最近快发布了，你的报错看起来是缺少了一些类，可以参考下hbase connector的依赖[2]。 [1]

Re: Re: standalone模式下metaspace内存溢出

你好， 社区已经发现过一些与 metaspace 泄漏相关的问题，其中部分已经修复，还有一些则与第三方依赖有关。由于你的图片没有显示出来，我不确定你遇到的是否是相同的问题。 此外，在即将发布的 1.11 中，Flink 针对这一问题进行了优化，作业会采用单独的 ClassLoader 运行，以避免作业结束之后 metaspace 还有泄漏的问题。1.11.0 已经进入发布前的测试阶段，RC1 已经发布，欢迎试用。 Thank you~ Xintong Song On Fri, Jun 5, 2020 at 9:29 AM 胡泽康 wrote: > flink版本：1.10

Re: Flink 1.6 版本，RocksDBStateBackend ,增量 Checkpoint 目录问题

对 我认为这也不是很合理, 不知道有什么更好的解释吗 LakeShen 于2020年1月16日周四 下午5:06写道： > 原来是这样，中间的确有很多 Checkpoint 超时失败，对于这种由于 Checkpoint 失败创建的目录，Flink 本身是不会删除的，对吗？ > 也就是这部分状态文件，会一直存储在HDFS 上面，这种情况是否会造成 Flink Checkpoint 目录下 shared目录无限增大呢， > 非常感谢你，lucas，期待你的回复。 > > lucas.wu 于2020年1月16日周四 下午4:50写道： > > >

??????flink1.9 Sql ????????????????????????state??????

?? ---- ??:"zhiyezou"<1530130...@qq.com; :2020??6??5??(??) 10:31 ??:"user-zh"

??????flink1.9 Sql ????????????????????????state??????

Hi ??TTL ---- ??:"star"<3149768...@qq.com; :2020??6??5??(??) 10:22 ??:"user-zh@flink.apache.org"

Re: standalone模式下metaspace内存溢出

那我先试试1.11. thank you~ --Original-- From: "Xintong Song"

flink1.9 Sql ????????????????????????state??????

?? flink 1.9??blink planner ?? ?? id hbasemonthtable rowkey??month+city ?? city ??hbasetotalTable ??rowkey?? city 18?? restore??ck??

Re: standalone模式下metaspace内存溢出

你好， 社区已经发现过一些与 metaspace 泄漏相关的问题，其中部分已经修复，还有一些则与第三方依赖有关。由于你的图片没有显示出来，我不确定你遇到的是否是相同的问题。 此外，在即将发布的 1.11 中，Flink 针对这一问题进行了优化，作业会采用单独的 ClassLoader 运行，以避免作业结束之后 metaspace 还有泄漏的问题。1.11.0 已经进入发布前的测试阶段，RC1 已经发布，欢迎试用。 Thank you~ Xintong Song On Fri, Jun 5, 2020 at 9:29 AM 胡泽康 wrote: > flink版本：1.10

回复： flink sql upsert模式写入mysql，es等key一定是groupby之后所有字段吗

Hi 这里存在一个问题是，使用了last_value或者first_value这样的udaf，但是如果多条数据来到经过udaf处理后结果还是和之前一样的情况下，是不会产出回撤流数据的，可以观察下你是否需要考虑这种情况。 Best, Yichao Yang --原始邮件--

Re: flink整合hbase

Hi, Flink hbase connector官网有的[1],1.10以前的包名是flink-hbase, 1.11后社区统一了所有connector的包名，包名为flink-connector-hbase[1]，1.11最近快发布了，你的报错看起来是缺少了一些类，可以参考下hbase connector的依赖[2]。 Best, Leonard Xu [1] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connect.html#hbase-connector

回复：回复: flink整合hbase

hi, sink是自己的,我还没发现flink有原生的hbase的sink,正在研究. 在2020年6月5日 09:06，xueaohui_...@163.com 写道： Sink 是自己的还是官方的。 xueaohui_...@163.com 发件人： liunaihua521 发送时间： 2020-06-05 00:27 收件人： user-zh@flink.apache.org 主题： flink整合hbase hi

回复： flink整合hbase

hi, 您能说下您本地的各组件版本吗?谢谢了 报错内容(不方便看附件有文本): org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: org.apache.flink.client.program.ProgramInvocationException: Job failed (JobID: cacd929120e6cefa79e1f8cca65e3d3f) at

Re:flink整合hbase

你好，可否提供一下报错信息呢？ 在 2020-06-05 00:27:42，liunaihua...@163.com 写道： >hi > 咨询下，想用flink1.10整合hbase，请问一下hadoop，hbase都需要什么版本？有哪位跑通过，可否提供下配置文件，我在本地测试，所有集群都正常启动，提交任务（flink读写hbase）总是报错，找不到某些类，是否跟版本有问题？ >提前谢谢了！ > >| | >liunaihua521 >| >| >邮箱：liunaihua...@163.com >| > >签名由 网易邮箱大师 定制

standalone模式下metaspace内存溢出

flink版本：1.10 standalone模式下，同一个batch作业提交多次后，taskmananger会metaspace内存溢出。 我用visualVm观察到，多次提交后，taskmanager的metaspace如下图： 虽然可以通过启动时调大jvm参数，但是这个应该是有问题的

Re: flink整合hbase

Hi, 可以贴下你的配置和异常栈吗？1.10的时候我本地也测试过，是ok的。 Best, Leonard Xu > 在 2020年6月5日，00:27，liunaihua...@163.com 写道： > > hi > 咨询下，想用flink1.10整合hbase，请问一下hadoop，hbase都需要什么版本？有哪位跑通过，可否提供下配置文件，我在本地测试，所有集群都正常启动，提交任务（flink读写hbase）总是报错，找不到某些类，是否跟版本有问题？ > 提前谢谢了！ > > | | > liunaihua521 > | > | >

Re: flink sql upsert模式写入mysql，es等key一定是groupby之后所有字段吗

感谢各位， 我先试试用Last_value这样的aggregate function绕过去。 @kcz 可能我表达不很清楚，具体描述一下遇到的具体场景就是：收到原始的数据，去mysql或者es里做维表关联，然后再以upsert的模式将结果写回mysql或es。 举个例子来说，我想按id为key更新整行数据（比如还有个字段amount是个随机的double类型值） select id， amount， …… groupby id； 这样子不行，必须 select id， amount， … groupby id，amount，……；

回复: flink整合hbase

Sink 是自己的还是官方的。 xueaohui_...@163.com 发件人： liunaihua521 发送时间： 2020-06-05 00:27 收件人： user-zh@flink.apache.org 主题： flink整合hbase hi 咨询下，想用flink1.10整合hbase，请问一下hadoop，hbase都需要什么版本？有哪位跑通过，可否提供下配置文件，我在本地测试，所有集群都正常启动，提交任务（flink读写hbase）总是报错，找不到某些类，是否跟版本有问题？ 提前谢谢了！ | | liunaihua521 | |

回复：flink sql upsert模式写入mysql，es等key一定是groupby之后所有字段吗

我大概get到你要说的需求，select那些其实是明细数据？但是没有跟聚合的数据拆开，所以才出现这种情况吧？ -- 原始邮件 -- 发件人: Leonard Xu https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/streaming/query_configuration.html#idle-state-retention-time

flink1.9 stream job的异常日志

附件是错误日志 我感觉看到错误日志之后，没有什么调查方向，应该怎么调查呢。2020-06-04 13:19:15,590 INFO org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction - FlinkKafkaProducer 7/12 - checkpoint 3425 complete, committing transaction TransactionHolder{handle=KafkaTransactionState [transactionalId=null,

flink整合hbase

hi 咨询下，想用flink1.10整合hbase，请问一下hadoop，hbase都需要什么版本？有哪位跑通过，可否提供下配置文件，我在本地测试，所有集群都正常启动，提交任务（flink读写hbase）总是报错，找不到某些类，是否跟版本有问题？ 提前谢谢了！ | | liunaihua521 | | 邮箱：liunaihua...@163.com | 签名由 网易邮箱大师 定制

Re: Flink SQL 子查询优化问题

hi Yichao, 最好的解决方式是在code gen的时候支持表达式复用，后续会慢慢加上这些优化。 Best, Godfrey 1048262223 <1048262...@qq.com> 于2020年6月4日周四 下午9:52写道： > Hi > > > 感谢答复，我明天使用下这种方式 > 社区也有同学提供了udf使用udtf代替的方法 > 我会尝试使用上述两种方式实现下 > > > 还有想问下之后有对这部分做优化的计划吗？ > > > Best, > Yichao Yang > > > > > > -- 原始邮件

回复：Flink SQL 子查询优化问题

Hi 感谢答复，我明天使用下这种方式 社区也有同学提供了udf使用udtf代替的方法 我会尝试使用上述两种方式实现下 还有想问下之后有对这部分做优化的计划吗？ Best, Yichao Yang -- 原始邮件 -- 发件人: godfrey he

Re: flink sql upsert模式写入mysql，es等key一定是groupby之后所有字段吗

hi yihan, 如 Leonard 所说，你可以考虑使用 first_value, last_value 等聚合函数和赛选其他字段。 1.11开始支持ddl定义pk信息， 如果id在source表中也是pk字段，可以直接定义， planner会利用该信息传递pk到sink表。 Bests， Godfrey Leonard Xu 于2020年6月4日周四 下午9:01写道： > Hi, > > > 但这样子不能通过calcite的sqlvalidation，select后面不能有非聚合项， > > select后费聚合值可以通过max()或sum()来取，因为已经按照key

Re: 关于flinksql 与维表mysql的关联问题

hi 可以考虑使用 temporal table join [1] Best, Godfrey [1] https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/streaming/joins.html#join-with-a-temporal-table 小屁孩 <932460...@qq.com> 于2020年6月4日周四 下午5:51写道： > 您的意思是open 全量预加载吗？我目前的逻辑是自己自定义的source 广播出去 > 这是我的source > > > import

Re: Flink SQL 子查询优化问题

hi Yichao, 目前 planner 会 try best 的将两个相邻的 project 节点合并在一起，除非两个project被分开。 就像你上面的那种做法。但是加一个group by的执行代价比较高。 对于blink planner 而言，*有一个绕的办法*，可以在子查询的结果加一个print sink（可以ignore输出）， 利用多sink的优化特性，将两个project分开，从而阻止优化器将两个project合并。 Best, Godfrey 1048262223 <1048262...@qq.com> 于2020年6月4日周四 下午4:56写道： > Hi

Re: flink sql upsert模式写入mysql，es等key一定是groupby之后所有字段吗

Hi, > 但这样子不能通过calcite的sqlvalidation，select后面不能有非聚合项， select后费聚合值可以通过max()或sum()来取，因为已经按照key group by了，所以取出来的非聚合值只能有一条， > 这样子key的state无限增长(比如说amount是一个随机的double数)，job跑不久就会fail掉。 State 可以配置ttl的，过期清理参考[1] 另外，即将发布的1.11中，支持在jdbc table 上定义primary key， 不用强制要求写upsert 的query，文档正在撰写中[2] Best, Leonard

Re: Re： flink数据sink到mysql 是事务处理

您好： 我是在样的情况，flink处理完数据后数据被我组织成一个3元组，第一个元素代表的是对这个元素的操作(插入还是删除)，第二个是我的user_ID,第三个是一个list里面有多个实体； 数据要插入或者删除，因为list的大小不确定所以在操作mysql的时候开起了事务保证这一批次的数据幂等，但是程序运行一段时间后出现死锁，但是不影响数据，所以发邮件想知道大家 是怎么处理的，结果这么多天才有回复讨论； 伪代码如下： public RdsOperaterSink2(String url, String name, String password)

Re: pyflink下mysql数据更新操作

Hi, 小学生 我的理解是key如果是mysql主键的话，你正常sink进去，那条key=1,value=3的数据就会覆盖掉key=1,value=2的数据 Best, Xingbo 小学生 <201782...@qq.com> 于2020年6月4日周四 下午6:05写道： > 您好，第一个问题您说的这样我理解，但是我的问题是比如先前在mysql表已经有一条key=1,value=2的数据，但是接着我要按key=1时，value=3去覆盖掉key=1,value=2的数据，也就是mysql > 命令中update table **SET value=3 WHERE key=1;

??????yarn-session??job????????????????????????

-n 1.10 ---- ??:""

??????flink sql upsert????????mysql??es??key??????groupby??????????????

Hi ??key https://blog.csdn.net/lp284558195/article/details/104609739 https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/streaming/query_configuration.html#:~:text=The%20minimum%20idle%20state%20retention,kept%20before%20it%20is%20removed. Best, Yichao

yarn-session多job的模式下是动态申请资源吗

你好。我目前用的是flink-1.7.2版本，使用的是yarn-session多job的模式。启动yarn-session之后，在flink ui上并没有看到申请配置的资源，而是提交flink job的时候才申请的container。当提交多个job时，yarn-session会按照job需要的slot数量动态申请资源。 比如，启动yarn-session的配置为: ./bin/yarn-session.sh -tm 8192 -s 2 �Cn 4 提交一个job需要的资源是 8个slot，那么yarn-session会申请4个taskmanager来运行该job。

flink sql upsert模式写入mysql，es等key一定是groupby之后所有字段吗

举个例子比如我想要 insert into mysql_sink select ID， amount， ……… from source groupby ID； 这里就是想按照id为key，在数据库里更新这个id对应的amount等其他值。 但这样子不能通过calcite的sqlvalidation，select后面不能有非聚合项，必须在groupby后面加上所有select后面的项。 但这样带来一个问题，这样子key的state无限增长(比如说amount是一个随机的double数)，job跑不久就会fail掉。

flink sql ddl能设置ttl吗？

各位好， 我有一个需求是要维表关联mysql数据，再upsert写入mysql，就是个丰富字段的简单需求。 现在有两个问题： 1.为了实现upsert插入，我试下来必须使用groupby指定key，但是select内又有很多非聚合项，所以没办法在groupby后面加上所有那些非聚合项，否则通不过calcite validation。 2.现在遇到了一个问题，跑一段时间会抱full GC，我估计是因为groupby导致的状态不清理。 请问flink sql ddl能通过类似SET 语句来设定ttl吗，搜了官方文档只有提到table

Re: pyflink??mysql????????????

??mysqlkey=1,value=2??key=1value=3key=1,value=2??mysql ??update table **SET value=3 WHERE key=1;

Re: pyflink下mysql数据更新操作

Hi, 小学生 1. 关于问题1，这个是支持的，你通过TableEnvrironment的sql_update就行 例如： s_env = StreamExecutionEnvironment.get_execution_environment() s_env.set_parallelism(1) st_env = StreamTableEnvironment.create(s_env) mysql_source_ddl = """ CREATE TABLE buybacksource ( a VARCHAR, b VARCHAR, c VARCHAR, proctime AS

??????????flinksql ??????mysql??????????

??open source source import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.functions.source.RichSourceFunction; import java.sql.Connection; import java.sql.PreparedStatement; import

Re:?????? flink 1.9 ????????????????

?? json??kafka??flink ??json??table?? json ?? 2020-06-03 16:31:15??"star" <3149768...@qq.com> ?? > > >??select

Re:????flinksql ??????mysql??????????

open ?? 2020-06-04 14:15:05??"??" <932460...@qq.com> ?? >dear?? ??,flinksql??mysql >mysql??

Batch 模式 Table API 增加cache算子

hi all, 最初blink分支上有对batch模式下的table cache操作, 后续会merge到flink上来么？ -- Best, Jun Su

Re: flink1.10整合hbase测试遇到的问题

可以尝试把依赖包 shaded 到你的 jar 包里，保证依赖的完整 Best Weihua Hu > 2020年6月3日 22:52，liunaihua521 写道： > > > > - 转发邮件信息 - > > 发件人： liunaihua521 > 发送日期： 2020年6月3日 22:18 > 发送至： user-zh-i...@flink.apache.org   > 、

Flink SQL ??????????????

Hi all ?? Flink ??1.10 Planner??old planner/ blink planner ??Flink SQL source??Tuple2.of(1, "{\"name\": \"a\"}"); query??select a.id, a.name, a.name from (select id, body_json_to_map(name) as name from data) a sink??print udf??body_json_to_map

pyflink??mysql????????????

1.pyflink sql??update1.10.0 2.pyflink??Table

Re: pyflink ????????????????

get??udf??

Re: pyflink 嵌套使用函数出错

Hi, 小学生 我稍微修改了一下你的code(你的from_elements那样写按理说就没法运行) code是能够正确运行的，你可以参考一下，你去掉的是不是有问题，或者你把你修改后的代码贴上来，再一起看看 from pyflink.table import StreamTableEnvironment, DataTypes, EnvironmentSettings, TableConfig, \ BatchTableEnvironment from pyflink.table.descriptors import Schema, OldCsv, FileSystem from

Re: pyflink ????????????????

??

Re: pyflink 嵌套使用函数出错

Hi, 小学生。 把函数get的标签udf给去掉，它只是普通的Python函数，不要加上@udf，加上之后就不是python的函数了。只有Python的UDF你才要加上@udf Best, Xingbo 小学生 <201782...@qq.com> 于2020年6月4日周四 下午2:46写道： > 各位大佬好，初学pyflink,有一个问题需要帮忙解决下。 > > > 代码为： > from pyflink.table import StreamTableEnvironment, DataTypes, >

pyflink ????????????????

pyflink,?? from pyflink.table import StreamTableEnvironment, DataTypes, EnvironmentSettings,TableConfig,BatchTableEnvironment from pyflink.table.descriptors import Schema, OldCsv, FileSystem from pyflink.table.udf import udf from

??????????flinksql ??????mysql??????????

Hi jobjobsavepoint?? Best, Yichao Yang ---- ??:"??"<932460...@qq.com; :2020??6??4??(??) 2:15 ??:"user-zh"

??????????flinksql ??????mysql??????????

Hi?? JOIN??WaterMark??maxoutoforderness ---- ??:"??"<932460...@qq.com; :2020??6??4??(??) 2:15 ??:"user-zh"

????flinksql ??????mysql??????????

dear?? ??,flinksql??mysql mysql?? mysql??mysqlmysql