Hi,benchao??source??????????partition??????????????????checkpoint????????
| | Sun.Zhu | | ??????17626017...@163.com | Signature is customized by Netease Mail Master ??2020??04??19?? 22:43???????????????? ?????? ?????????????? ------------------ ???????? ------------------ ??????: "Benchao Li"<libenc...@gmail.com>; ????????: 2020??4??19??(??????) ????9:25 ??????: "user-zh"<user-zh@flink.apache.org>; ????: Re: ????????-flinksql??kafkasource???? ??????????????????????????source????????????????kafka partition?????????????????? Jark Wu <imj...@gmail.com> ??2020??4??19?????? ????8:22?????? > Hi, > > ?????????????????????????????????????????????????????????????????? > ???????? Java ???????????? event time ??????????????????????????????????????????????????????????????????????watermark > ?????? 5s ??????. > ???????????????????? flink ??????????????event time ???????????????????????????? partition ?????????? partition > ?????????????????? > ????????????????????????????????????????4s????????????10s???????????????????????????????????????????????????????????????????? > > ?????????????????????? FLIP-27 ???????? > ???????????????? watermark delay?????????????????????? > > Best, > Jark > > > On Sat, 18 Apr 2020 at 22:53, ?????????????? <cai...@qq.com> wrote: > > > ???? > > > > > ??????????????????????????????????????????????kafka????????????3??????????????????????????????????????????????????????????????????140????????????????????????????????????1100?????????????????? > > > > > > > > ???? > > userbehavior???????? > > CREATE TABLE user_behavior ( > > &nbsp; &nbsp; user_id BIGINT, > > &nbsp; &nbsp; item_id BIGINT, > > &nbsp; &nbsp; category_id BIGINT, > > &nbsp; &nbsp; behavior STRING, > > &nbsp; &nbsp; ts TIMESTAMP(3), > > &nbsp; &nbsp; proctime as PROCTIME(), &nbsp; -- ???????????????????????????? > > &nbsp; &nbsp; WATERMARK FOR ts as ts - INTERVAL '5' SECOND &nbsp;-- > > ??ts??????watermark??ts?????????????? > > ) WITH ( > > &nbsp; &nbsp; 'connector.type' = 'kafka', &nbsp;-- ???? kafka connector > > &nbsp; &nbsp; 'connector.version' = 'universal', &nbsp;-- kafka > > ??????universal ???? 0.11 ?????????? > > &nbsp; &nbsp; 'connector.topic' = 'user_behavior', &nbsp;-- kafka topic > > &nbsp; &nbsp; 'connector.startup-mode' = 'earliest-offset', &nbsp;-- ?????? > > offset ???????? > > &nbsp; &nbsp; 'connector.properties.zookeeper.connect' = ' > > 192.168.0.150:2181', &nbsp;-- zookeeper ???? > > &nbsp; &nbsp; 'connector.properties.bootstrap.servers' = ' > > 192.168.0.150:9092', &nbsp;-- kafka broker ???? > > &nbsp; &nbsp; 'format.type' = 'json' &nbsp;-- ???????????? json > > ) > > > > ???????????????????? > > CREATE TABLE buy_cnt_per_hour (&nbsp; > > &nbsp; &nbsp; hour_of_day BIGINT, > > &nbsp; &nbsp; buy_cnt BIGINT > > ) WITH ( > > &nbsp; &nbsp; 'connector.type' = 'elasticsearch', -- ???? elasticsearch > > connector > > &nbsp; &nbsp; 'connector.version' = '6', &nbsp;-- elasticsearch ??????6 ?????? > > es 6+ ???? 7+ ?????? > > &nbsp; &nbsp; 'connector.hosts' = 'http://192.168.0.150:9200', &nbsp;-- > > elasticsearch ???? > > &nbsp; &nbsp; 'connector.index' = 'buy_cnt_per_hour', &nbsp;-- > > elasticsearch ?????????????????????????? > > &nbsp; &nbsp; 'connector.document-type' = 'user_behavior', -- > > elasticsearch ?? type???????????????????? > > &nbsp; &nbsp; 'connector.bulk-flush.max-actions' = '1', &nbsp;-- ?????????????? > > &nbsp; &nbsp; 'format.type' = 'json', &nbsp;-- ???????????? json > > &nbsp; &nbsp; 'update-mode' = 'append' > > ) > > > > ???????? > > INSERT INTO buy_cnt_per_hour&nbsp; > > SELECT HOUR(TUMBLE_START(ts,INTERVAL '1' HOUR)),COUNT(*)&nbsp; > > FROM user_behavior > > WHERE behavior = 'buy' > > GROUP BY TUMBLE(ts,INTERVAL '1' HOUR) > > > > kafka???????????? > > > > import com.alibaba.fastjson.JSONObject; > > import org.apache.kafka.clients.producer.KafkaProducer; > > import org.apache.kafka.clients.producer.ProducerRecord; > > > > import java.text.SimpleDateFormat; > > import java.util.*; > > > > > > public class UserBehaviorProducer { > > public static final String brokerList = "192.168.0.150:9092"; > > > > // public static final String topic="user_behavior"; > > public static final String topic = "user_behavior"; > > > > public static void main(String args[]) { > > > > //???????????????????? > > //???????????? > > Properties properties = new Properties(); > > properties.put("key.serializer", > > "org.apache.kafka.common.serialization.StringSerializer"); > > properties.put("value.serializer", > > "org.apache.kafka.common.serialization.StringSerializer"); > > properties.put("bootstrap.servers", brokerList); > > //????KafkaProducer ???? > > KafkaProducer<String, String&gt; producer = new > > KafkaProducer<&gt;(properties); > > //???????????????? > > //{"user_id": "952483", "item_id":"310884", "category_id": > > "4580532", "behavior": "pv", "ts": "2017-11-27T00:00:00Z"} > > //{"user_id": "794777", "item_id":"5119439", "category_id": > > "982926", "behavior": "pv", "ts": "2017-11-27T00:00:00Z"} > > String[] behaviors = {"pv", "buy", "coll", > "cart"};//???????????????????????????? > > JSONObject jsonObject = new JSONObject(); > > HashMap<String, String&gt; info = new HashMap<&gt;(); > > Random random = new Random(); > > SimpleDateFormat format = new > > SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'"); > > long date_long=getDate(); > > while (true) { > > jsonObject.put("user_id", random.nextInt(900000) + 100000 + > > ""); > > jsonObject.put("item_id", random.nextInt(900000) + 100000 + > > ""); > > jsonObject.put("category_id", random.nextInt(1000) + ""); > > jsonObject.put("behavior", behaviors[random.nextInt(4)]); > > jsonObject.put("ts", format.format(new Date(date_long))); > > String msg = jsonObject.toString(); > > System.out.println(msg); > > ProducerRecord<String, String&gt; record = new > > ProducerRecord<&gt;(topic, msg); > > producer.send(record); > > // date_long +=500+random.nextGaussian()*1000; > > date_long +=800+random.nextGaussian()*1500; > > try { > > Thread.sleep(60); > > } catch (InterruptedException e) { > > e.printStackTrace(); > > } > > } > > > > } > > > > private static long getDate() { > > Date date = new Date(); > > Calendar c = Calendar.getInstance(); > > c.setTime(date); > > //??????1??,?????????????????????? > > c.set(Calendar.DAY_OF_MONTH, 1); > > //????????0 > > c.set(Calendar.HOUR_OF_DAY, 0); > > //????????0 > > c.set(Calendar.MINUTE, 0); > > //??????0 > > c.set(Calendar.SECOND,0); > > //????????0 > > c.set(Calendar.MILLISECOND, 0); > > // ?????????????????????????????? > > return c.getTimeInMillis(); > > } > > } > > > > ------------------&nbsp;????????&nbsp;------------------ > > ??????:&nbsp;"Jark Wu"<imj...@gmail.com&gt;; > > ????????:&nbsp;2020??4??18??(??????) ????10:08 > > ??????:&nbsp;"user-zh"<user-zh@flink.apache.org&gt;; > > > > ????:&nbsp;Re: ????????-flinksql??kafkasource???? > > > > > > > > Hi, > > > > ????????????????Flink 1.10 ???? SQL CLI ?????? CREATE VIEW ?????? TableEnvironment ???????????? > > ??????????????????????????????????"????????????????/partition??????????????????????" ???????????????????????????????? > > > > Best, > > Jark > > > > On Sat, 18 Apr 2020 at 18:38, ?????????????? <cai...@qq.com&gt; wrote: > > > > &gt; ???????? > > &gt; &amp;nbsp; &amp;nbsp; > > &gt; > &amp;nbsp;????????????????????????????????flinkSQL???????????????????????????????????????????????????????????????????????? > > &gt; &amp;nbsp; &amp;nbsp; > > &gt; > > > &amp;nbsp;??????????????????sql-client??????kafkasource??????????????kafka????????????????????????????????????????????????????kafka??????????????????????????kafka????????????????????????????????????kafkasource????????????????????????????????????????????????????/partition??????????????????????????????kafkasource??????????watermark??partition???????????????????????????????????? > > &gt; &amp;nbsp;&amp;nbsp; > > &gt; &amp;nbsp; &amp;nbsp; > > ????????????sql-client??????????view??????????????tableEnv.sqlUpdate("create > > &gt; view as ...")???????????????????? > > &gt; Exception in thread "main" > org.apache.flink.table.api.TableException: > > &gt; Unsupported query: CREATE VIEW rich_user_behavior AS > > &gt; SELECT U.user_id, U.item_id, U.behavior,&amp;nbsp; > > &gt; &amp;nbsp; CASE C.parent_category_id > > &gt; &amp;nbsp; &amp;nbsp; WHEN 1 THEN '????????' > > &gt; &amp;nbsp; &amp;nbsp; WHEN 2 THEN '????????' > > &gt; &amp;nbsp; &amp;nbsp; WHEN 3 THEN '????' > > &gt; &amp;nbsp; &amp;nbsp; WHEN 4 THEN '????' > > &gt; &amp;nbsp; &amp;nbsp; WHEN 5 THEN '????' > > &gt; &amp;nbsp; &amp;nbsp; WHEN 6 THEN '3C????' > > &gt; &amp;nbsp; &amp;nbsp; WHEN 7 THEN '????????' > > &gt; &amp;nbsp; &amp;nbsp; WHEN 8 THEN '????' > > &gt; &amp;nbsp; &amp;nbsp; ELSE '????' > > &gt; &amp;nbsp; END AS category_name > > &gt; FROM user_behavior AS U LEFT JOIN category_dim FOR SYSTEM_TIME AS OF > > &gt; U.proctime AS C > > &gt; ON U.category_id = C.sub_category_id > > &gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at > > &gt; > > > org.apache.flink.table.planner.delegation.ParserImpl.lambda$parse$0(ParserImpl.java:67) > > &gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at > > &gt; > > > org.apache.flink.table.planner.delegation.ParserImpl$$Lambda$213/1241938981.get(Unknown > > &gt; Source) > > &gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at > > java.util.Optional.orElseThrow(Optional.java:290) > > &gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at > > &gt; > > > org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:67) > > &gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at > > &gt; > > > org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlUpdate(TableEnvironmentImpl.java:484) > > &gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at > > &gt; > > > com.cmb.sql_train.myexample.MyUserBehaviorExample.main(MyUserBehaviorExample.java:51) > > &gt; > > &gt; > > &gt; > > &gt; > > &gt; > > &gt; &amp;nbsp; &amp;nbsp; &amp;nbsp; &amp;nbsp; ?????????????????? > -- Benchao Li School of Electronics Engineering and Computer Science, Peking University Tel:+86-15650713730 Email: libenc...@gmail.com; libenc...@pku.edu.cn