维保 Join 理论上不会改变流的模式，我理解原来你的流是什么，就是什么。
Best,
LakeShen
WeiXubin <18925434...@163.com> 于2021年6月10日周四 下午5:46写道：
> 感谢你的回答，我这边看了官网目前 join 一共可以分为 Regular Joins 、 Interval Joins 以及 Temporal
> Joins
> 三大类。 我上面问题所述的确是采用了 Regular Joins 的方式。 之后我也尝试使用了 Lookup Join 但发现其最后也是转为
> INSERT INT
Hi Jason,
可以把 rocksdb statebackend 相关监控指标开启，然后结合指标看看。
具体参考：
https://ci.apache.org/projects/flink/flink-docs-release-1.10/ops/config.html#rocksdb-native-metrics
.
Best,
LakeShen
Jason Lee 于2021年6月11日周五 上午11:19写道：
>
>
> 各位社区的伙伴大家好
>
>
> 目前我们在使用Flink SQL 开发任务过程中遇到一个问题，有比较
看下最近是不是流量变大了，以及看下 Sink 的外部存储的集群压力是不是很大。
Best,
LakeShen
田磊 于2021年6月10日周四 上午11:36写道：
> 好的，谢谢，我看看。
>
>
> | |
> totorobabyfans
> |
> |
> 邮箱：totorobabyf...@163.com
> |
>
> 签名由 网易邮箱大师 定制
>
> 在2021年06月10日 10:50，Lin Li 写道：
> 你好，如果之前一直运行正常，建议检查下 sink 节点慢的原因（io
你可以先结合你的任务逻辑，以及 Flink Web UI 反压监控，看看到底是什么地方引起反压。
一般 source 反压，是下游的算子一直反压到 source。可以看看那个算子引起
Best,
LakeShen
yidan zhao 于2021年6月8日周二 上午10:28写道：
> 该任务有时候正常，偶尔反压。
> 最近观察发现，反压时，kafkaSouce节点100%反压到停滞，后续算子什么也收不到，任务整体停滞。
>
> 这类错误遇到过很多次了，目前我生产中flink有个很大问题就是这些稳定性，压力大不是需要时间去追赶，而是压力一大就整体处于停滞状态。
>
一种方法就是借助 Flink SQL Parser，解析你的 SQL，然后获取到不同的 SQL node，
然后每个 SQL Node 都有对应的类型，以及 connector 后面的 with 参数，你需要自己在
写代码判定一下即可。本质是通过解析 SQL，来获取血缘关系。
Best，
LakeShen
casel.chen 于2021年6月8日周二 上午12:05写道：
> 如何获取flink sql的血缘关系？如：表A -> 表B。有代码示例吗？谢谢！
Hi Jacob,
Maybe you miss the kafka connector dependency in your pom,
you could refer to this url :
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/kafka.html
Best,
LakeShen
Jacob <17691150...@163.com> 于2021年6月1日周二 上午9:54写道：
> Dear All,
>
> 我在使用F
或许你可以参考这个：
[image: image.png]
https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/dev/table/config/
Best，
LakeShen
chenchencc <1353637...@qq.com> 于2021年5月28日周五 下午4:30写道：
> 想问下state ttl能针对单表设置吗？
>
>
>
> --
> Sent from: http://apache-flink.147419.n8.nabble.com/
>
在增量 checkpoint 下，你可以简单理解状态几乎都存在 checkpoint 目录中的 shared 目录，
所以即使清理 checkpoint，也只是先将这次 checkpoint 引用的相关文件句柄的引用数减1，
只有一个文件没有 checkpoint 引用它时，才会真正删除该文件。
Best，
LakeShen.
刘建刚 于2021年5月28日周五 下午7:03写道：
> 增量快照的原理是sst文件共享，系统会自动帮助你管理sst文件的引用，类似java的引用，并不会因为一个快照删除了就会把实际的数据删除掉。
> 也就不会发生你说的情况
>
&
Hi，或许 Flink SQL interval join 能够满足你的需求。
Best，
LakeShen.
Shuo Cheng 于2021年5月31日周一 下午12:10写道：
> state ttl 只能是全局算子维度, table.exec.state.ttl
>
>
>
> --
> Sent from: http://apache-flink.147419.n8.nabble.com/
>
看下你的 Flink 版本是多少，如果是高版本的话，社区有提供 DataStream 的 HBase Sink。
Best，
LakeShen.
Zorro 于2021年5月31日周一 下午2:41写道：
> 由于你的DDL是变化的，无法提前预知所有字段，所以首先可以确定的是这个场景无法使用Flink SQL解决。
>
> 如果使用DataStream解决的话是可行的，唯一可能存在的问题就是目前社区没有提供DataStream的HBase sink。
> 如果你需要在DataStream中使用HBase sink的话，可能需要你自定义一个HBase sink
Hi，
集群重启，具体是指什么重启呢，这个能在描述详细一点吗？
Best,
LakeShen
datayangl 于2021年5月26日周三 上午9:43写道：
> FixedDelaStrategy 默认是从最近一个ck
> 恢复，其他的策略可以看官网。如果你是想问怎么实现的，不建议在邮件列表里问实现原理的问题。可以google找相关文章、相关flip 或者
> 直接debug源码。
>
>
>
> --
> Sent from: http://apache-flink.147419.n8.nabble.com/
看下你的 flink 命令对不对，然后去 Flink Web UI Checkpoint 界面，看下是否从 Savepoint 恢复（下面有个
restore path）.
之后再看下你的窗口时间类型用的是什么。
Best,
LakeShen
王春浩 于2021年5月27日周四 上午9:26写道：
> hi, 社区
>
> 版本flink 1.7
>
>
> 我正在尝试从保存点（或检查点）还原flink作业，该作业的工作是从kafka读取->执行30分钟的窗口聚合（只是AggregationFunction，就像一个计数器）
实现肯定可以实现，不过这个周期性加载 Mysql 并更新的 Cache 的功能，可能需要你自己定制化开发下。
Best,
LakeShen
WeiXubin <18925434...@163.com> 于2021年5月22日周六 下午4:09写道：
> 我想实现将MySQL中的 A 表数据预先查询出来进行缓存，用于给流表 B 进行 join关联。接下来定时查询并更新 A 表内的缓存数据，请问目前
> FlinkSQL 可以实现吗？
>
>
>
> --
> Sent from: http://apache-flink.147419.n8.nabble.com/
Hi fanrui,
thank you so much!
Best,
LakeShen
范瑞 <836961...@qq.com> 于2021年4月1日周四 下午7:36写道：
> Hi Lake:
>
>
> 目前的 Flink 版本应该都是不支持的，也就是说换了 StateBackend 不能正常恢复。Flink 1.13
> 做了这个事情，具体参考：FLIP41 和 FLINK-20976
>
>
>
> https://cwiki.apache.org/confluence/display/FLI
确定了 不能
LakeShen 于2021年4月1日周四 下午7:15写道：
> Hi 社区，
>如果实时任务状态后端之前是 FS StateBackend ，然后任务停止后，换成 RocksDB StateBackend
> 做恢复，作业状态能恢复吗？
>
> Best,
> LakeShen
>
Hi 社区，
如果实时任务状态后端之前是 FS StateBackend ，然后任务停止后，换成 RocksDB StateBackend
做恢复，作业状态能恢复吗？
Best,
LakeShen
Hi ，
延迟维表关联这个特性我觉得还是一个比较通用的特性，目前我们考虑借助 Timer 来实现的，社区如果有这个功能的话，我觉得对于 Flink
使用方会有很大帮助的。
我看社区有这样的一个 JIRA 再跟踪了[1]，我会持续关注。
[1] https://issues.apache.org/jira/browse/FLINK-19063
Best,
LakeShen
小屁孩 <932460...@qq.com> 于2020年6月8日周一 上午9:28写道：
> hi,目前我就是这样做的 数据在启动时会有
JIRA 再跟踪了[1]，我会持续关注。
[1] https://issues.apache.org/jira/browse/FLINK-19063
Best,
LakeShen
郑斌斌 于2020年8月27日周四 上午9:23写道：
> 小伙伴们：
>
> 大家好，请教一个问题，流表和维表在JOIN时，如果流表的数据没在维表中时，能否进行延迟join，比如，每10分钟进行match一下，连续match6次都没有match上的话，丢弃该数据。
> 这个场景怎么通过flink SQL或UDF实现，目前是通过timer来实现的，感觉有些麻烦。
>
> Thanks
Hi,
Benchao，这种发送到另外一个 topic
，再来关联，虽然感觉可以减缓这种没有关联到维表数据对于下游业务的影响，不过还是很难控制到具体的延缓时间，我们现在也考虑怎么优雅的实现。
Benchao Li 于2020年8月27日周四 上午10:08写道：
> Hi，
>
> 我们也遇到过类似场景，我们的做法是修改了一下维表Join算子，让它来支持延迟join。
>
> 其实还有个思路，你可以把这种没有join到的数据发送到另外一个topic，然后再消费回来继续join。
>
> 郑斌斌 于2020年8月27日周四 上午9:23写道：
>
> > 小伙伴们：
> >
; Error file: prelaunch.err.
> Last 4096 bytes of prelaunch.err :
>
>
> Yarn那边的日志显示：Could not find or load main class
> org.apache.flink.yarn.entrypoint.YarnJobClusterEntrypoint
>
>
> 不过我是Flink 1.12 的API，然后提交的集群还是Flink1.10.1的，不知道哪里的问题
>
>
> | |
> 凌战
> |
> |
&g
Yarn 后，状态之后从 Accepted 到 FAILED 状态。
希望有人帮我解惑，感谢
Best,
LakeShen
Hi flink2021，
你看下的聚合逻辑是不是在一个可撤回流上面进行聚合的呢，如果是的话，可以添加一下 mini batch 聚合优化参数，具体可以参考[1].
[1]
https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/tuning/streaming_aggregation_optimization.html
Best,
LakeShen
Robin Zhang 于2021年2月18日周四 下午2:44写道：
> Hi,flink2021
>首先看看业
Hello 社区，
最近开始考虑整理 Flink 1.10 升级到 1.12 的整体收益，想问下， Flink 1.12 SQL 语法是否完全兼容 1.10
版本的 SQL 语法，我的理解应该是兼容的.
Best,
LakeShen
如果是窗口类聚合，可以尝试一下自定义窗口 Trigger
Best,
LakeShen
林影 于2021年1月28日周四 下午5:46写道：
> Hi, Jessica.J.Wang
> 开源flink看起来没这个功能哈，文档翻了一遍没找到
>
> Jessica.J.Wang 于2021年1月28日周四 下午5:25写道：
>
> > 你使用的是什么窗口呢，是 Tumble或者Hop吗，如果没数据 但是想提前输出结果可以用 emit
> >
> >
> https://help.aliyu
我这边是底层其实有个 Flink Jar 任务，然后将 Flink SQL 代码以及作业相关配置参数，当做参数一起传入到底层
的 Flink Jar 中去，当然，现在也有很多其他的方式能够实现，也可以参考楼上的链接。
Best,
LakeShen
Peihui He 于2021年1月8日周五 上午9:21写道：
> 可以尝试下zeppelin 0.9
> http://zeppelin.apache.org/
>
>
> jiangjiguang719 于2021年1月7日周四 下午8:34写道：
>
> > 目前我司的SQL作业
看下 Flink 任务运行,是否是其他机器上的资源先达到瓶颈，而不是 CPU，比如 IO，同时看下你的 flatmap 处理单条记录的时间。
同时也参考上面同学的，是否存在反压，如果 flatmap 逻辑比较复杂，也有这个可能。
Best,
LakeShen
赵一旦 于2021年1月5日周二 下午9:13写道：
>
> 可以看看是否反压。反压说明并行度还是不够，不反压的话看处理速度是否符合预期。符合预期就不用调了，说明你的任务不复杂，那点cpu占用就够了。如果不符合预期，也没有任何反压，那就是source消费速度太慢。
>
>
>
> house
Hi 夜思流年梦,
看下你的 dwd_XXX 这张表的类型，是 append 数据流，还是 retract 数据流。
如果是 retract ，应该就不能再上面进行窗口计算了。
Best,
LakeShen
史 正超 于2020年11月3日周二 下午6:34写道：
> canal-json 的format也是会有delete 和update的数据的，同样changelog-json也是。他们的都支持 INSERT
> UPDATE DELETE, 相关代码如下:
>
> @Override
> publi
kill 掉后，从 Savepoint 恢复，可以定时做一次 savepoint
，不过这种方法由于状态和当前时间有间隙，可能会导致数据重放，从而下游可能会有消息重复。
建议选择 1 来实现。
Best,
LakeShen
hailongwang <18868816...@163.com> 于2020年11月3日周二 下午7:21写道：
> Hi bradyMk，
>
>
> 在 on yarn 的模式下，如果某个container 被kill 了，是会重新拉起的。
> 至于整个 job 被kill 了，这种情况应该是自己手动显示的去停
Hi 社区，
我的一个 Flink SQL 任务，版本为 1.10，代码中使用了中文值，具体如下：
select xxx, case when a = 'a' then '你好' when a = 'b' then '你好呀' end as va
from xxx ;
然后会把这个结果输出，最后发现 va 的值乱码了，也就是中文乱码。
目前有什么比较好的解决方法吗。
Best,
LakeShen
Hi zhanglachun,
你们使用 checkpoint 的方式是什么？是否有 full gc 的情况呢
Best,
LakeShen
徐骁 于2020年8月26日周三 上午2:10写道：
> input
> .keyBy()
> .timeWindow()
> .reduce(new MyReduceFunction(), new MyProcessWindowFunction());
>
> 可以看一下官网的 reduce 用法, 可以在数据进入窗口的时候进行处理(排序), 而没必要等到 window 结束后再进行 这样可
嗯嗯，Congxian,感谢你的回复，我通过 Maven Shaded 解决问题。
Congxian Qiu 于2020年7月16日周四 下午8:19写道：
> Hi
>
> 你的图挂了，如果单纯想解决 jar 包冲突的问题，那么 maven shade plugin[1] 或许对你有用
>
> [1]
>
> https://maven.apache.org/plugins/maven-shade-plugin/examples/class-relocation.html
> Best,
> Congxian
>
&
-table_2.11-1.6-RELEASE.jar 中也有这个依赖，这两个都在 lib 目录下，然后就出现了类冲突问题。
请问怎么解决这个问题呢，非常期待你的回复。
Best,
LakeShen
你的 Yarn 环境，Flink 任务使用的 Kafka 地址，应该是 Yarn 环境的 kafka broker 地址。
LakeShen 于2020年7月10日周五 上午10:08写道：
> Hi,
>
> 从日志看出，应该是你提交到 Yarn 的环境，这个环境和你的测试环境的 kafka 连接不上，获取不到元数据。
>
> 这里你检查一下你的 Yarn 环境，Flink kafka broker 地址是否是测试环境的 kafka broker 地址。
>
> Best,
> LakeShen
>
> m...@sino
Hi,
从日志看出，应该是你提交到 Yarn 的环境，这个环境和你的测试环境的 kafka 连接不上，获取不到元数据。
这里你检查一下你的 Yarn 环境，Flink kafka broker 地址是否是测试环境的 kafka broker 地址。
Best,
LakeShen
m...@sinoiov.com 于2020年7月9日周四 下午9:21写道：
> hi：zhisheng：
>
> 这是TM日志，在这之前没有任何错误日志，
>
> 代码逻辑很简单：
> SingleOutputStreamOperat
Hi Peihui,
如果消费的 Kafka json 中，json 比较复杂的话，比如存在嵌套，就像下面的格式：
{
"a":"b",
"c":{
"d":"e",
"g":"f"
}
}，
那么在 kafka table source 可以使用 row 来定义:
create table xxx (
a varchar,
c row
)
如果 还存在嵌套，可以继续再使用
, 然后设置 '
connector.sink-partitioner-class'. Best,
LakeShen
shizk233 于2020年7月2日周四 下午7:46写道：
> Hi air23,
>
> sql似乎不支持相关的设置，可以通过env或配置文件设置所有蒜子的并行度。
> 你可以试试流转表，可以做到细粒度的控制。
>
> Best,
> shizk233
>
> air23 于2020年7月2日周四 下午6:40写道：
>
> > hi
> > 就是我用
> >flink sql
我在较低版本，Flink on k8s ，也遇到 OOM 被 kill 了。
我感觉可能是 TaskManager 堆外内存不足了，我目前是 Flink 1.6 版本，Flink on k8s , standalone per
job 模式，堆外内存默认没有限制~。
我的解决方法增加了一个参数：taskmanager.memory.off-heap: true.
目前来看，OOM被 kill 掉的问题没有在出现了。希望能帮到你。
Best,
LakeShen
SmileSmile 于2020年6月30日周二 下午11:19写道：
>
> 补充一下，内核版本为 3.
Hi shizk233,
可以看下这个 Jira ， https://issues.apache.org/jira/browse/FLINK-16681。
这个就是长时间没有数据，导致 connection 断开问题。
Best,
LakeShen
shizk233 于2020年6月30日周二 下午1:34写道：
> Hi Zhong Tang,
>
>我查看了该jira有关的重连pr，https://github.com/apache/flink/pull/8429
> ，但该pr主要通过重连机制来规避网络问题导致的连接失效，
> 但由
JobManager 是一个 K8s Job,所以配置 K8s Job
的 restartPolicy 为 OnFailure。
这只是我的一些经验，仅供参考。
Best,
LakeShen
zhisheng 于2020年6月30日周二 上午8:51写道：
> hi，Tony
>
> 你可以把 Checkpoint 间隔时间稍微设置大一些，看起来像是作业启动的时候 Task 还没 Running，就开始执行 Checkpoint
> 了，而 Checkpoint 是要求所有的 Task 是处于 Running 状态的，所以就会丢弃掉那次
> Checkpoint，BT，就
或者补齐一个非 Null ，但是又不影响业务逻辑的数值
Jingsong Li 于2020年6月30日周二 上午9:58写道：
> Hi,
>
> 我记得NULL的literal是可以的，不过需要cast成确定的类型，比如 select CAST(null AS VARCHAR);
> 你试试。
>
> Best,
> Jingsong
>
> On Tue, Jun 30, 2020 at 9:40 AM seeksst wrote:
>
> > Hi,
> >
> >
> > 按照你的意思是想将两个不同的数据集进行union，但是由于字段不同需要补充NULL。
> >
哈哈，学习了一波
Jingsong Li 于2020年6月30日周二 上午9:59写道：
> > 用Row 和 Tuple 性能上会有差别吗？
>
> 理论上有细微的差别，
> 但是，基本上性能瓶颈不会在这里。。所以你应该感受不到
>
> Best,
> Jingsong
>
> On Tue, Jun 30, 2020 at 8:51 AM zhisheng wrote:
>
> > 可以测试一下
> >
> > Tianwang Li 于2020年6月29日周一 下午8:13写道：
> >
> > > >
> > > > 用 Row 啊，支持 Int.MAX 个元素，还能支持
Hi 李军，
目前我们在 Yarn 上面的话，用的是 Flink On Yarn Per Job 模式，在 K8s 上面的话，就是 Standalone per
Job 模式。
Best,
LakeShen
刘佳炜 于2020年6月28日周日 下午5:14写道：
> 如果你公司用hadoop的话就是YARN StandAlone一般都是单机测试练习的
>
>
>
> 发自我的iPhone
>
>
> -- 原始邮件 --
> 发件人: 李军 发送时间: 2020
Hi Tianwang Li,
偶尔一两天出现 Checkpoint 超时，看下你的任务中，是否可能存在某类 key 在这一两天会突然增多的情况。
Best,
LakeShen
zhisheng 于2020年6月28日周日 上午10:27写道：
> hi, Tianwang Li
>
> 看到有三个图片挂了，可以试着把图片上传到第三方的图床，然后贴个链接过来，另外：
>
> > 任务经常会出现反压（特别是在窗口输出的时候）
>
> 这个检查一下窗口下游算子的情况，比如是不是窗口输出的数据过多，而 sink 的并发还和之
Hi 张立志,
一般 Checkpoint 超时，可以先看看你的任务中，是否存在反压，比如 Sink 阶段，又或者是某个地方有 flatMap操作导致。
然后看下自己任务中，是否存在热点问题等。如果一切都是正常的话，可以尝试使用 RocksDB 的增量 Checkpoint ，具体参考[1]。
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.10/ops/state/state_backends.html#rocksdb-state-backend-details
Best,
LakeShen
张立志
Hi guanyq,
你为什么希望 app id 不变呢？
Best,
LakeShen
guanyq 于2020年6月28日周日 上午9:10写道：
> 问题1
>
> ./bin/flink run -m
> yarn-cluster每次提交是都会产生一个新的APP-ID如:application_1567067657620_0254
>
> 当yarn application -kill application_1567067657620_0254后，
>
> 在提交./bin/flink run -m yarn-clust
Hi guanyq,
从日志中，我看到 TaskManager 所在机器的本地存储几乎快用完了。
看下是否因为 TaskManager 所在机器的存储不够导致
Best,
LakeShen
xueaohui_...@163.com 于2020年6月20日周六 上午9:57写道：
> 不知道有没有yarn上面的详细日志。
>
> hdfs是否有权限问题
>
>
>
> xueaohui_...@163.com
>
> 发件人： guanyq
> 发送时间： 2020-06-20 08:48
> 收件人： user-
Hi xuhaiLong,
看你的依赖，应该用的 old planner,你要使用 blink planner 才能使用row_number 函数。要使用
flink-table-planner-blink_2.11
具体文档参考：
https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/#table-program-dependencies
Best,
LakeShen
xuhaiLong 于2020年6月23日周二 下午8:14写道：
> "org.apac
Hi ,
正如 Congxian 所说，当 Flink 任务容错恢复重启时，会从上一次成功的 Checkpoint 进行恢复。
所以你所说的 last checkpoint 失败，具体是什么含义呢？
Best,
LakeShen
Congxian Qiu 于2020年6月22日周一 下午8:23写道：
> hi
>
> 这里说的 state 不一致是什么意思呢？checkpoint 恢复保证全局的 state 被重置到之前某个成功的 checkpoint。
>
> Best,
> Congxian
>
>
> 莫
Hi xuhaiLong，
看下是否能够在 Flink SQL 中，通过自定义 UDTF 来满足你的需求。
Best,
LakeShen
xuhaiLong 于2020年6月23日周二 下午7:18写道：
> Hi
>
> 请教一个问题
>
>
> 我需要对一个类似这样的数据进行计算获取用户 categoryId
> | userId | articleID | categoryId | score |
> | 01 | A | 1 | 10 |
> | 01 | B | 1 | 20 |
> | 01 | C
Hi ZheFu,
可以把你的 Flink 版本说一下，我大致理解是这样的，每次 sink 端 在 snapshotState 的时候，会检查该次 Sink
的数据是否都已经 Sink 到了 kafka.
也就是说，你这次 Checkpoint 的时候，由于你的 Checkpoint 间隔较短，Kafka 那边给回的消息记录 Ack
还没有弄完，所以有这个问题。建议 Checkpoint 间隔弄长点。
具体代码查看：FlinkKafkaProducerBase.snapshotState 这个方法。
Best,
LakeShen
Congxian Qiu 于2020年6月11日
Hi,
Flink 任务运行时，本身就是一个 DAG 图，从 Source 出发，到 Sink 结束。
所以你所说的顺序，完全是按照你的业务逻辑来的。
Best,
LakeShen
Caizhi Weng 于2020年6月8日周一 下午5:41写道：
> Hi，
>
> 不能设置这两个算子的执行顺序。流作业很可能是不会结束的，如果非要设定顺序的话可能另一个算子永远都不会执行。
>
> 不过如果已知是有限流的话，可以把这两个算子分成两个 job 提交，拿到 job client 以后用 job client 等待一个 job
> 的结束，再
Hi,
RetractStream 目前是无法输出到 kafka 的，因为 kafka 是 Append 模式。
不过你应该可以定义一个时间窗口 T ，滚动窗口的时间就是 T，然后聚合一次，输出到 kafka，后面都使用这个 kafka topic。
Best,
LakeShen
star <3149768...@qq.com> 于2020年6月3日周三 下午4:31写道：
> 感谢两位的回复，
> 转成回撤流的这个流其实是一张轻度汇总表，
> 例如，select year,month,day,province,sub_name,su
或者可以通过 Kafka-Manager 来查看
guanyq 于2020年6月3日周三 下午4:45写道：
>
>
>
> 找到了，原生就有的committedOffsets-currentOffsets
>
> https://ci.apache.org/projects/flink/flink-docs-release-1.9/monitoring/metrics.html#reporter
> Connectors
> Kafka Connectors
> | Scope | Metrics | User Variables | Description | Type |
>
Hi Benchao,
TableEnvironment 和 StreamTableEnvironment 具体有什么差异吗，我看StreamTableEnvironment
继承了 TableEnvironment。
这块我不是很了解，有什么文档介绍吗，感谢。
Best,
LakeShen
Benchao Li 于2020年5月28日周四 下午5:52写道：
> Hi zhisheng，
>
> 这块内容在FLIP-84[1]里面做了大量的重构和优化，而且大部分工作在1.11已经完成了，你们可以看下。
>
> [1]
> https:
Hi,
关于 Json 的解析，当你的 Json 里面的一个字段一个镶嵌类型的话，可以将其定义为一个 row，row 里面还可以定义 row 字段。
注意 row 里面的字段名称要和原始json 里面的字段一致。
Best,
LakeShen
claylin <1012539...@qq.com> 于2020年5月26日周二 上午10:17写道：
> 嗯 谢谢 我试下看下
>
>
>
>
> --原始邮件--
> 发件人:"Benchao Li&quo
Hi,
看下是否存在热点问题，我看你根据 server,reason 这两个字段来进行 group by
Best,
LakeShen
Benchao Li 于2020年5月26日周二 下午6:50写道：
> Hi，
>
> 看起来你的写法应该没有太大问题。可能有两个问题需要确认一下：
> 1. 你的watermark生成的正确吗？也就是说window的结果有正常输出么？如果watermark延迟很高，是会导致有多个window同时存在的
> 2. 你是怎么判断state上升呢？通过checkpoint看出来的？还是看到heap一直上升？
>
Hi,
你可以描述一下你的 Flink 版本，具体空闲状态保留时间的含义，请参考一下[1]:
[1]
https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/streaming/query_configuration.html#idle-state-retention-time
Best,
LakeShen
claylin <1012539...@qq.com> 于2020年5月17日周日 下午10:24写道：
> 过期时间是10-15分钟，按理说我是按照每分钟作为key分组的
Hi ,
你可以把你的场景在描述的详细一些。
Best,
LakeShen
请叫我雷锋 <854194...@qq.com> 于2020年5月14日周四 下午9:42写道：
> 各位大佬好，请问有啥好的save point容灾方案嘛？
>
>
>
> 发自我的iPhone
Hi 各位，
最近有业务需求，需要用到双流 Join (Interval Join)，如果两个流中，其中一个流式可撤回流（Retract），或者两个流都是
Retract 流，那么他们还能进行双流 Join 吗？
目前我知道如果两个流是 Append 流的话，肯定可以双流 Join，但是其中一个流式 Retract 流，就不知道还能不能 Join 了。
期望你的回复。
Best,
LakeShen
Yarn 日志的话，直接根据 任务的 Application ID ，去 Yarn 的 Web UI 上面看吧。
Best,
LakeShen
guaishushu1...@163.com 于2020年5月8日周五 下午3:43写道：
> 日志全部输出到.err日志里面了，flink-web也看不到日志有人知道吗？
>
> --
> guaishushu1...@163.com
>
Hi ，
你可以看下你的内存配置情况，看看是不是内存配置太小，导致 networkd bufffers 不够。
具体文档参考：
https://ci.apache.org/projects/flink/flink-docs-release-1.10/ops/memory/mem_detail.html
Best,
LakeShen
a511955993 于2020年5月7日周四 下午9:54写道：
> hi， all
>
>
> 集群信息：
> flink版本是1.10，部署在kubernetes上，kubernetes版本为1.17.4
. But I go
into the /tmp dir ,I
couldn't find the flink checkpoint state local directory.
What is the RocksDB local directory in flink checkpointing? I am looking
forward to your reply.
Best,
LakeShen
Hi 首维，
你用的 Flink 版本是多少呢，然后你的 Checkpoint interval 设置的时间是多少，这两个信息提供一下。
Best,
LakeShen
刘首维 于2020年4月28日周二 下午6:28写道：
> Hi all,
>
>
>
> 今天发现有一个作业日志中连续打印下面这个报警
>
> "Committing offsets to Kafka takes longer than the checkpoint interval.
> Skipping commit of previous o
Hi 阿华,
数据延迟有可能是逻辑中某个环节比较耗时，比如查询 mysql，或者某处逻辑较复杂等等。
可以看看自己代码中，有么有比较耗时的逻辑。同时可以将自己认为比较耗时的地方，加上日志，看下处理时间。
Best,
LakeShen
阿华田 于2020年4月29日周三 上午9:21写道：
> 好的 感谢大佬
>
>
>
> | |
> 王志华
> |
> |
> a15733178...@163.com
> |
> 签名由网易邮箱大师定制
>
>
> 在2020年04月29日 09:0
t;
> *- Dtaskmanager.heap.size=2000m*
>
>
> Thank you~
>
> Xintong Song
>
>
> [1]
>
> https://ci.apache.org/projects/flink/flink-docs-release-1.6/ops/deployment/kubernetes.html
>
>
>
> On Fri, Apr 24, 2020 at 11:10 AM LakeShen
> wrote:
>
> > Hi
Hi Xintong,
非常感谢你的回复。想再请教一个问题，什么地方会使用到 docker-compose.yaml 呢。
我目前使用一种绕开的方式解决这个问题，就是在 DockerFile 打镜像的时候，先把 conf 目录拷贝出来(这个里面的 TaskManger
内存动态传入的)
然后在 config.sh 中，强行设置了 FLINK-CONF-DIR . 但是我觉得你的方式更优雅一些。
所以想问一下 什么地方会使用到 docker-compose.yaml呢 。
Best,
LakeShen
Xintong Song 于2020年4月24日周五 上午10:49写道
录中的 flink-conf.yaml 里面的配置来启动。
我现在在打镜像的时候，也会把flink-dist 目录打进去，同样把 taskmanager.heap.mb动态传入到
flink-conf.yaml中，但是最终我在启动我的作业的时候，登录到 taskmanager 的一个 pod 上面查看，发现其
flink-conf.yaml 里面， taskmanager.heap.mb 始终是 1024.
是不是在什么地方，把 taskmanager.heap.mb 写死到了 flink-conf.yaml 中呢？
Best,
LakeShen
社区版的 Planner 针对 Key 状态的清理，使用的 Timer 来进行清理。
1.9.1 Blink planner 最底层状态清理 还是使用的 StateTTLConfig 来进行清理（不是
Background），所以存在部分状态后面没读，
状态没有清理的情况
Benchao Li 于2020年4月21日周二 下午11:15写道：
> 我对checkpoint不是很熟悉。 不过这个错误看起来是因为修改了TtlConfig导致serializer不兼容导致的，可能不是很好解决。
>
> 酷酷的浑蛋 于2020年4月21日周二 下午10:37写道：
>
> >
Hi tao wang,
你可以在你的 flink-conf.yaml 里面配置 Checkpoint 的目录,就像楼上 Yangze 所说
state.checkpoints.dir: hdfs://namenode:port/flink/checkpoints/
Best,
LakeShen
Yangze Guo 于2020年4月15日周三 下午2:44写道：
> checkpoint的目录设置key为state.checkpoints.dir
>
> 你可以这样设置
> state.checkpoints.dir: hdfs://namenod
Hi Junzhong ，
图片没有显示,能否把图片重新上传一下。
Best,
LakeShen
Junzhong Qin 于2020年4月11日周六 上午10:38写道：
> 在跑Flink任务时，遇到了operator反压问题，任务执行图如下，source(读Kafka),
> KeyBy(抽取数据字段供keyBy操作使用)，Parser(业务处理逻辑)，Sink(写Kafka)，除了KeyBy->Parser使用hash(KeyBy操作)链接，其他都使用RESCALE链接。(并发度仅供参考，这个是解决问题后的并发度，最初的并发度为
> 500
-position-configuration
Best,
LakeShen
gang.gou 于2020年4月7日周二 下午4:17写道：
> 好的，我试一下，有结果了同步大家，谢谢！
>
> 在 2020/4/7 下午3:52，“Evan” 163@flink.apache.org 代表 chengyanan1...@foxmail.com> 写入:
>
> 之前的代码好像乱码了，我设置了一下，重新发一下，建议你
> 在获取consumer之后，再设置一下consumer.setStartFromLatest();，这样设置的参考就是官
Hi Qijun,
看下 kafka 是不是所有分区都有数据呢，或者在这个时间截后：158567040L，后面是不是只有分区3写入数据，个人的想法。
Best,
LakeShen
Qijun Feng 于2020年4月2日周四 下午5:44写道：
> Dear All，
>
> 我的 Kafka cluster 有三个机器，topic 也分了三个 partition， 我用 Flink 读取 Kafka
> 消息的时候，感觉只从一个 partition 读取了东西， 一开始我的 bootstrap.servers 只写了一个地址，
>
Hi ，
这种情况可能是你改变的 Flink SQL 的拓扑结构，导致部分算子的 uid 发生变化，然后在从状态恢复的时候，没有找到算子的状态。
所以在开发 SQL 任务的时候，一般更改 SQL 代码时，不要改变其拓扑结构,SQL 任务上线后，就不要在轻意改了。
Best,
LakeShen
酷酷的浑蛋 于2020年4月2日周四 下午6:31写道：
> 关键我的程序是flink-sql，其它的算子基本都设置过uid了，flink-sql可以设置uid吗，或者说sql中的自动分配的uid怎么查找呢
>
>
> | |
> apache22
the containerized.heap-cutoff-ratio be 0.15.
Is there any problem for this config?
I am looking forward to your reply.
Best wishes,
LakeShen
嗯嗯，非常感谢你的回答，Congxian Qiu 。
Congxian Qiu 于2020年3月28日周六 上午11:39写道：
> Hi
>
> 这个地方我理解是，每次处理一定数量的 StateEntry 之后，会获取当前的 timestamp 然后在 RocksDB 的 compaction
> 时对所有的 StateEntry 进行 filter。
> > Calling of TTL filter during compaction slows it down.
>
> Best,
> Congxian
>
&
哈哈，学习了，Benchao，
Benchao Li 于2020年3月28日周六 下午11:26写道：
> Hi，
>
> 这个输出是retract的是by design的，你可以自己改造下sink，来输出你想要的结果。
> fast
> emit是按照处理时间来提前输出的。比如某个key下面来了第一条数据之后，开始设置一个固定周期的定时，如果下个周期聚合结果有发生变化，则输出。
>
> flink小猪 <18579099...@163.com> 于2020年3月28日周六 下午8:25写道：
>
> >
> >
> >
> >
统一对 Flink 项目源码进行编译打包，你会在 flink-dist 这个模块下面的 target 目录下面看到相关 Flink
命令行的一些东西，同时在lib 包下面，
会有一些 Flink Jar 包
Best wishes,
沈磊
godfrey he 于2020年3月26日周四 下午8:51写道：
> 目前 flink 支持 Scala 2.11 和 Scala 2.12， 默认情况下，通过 mvn package 打包出来的是包含 Scala
> 2.11 的包，例如 flink-table-blink_*2.11*-1.10.0.jar。
> 可以通过
Hi 社区的小伙伴，
我现在有个关于 Flink 1.10 RocksDBStateBackend 状态清理机制的问题，在 1.10中，RocksDB 默认使用 in
background 方式进行状态清理，使用 compaction filter 方式。正如官方文档所说：
> RocksDB compaction filter will query current timestamp, used to check
> expiration, from Flink every time after processing certain number of state
>
Hi farron ,
能否在详细描述一下你的 SQL 的逻辑
faaron zheng 于2020年3月23日周一 下午10:12写道：
>
> 大家好，我在用flink1.10执行sql时，当数据比较大的时候，3T左右，100多亿条数据，在执行hash和sort的时候经常超出内存限制，被yarn杀掉，我的tm给了40g内存，每个有10个slot，每个slot3g内存。我也试过给更大的内存，但是没什么效果。不知道这是什么原因？
>
>
>
>
QL环境中停止任务时，是如何与yarn进行交互反馈的，仍需努力学习，不过感激不尽，谢谢
>
> Best,
> amenhub
>
>
>
> amen...@163.com
>
> 发件人： LakeShen
> 发送时间： 2020-03-20 16:18
> 收件人： user-zh
> 主题： Re: Flink SQL中关于chk删除和保留的疑问
> Hi amenhub,
>
> 我这边针对你这三个问题回答一下：
>
> 1. 官网描述的[when program is cance
Hi amenhub,
我这边针对你这三个问题回答一下：
1. 官网描述的[when program is cancelled]，这个cancelled是指除故障导致停止之外的一切任务停止方式吗？
这里的 canceled 主要是针对你使用 cancel 或者 cancel with savepoint 来停止作业，你可以通过命令行或者
REST API 来调用。
2.如果是，在flink on yarn模式下，停止任务时采取 yarn application -kill [yarnAppId]
形式停止flink任务，chk似乎没有自动删除？
是的，如果你直接 yarn
Compaction Filter 算法来清理。
第二个就是使用增量 Checkpoint 方式吧。
Best wishes,
LakeShen
lucas.wu 于2020年3月20日周五 上午11:50写道：
> 可以考虑自己实现一个udf ，使用bitmap或者hyperloglog去实现。
>
>
> 原始邮件
> 发件人:zhishengzhisheng2...@gmail.com
> 收件人:user-zhuser...@flink.apache.org
> 发送时间:2020年3月20日(周五) 11:44
&g
to your reply.
Best regards,
LakeShen
大家好，本文为 Flink Weekly 的第九期，由沈磊(LakeShen)整理，主要内容包括：近期社区开发进展，邮件问题答疑以及 Flink
中文社区相关技术博客的分享。
社区开发进展
[Table API & SQL] Jingsong Li 发起 FLIP-115 的讨论，主要在 Flink Table 支持 FileSystem
Connector,FLIP-115 主要内容包括：
1. 在 Flink Table 中支持 FileSystem Table Factory，同时支持csv/parquet/orc/json/avro
格式。
2. 支持在流应用或者 F
ent
> timestamp each time 1000 entries have been processed.
What's the meaning of 1000 entries? 1000 different key ?
Thanks to your reply.
Best regards,
LakeShen
Ok, thanks! Arvid
Arvid Heise 于2020年3月10日周二 下午4:14写道：
> Hi LakeShen,
>
> you can change the port with
>
> conf.setInteger(RestOptions.PORT, 8082);
>
> or if want to be on the safe side specify a range
>
> conf.setString(RestOptions.BIND_PORT, "8081-8099");
Hi 志华，
你可以完全自己扩展 Flink SQL DDL 语法的功能，用来支持你们公司自己的实时数据源，或者 Sink
等等，具体实现，请参考楼上 jinhai
的链接
Best wishes,
沈磊
jinhai wang 于2020年3月13日周五 下午7:22写道：
> Page on “User-defined Sources & Sinks”. For flink 1.10:
> https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/sourceSinks.html
>
Thanks a lot!, tison
tison 于2020年3月12日周四 下午5:56写道：
> The StoppableFunction is gone.
>
> See also https://issues.apache.org/jira/browse/FLINK-11889
>
> Best,
> tison.
>
>
> LakeShen 于2020年3月12日周四 下午5:44写道：
>
>> Hi community,
>> now I am seei
reply.
Best wishes,
LakeShen
only
suit for the sources that implement the StoppableFunction interface, is it
correct?
Thanks to your reply.
Best wishes,
LakeShen
that?
Thanks to your reply.
Best wishes,
LakeShen
In my thought , I think I should config the correct flink jobserver for
flink task
LakeShen 于2020年3月4日周三 下午2:07写道：
> Hi community,
> now we plan to move all flink tasks to k8s cluster. For one flink
> task , we want to see this flink task web ui . First , we create the k8s
Hi community,
now we plan to move all flink tasks to k8s cluster. For one flink
task , we want to see this flink task web ui . First , we create the k8s
Service to expose 8081 port of jobmanager, then we use ingress controller
so that we can see it outside.But the flink web like this :
Hi community，
now I have a flink sql job, and I set the flink sql sate retention
time, there are three dir in flink checkpoint dir :
1. chk -xx dir
2. shared dir
3. taskowned dir
I find the shared dir store the last year checkpoint state,the only reason
I thought is that the latest
另外，我的容错恢复是 Flink 自身行为 Checkpoint 的容错恢复，我看 CheckpointCoordinator 有两个方法：
restoreLatestCheckpointedState 和 restoreSavepoint，所以这两个方法有什么区别呢？
LakeShen 于2020年1月19日周日 下午4:30写道：
> 非常感谢唐云的答复，目前在每个 Job-id 的 Checkpoint 子目录，有三个目录：
> 1. chk-id 的目录
> 2. shared 目录，其中状态非常大
> 3. taskowned
>
> 我
kpoint：在确保当前作业的checkpoint有最新完成的情况下，可以直接删除 job-id 目录下的 chk-
> 开头的目录。
>
> 祝好
> 唐云
>
>
> From: LakeShen
> Sent: Sunday, January 19, 2020 15:42
> To: user-zh@flink.apache.org
> Subject: Re: Flink 增量 Checkpoint ，容错恢复后，随着时间的推移，之前的 Checkpoint 状态
link/runtime/checkpoint/Checkpoints.java#L214
>
> 祝好
> 唐云
>
>
>
>
>
> From: LakeShen
> Sent: Friday, January 17, 2020 16:28
> To: user-zh@flink.apache.org
> Subject: Flink 增量 Checkpoint ，容错恢复后，随着时间的推移，之前的 Checkpoint 状态没有清理
>
>
是否可以使用 空闲状态 Retention Time 来设置
JingsongLee 于2019年9月4日周三 下午6:12写道：
> 一般是按时间(比如天)来group by，state配置了超时过期的时间。
> 基本的去重方式就是靠state(比如RocksDbState)。
> 有mini-batch来减少 对state的访问。
>
> 如果有倾斜，那是解倾斜问题的话题了。
>
> Best,
> Jingsong Lee
>
>
> --
>
大家好，我现在有个任务，状态比较大，使用的增量的 Checkpoint，Flink 1.6 版本，默认的 Checkpoint 保留数为1，
我看了一下源码，发现当完成的 Checkpoint 数大于 state.checkpoints.num-retained的数值时，会对之前的完成的
Checkpoint 状态做清理。
当时我早上八点40多任务 Checkpoint 成功，当时的 chk-id 为
94040，然后到下午15点之间，Checkpoint都是超时失败的，然后15点容错恢复，从
chk-94040 恢复（八点40的状态），最新的 chk-id 为
Hi community， now I am using Flink sql , and I set the retention time, As I
all know is that Flink will set the timer for per key to clear their state,
if Flink task always checkpoint failure, are the key state cleared by
timer?
Thanks to your replay.
共有 124 项搜索結果，以下是第 1 - 100 matches
Mail list logo
