??????????????-flinksql??kafkasource????

[Sun.Zhu]

Hi,benchao??source??????????partition??????????????????checkpoint????????




| |
Sun.Zhu
|
|
??????17626017...@163.com
|

Signature is customized by Netease Mail Master

??2020??04??19?? 22:43???????????????? ??????
??????????????
 



------------------ ???????? ------------------
 ??????:&nbsp;"Benchao Li"<libenc...@gmail.com&gt;;
????????:&nbsp;2020??4??19??(??????) ????9:25
??????:&nbsp;"user-zh"<user-zh@flink.apache.org&gt;;

????:&nbsp;Re: ????????-flinksql??kafkasource????



??????????????????????????source????????????????kafka 
partition??????????????????

Jark Wu <imj...@gmail.com&gt; ??2020??4??19?????? ????8:22??????

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


--

Benchao Li
School of Electronics Engineering and Computer Science, Peking University
Tel:+86-15650713730
Email: libenc...@gmail.com; libenc...@pku.edu.cn