回复：关于 richfunction中初始化数据库连接的问题
Hi, 是执行到哪步出现了问题？可以提供下面一些内容来帮忙定位问题吗？ 1.截图或者日志 2.不同的数据库都是哪些数据库，以及版本是哪些 3.单写一个测试用例加载两个数据库是否能够加载成功 4.代码伪编码 Best, Yichao Yang -- 原始邮件 -- 发件人: hdxg1101300...@163.com
回复： State里面用guava Cache
Hi, 每次cache的长度都是一有没有可能并发比较大，每一个1都是不同的算子输出的。 你的场景我们实践中的方法是按照用户id keyby之后再做localcache，并且如果用户id是long类型的话，localcache可以使用roaringbitmap，效率会比单纯的cache效率更好，占用内存更小。 并且频繁update state在资源有限的情况下是会有性能瓶颈的，这种场景下建议开窗口，窗口结束时update一次state即可。 Best, Yichao Yang -- 原始邮件 -- 发件人: user-zh-return-5056-1048262223=qq.com <520075...@qq.com> 发送时间: 2020年7月8日 18:09 收件人: user-zh https://ci.apache.org/projects/flink/flink-docs-master/zh/dev/stream/state/state.html#%E7%8A%B6%E6%80%81%E6%9C%89%E6%95%88%E6%9C%9F-ttl > Best, > Congxian > > > op <520075...@qq.com&gt; 于2020年7月8日周三 下午3:53写道： > > &gt; 您好，我主要是觉得Cache的自动过期比较好用 > &gt; > &gt; > &gt; --&amp;nbsp;原始邮件&amp;nbsp;-- > &gt; 发件人:&amp;nbsp;"Congxian Qiu"
??????flink????????
Hi, keyby?? warningPojo + String ??keybykey?? timestamp assigner Best, Yichao Yang -- -- ??: "??"
?????? ?????????? richfunction????????????????????????
Hi, ?? Best, Yichao Yang -- -- ??: "hdxg1101300...@163.com"
??????DataStream????uv????
Hi, ??uv??uv[1]?? [1] https://lists.apache.org/thread.html/rbe00ee38e2d07310d4e3c796de86c65205d1f5deecfc1678d9ebbdea%40%3Cuser-zh.flink.apache.org%3E -- -- ??: "?g???U?["
Re?? ??????????????????UV??????????????MapState??BloomFilter,??checkpoint????????????????????
Hi, ??Longuv?? RoaringBitMap[1] [1] https://mp.weixin.qq.com/s/jV0XmFxXFnzbg7kcKiiDbA Best, Yichao Yang -- -- ??: "user-zh" https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/ops/state/state_backends.html#%E5%A2%9E%E9%87%8F%E5%BF%AB%E7%85%A7 Best, Congxian x <35907...@qq.com> ??2020??8??27?? 1:48?? > > UV??MapStateBloomFilter??,checkpoint??bloomMapState
?????? ????FlinkSQL????operatoer??????savepoint??????????????
Hi Flink sql ??uidsql??datastream api Best, Yichao Yang -- -- ??: "kcz"<573693...@qq.com>; : 2020??6??10??(??) 11:27 ??: "user-zh"https://ci.apache.org/projects/flink/flink-docs-stable/ops/state/savepoints.html > > > > ?S
回复：flink任务checkpoint无法完成snapshot，且报kafka异常
Hi 看报错是checkpoint失败次数超过了最大限制导致任务失败。checkpoint间隔设置太小了，在我们团队通常都是分钟级别的interval，我们一般设置5分钟，checkpoint只是一个容错机制，没有特殊的需求场景不需要设置间隔那么短，并且频繁checkpoint会导致性能问题。 Best, Yichao Yang -- 原始邮件 -- 发件人: Zhefu PENG
回复：flink sql 消费kafka失败
Hi 看报错应该是kafKa有脏数据。 Best, Yichao Yang 发自我的iPhone -- 原始邮件 -- 发件人: Zhou Zach
??????flink??????????????????
Hi sink ??kafkakafka1.0??kafkaEXACTLY-ONCE Best, Yichao Yang -- -- ??: "??"
??????TTL ??????????????
Hi 00 00 sink? Best, Yichao Yang -- -- ??: "star"<3149768...@qq.com>; : 2020??6??10??(??) 2:34 ??: "user-zh@flink.apache.org"
??????????DataStreamUtils.reinterpretasKeyedStream??????
Hi flatmapKeyedstreamkeyby??keyedstream?? Best, Yichao Yang -- -- ??: ""<318666...@qq.com>; : 2020??6??10??(??) 7:18 ??: "user-zh"
??????flink on yarn??????????????????????
Hi yarnyarnyarn?? Best, Yichao Yang -- -- ??: "zjfpla...@hotmail.com"
??????flink????????????????
Hi broadcast?? Best, Yichao Yang -- -- ??: "xue...@outlook.com"https://go.microsoft.com/fwlink/?LinkId=550986>;
????????????????????????????????????
Hi watermark ??5??a??15?? b20?? Best, Yichao Yang -- -- ??: "steven chen"
?????? flink1.11 ??????(???? DDL ??????(???? Table ????))
Hi 1.2??like[1]query ??querykeyquery [1]https://ci.apache.org/projects/flink/flink-docs-master/dev/table/sql/create.html#create-table Best, Yichao Yang -- -- ??: "Kurt Young"
回复：flink sql 中怎么把ROW类型转换成INT
Hi row类型是不能强转int的，可以找一下阿里云flink sql的文档，其中有介绍哪些数据类型可以互转。 Best, Yichao Yang 发自我的iPhone -- 原始邮件 -- 发件人: Zhou Zach
回复：Re:关于keyby算子的疑问，如果本身数据比较分散，还有keyby的必要吗
Hi 个人理解一般情况下都是业务需要才会做keyby操作，比如想统计一个用户一分钟pv按照userid keyby。如果你的任务没有这样的业务需求完全不用考虑使用这些算子的。 Best, Yichao Yang -- 原始邮件 -- 发件人: Michael Ran
回复：异步io请求hbase,hbase获取数据环节背压高成为瓶颈问题
Hi 使用到缓存的话大多数情况下都要用到keyby，并且在keyby之后的算子使用缓存，保证相同key只会访问一个缓存，否则缓存命中率一般情况下都会很低。 Best, Yichao Yang 发自我的iPhone -- 原始邮件 -- 发件人: Benchao Li
回复：对于维表频繁更新，状态越来越大的场景如何保证数据的准确性
Hi 个人理解能不能做keyby+localcache的方式呢，保证一组key只在一个算子内，每个算子都会访问一部分维表数据，但是不同算子不会访问相同key的维度数据。 Best, Yichao Yang -- 原始邮件 -- 发件人: wangxiangyan
回复：Flink 多Sink 数据一致性保证
Hi 能不能描述一下你是什么样的场景下才会需要到多sink一致性保证？ 可不可以换一下思路，选择只出kafka的数据，然后把kafka数据实时同步hbase。 Best, Yichao Yang 发自我的iPhone -- 原始邮件 -- 发件人: xueaohui_...@163.com
??????flink ??????????
Hi ??INFO??jobcheckpoint Best, Yichao Yang -- -- ??: "Tony"
??????????????????
Hi ?? Best, Yichao Yang -- -- ??: "wangxiangyan"
??????????????????
Hi oom flinkOOM??flink flink??flinkOOM Best, Yichao Yang -- -- ??: "wangxiangyan"
??????????????????
Hi ?? 1.consumergroupid??consumer??flink 2.oom??flink??flink?? ?? a.?? kafka??500records/s??flink??1000records/s??flink job?? b.oom?? kafka??1500records/s60??flink??1000records/s?? ??x ??flink??offset?? 1500 * 1 + 500 * (x - 1) = 1000 * x ?? x = 2 kafka * 1 + kafka * x - 1 = flink * x flink??2offset?? flink??22?? -- -- ??: "wangxiangyan"
????????Flink SQL????NULL??????????????????
Hi ?? [1]http://apache-flink.147419.n8.nabble.com/flink-sql-null-false-td3640.html#a3658 Best, Yichao Yang -- -- ??: ""<153488...@qq.com>; : 2020??6??25??(??) 11:31 ??: "user-zh"
??????flinksql????????????????????
Hi ??hive table Best, Yichao Yang -- -- ??: "MuChen"<9329...@qq.com>; : 2020??6??29??(??) 4:53 ??: "user-zh"https://s1.ax1x.com/2020/06/29/Nf2dIA.png INFO15:34?? 2020-06-29 14:53:20,260 INFO org.apache.flink.api.common.io.LocatableInputSplitAssigner - Assigning remote split to host uhadoop-op3raf-core12 2020-06-29 14:53:22,845 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Source: HiveTableSource(vid, q70) TablePath: dw.video_pic_title_q70, PartitionPruned: false, PartitionNums: null (1/1) (68c24aa5 9c898cefbb20fbc929ddbafd) switched from RUNNING to FINISHED. 2020-06-29 15:34:52,982 INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint - Shutting YarnSessionClusterEntrypoint down with application status SUCCEEDED. Diagnostics null. 2020-06-29 15:34:52,984 INFO org.apache.flink.runtime.dispatcher.DispatcherRestEndpoint - Shutting down rest endpoint. 2020-06-29 15:34:53,072 INFO org.apache.flink.runtime.dispatcher.DispatcherRestEndpoint - Removing cache directory /tmp/flink-web-cdb67193-05ee-4a83-b957-9b7a9d85c23f/flink-web-ui 2020-06-29 15:34:53,073 INFO org.apache.flink.runtime.dispatcher.DispatcherRestEndpoint - http://uhadoop-op3raf-core1:44664 lost leadership 2020-06-29 15:34:53,074 INFO org.apache.flink.runtime.dispatcher.DispatcherRestEndpoint - Shut down complete. 2020-06-29 15:34:53,074 INFO org.apache.flink.yarn.YarnResourceManager - Shut down cluster because application is in SUCCEEDED, diagnostics null. 2020-06-29 15:34:53,076 INFO org.apache.flink.yarn.YarnResourceManager - Unregister application from the YARN Resource Manager with final status SUCCEEDED. 2020-06-29 15:34:53,088 INFO org.apache.hadoop.yarn.client.api.impl.AMRMClientImpl - Waiting for application to be successfully unregistered. 2020-06-29 15:34:53,306 INFO org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponent - Closing components. 2020-06-29 15:34:53,308 INFO org.apache.flink.runtime.dispatcher.runner.SessionDispatcherLeaderProcess - Stopping SessionDispatcherLeaderProcess. 2020-06-29 15:34:53,309 INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Stopping dispatcher akka.tcp://flink@uhadoop-op3raf-core1:38817/user/dispatcher. 2020-06-29 15:34:53,310 INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Stopping all currently running jobs of dispatcher akka.tcp://flink@uhadoop-op3raf-core1:38817/user/dispatcher. 2020-06-29 15:34:53,311 INFO org.apache.flink.runtime.jobmaster.JobMaster - Stopping the JobMaster for job default: insert into rt_app.app_video_cover_abtest_test ... 2020-06-29 15:34:53,322 INFO org.apache.hadoop.yarn.client.api.async.impl.AMRMClientAsyncImpl - Interrupted while waiting for queue java.lang.InterruptedException at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.reportInterruptAfterWait(AbstractQueuedSynchronizer.java:2014) at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2048) at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442) at org.apache.hadoop.yarn.client.api.async.impl.AMRMClientAsyncImpl$CallbackHandlerThread.run(AMRMClientAsyncImpl.java:287) 2020-06-29 15:34:53,324 INFO org.apache.hadoop.yarn.client.api.impl.ContainerManagementProtocolProxy - Opening proxy : uhadoop-op3raf-core12:2 ps: 1. kafka 2. flink1.10.0 ??SUCCEEDED ?? sql?? # -- ??5??vid??vid_group -- ??55mysql insert into rt_app.app_video_cover_abtest_test select begin_time, vid, vid_group, max(dv), max(click), max(vv), max(effectivevv) from( select t1.begin_time begin_time, t1.u_vid vid, t1.u_vid_group vid_group, dv, click, vv, if(effectivevv is null,0,effectivevv) effectivevv from ( -- dv??click??vv select CAST(TUMBLE_START(bjdt,INTERVAL '5' MINUTE) AS STRING) begin_time, cast(u_vid as bigint) u_vid, u_vid_group, sum(if(concat(u_mod,'-',u_ac)='emptylog-video_display' and u_c_module='M011',1,0)) dv, sum(if(concat(u_mod,'-',u_ac)='emptylog-video_click' and u_c_module='M011',1,0)) click, sum(if(concat(u_mod,'-',u_ac)='top-hits' and u_f_module='M011',1,0)) vv FROM rt_ods.ods_applog_vidsplit where u_vid is not null and trim(u_vid)<>'' and u_vid_grou
??????flink????kafka????????
Hi ?? dercd_seeme-3 partition kafkapartition Best, Yichao Yang -- -- ??: "??"
??????Flink-1.10.0 source??checkpoint??????????????
Hi checkpoint??cpugc?? Best, Yichao Yang -- -- ??: "Tianwang Li"
??????????RichReduceFunction??RichAggregateFunction
Hi ??stateuid keybyBloomFilter Best, Yichao Yang -- -- ??: "BenChen"
