编辑readme.md

Author: zoudaokoulife

README.md

@@ -71,6 +71,7 @@ mvn clean package -Dmaven.test.skip
 
 * **confProp**
     * 描述：一些参数设置
+    * 格式: json
     * 必选：否
     * 默认值：无
     * 可选参数:

core/src/main/java/com/dtstack/flink/sql/table/AbsTableParser.java

@@ -102,7 +102,6 @@ public void parseFieldsInfo(String fieldsInfo, TableInfo tableInfo){
             System.arraycopy(filedInfoArr, 0, filedNameArr, 0, filedInfoArr.length - 1);
             String fieldName = String.join(" ", filedNameArr);
             String fieldType = filedInfoArr[filedInfoArr.length - 1 ].trim();
-            //Class fieldClass = ClassUtil.stringConvertClass(filedInfoArr[1].trim());
             Class fieldClass = ClassUtil.stringConvertClass(fieldType);
 
             tableInfo.addField(fieldName);

core/src/main/java/com/dtstack/flink/sql/watermarker/WaterMarkerAssigner.java

@@ -52,7 +52,6 @@ public DataStream assignWaterMarker(DataStream<Row> dataStream, RowTypeInfo type
 
         String[] fieldNames = typeInfo.getFieldNames();
         TypeInformation<?>[] fieldTypes = typeInfo.getFieldTypes();
-        String fields = StringUtils.join(fieldNames, ",");
 
         if(Strings.isNullOrEmpty(eventTimeFieldName)){
             return dataStream;

docs/colType.md

@@ -0,0 +1,16 @@
+| 支持的类型 | java对应类型   |
+| ------ | ----- |
+| boolean    | Boolean |
+| int        | Integer |
+| bigint     | Long |
+| tinyint    | Byte |
+| byte       | Byte |
+| short | Short | 
+| smallint | Short|
+| char | String|
+| varchar | String |
+| string | String|
+| float  | Float|
+| double | Double|
+| date   | Date |
+| timestamp | Timestamp |

docs/hbaseSink.md

@@ -0,0 +1,45 @@
+## 1.格式：
+```
+CREATE TABLE MyResult(
+    colFamily:colName colType,
+    ...
+ )WITH(
+    type ='hbase',
+    zookeeperQuorum ='ip:port[,ip:port]',
+    tableName ='tableName',
+    rowKey ='colFamily:colName[,colFamily:colName]',
+    parallelism ='1',
+    zookeeperParent ='/hbase'
+ )
+
+
+```
+
+## 2.参数：
+  * tableName ==> 在 sql 中使用的名称;即注册到flink-table-env上的名称
+  * colFamily:colName ==> hbase中的列族名称和列名称
+  * colType ==> 列类型 [colType支持的类型](colType.md)
+  
+  * type ==> 表明 输出表类型[mysql|hbase|elasticsearch]
+  * zookeeperQuorum ==> hbase zk地址,多个直接用逗号隔开
+  * zookeeperParent ==> zkParent 路径
+  * tableName ==> 关联的hbase表名称
+  * rowKey ==> hbase的rowkey关联的列信息
+  * parallelism ==> 并行度设置
+      
+  
+## 3.样例：
+```
+CREATE TABLE MyResult(
+    cf:channel STRING,
+    cf:pv BIGINT
+ )WITH(
+    type ='hbase',
+    zookeeperQuorum ='rdos1:2181',
+    tableName ='workerinfo',
+    rowKey ='cf:channel',
+    parallelism ='1',
+    zookeeperParent ='/hbase'
+ )
+
+ ```

docs/kafka09source.md

@@ -3,19 +3,33 @@
 CREATE TABLE tableName(
     colName colType,
     ...
-    function(channel) AS alias
+    function(colNameX) AS aliasName,
+    WATERMARK FOR colName AS withOffset( colName , delayTime )
  )WITH(
     type ='kafka09',
     bootstrapServers ='ip:port,ip:port...',
     zookeeperQuorum ='ip:port,ip:port/zkparent',
     offsetReset ='latest',
-    topic ='nbTest1',
-    parallelism ='1'
+    topic ='topicName',
+    parallelism ='parllNum'
  );
 ```
 ## 2.参数：
+  * tableName ==> 在 sql 中使用的名称;即注册到flink-table-env上的名称
+  * colName ==> 列名称
+  * colType ==> 列类型 [colType支持的类型](colType.md)
+  * function(colNameX) as aliasName ==> 支持在定义列信息的时候根据已有列类型生成新的列(函数可以使用系统函数和已经注册的UDF)
+  * WATERMARK FOR colName AS withOffset( colName , delayTime ) ==> 标识输入流生的watermake生成规则,根据指定的colName(当前支持列的类型为Long|Timestamp) 
+        和delayTime生成waterMark 同时会在注册表的使用附带上rowtime字段(如果未指定则默认添加proctime字段);
+        注意：添加该标识的使用必须设置系统参数 time.characteristic:EventTime; delayTime: 数据最大延迟时间(ms)
+  
   * type ==> kafka09
-  * bootstrapServers
+  * bootstrapServers ==> kafka bootstrap-server 地址信息(多个用逗号隔开)
+  * zookeeperQuorum ==> kafka zk地址信息(多个之间用逗号分隔)
+  * topic ==> 需要读取的 topic 名称
+  * offsetReset ==> 读取的topic 的offset初始位置[latest|earliest]
+  * parallelism ==> 并行度设置
+  
 ## 3.样例：
 ```
 CREATE TABLE MyTable(

docs/mysqlSide.md

@@ -0,0 +1,69 @@
+
+## 1.格式：
+```
+ CREATE TABLE tableName(
+     colName cloType,
+     ...
+     PRIMARY KEY(keyInfo),
+     PERIOD FOR SYSTEM_TIME
+  )WITH(
+     type='mysql',
+     url='jdbcUrl',
+     userName='dbUserName',
+     password='dbPwd',
+     tableName='tableName',
+     cache ='LRU',
+     cacheSize ='10000',
+     cacheTTLMs ='60000',
+     parallelism ='1',
+     partitionedJoin='false'
+  );
+```
+  
+## 2.参数
+
+  * tableName ==> 注册到flink的表名称(可选填;不填默认和hbase对应的表名称相同)
+  * colName ==> 列名称
+  * colType ==> 列类型 [colType支持的类型](colType.md)
+  * PERIOD FOR SYSTEM_TIME ==> 关键字表明该定义的表为维表信息
+  * PRIMARY KEY(keyInfo) ==> 维表主键定义;多个列之间用逗号隔开
+  * url ==> 连接mysql数据库 jdbcUrl 
+  * userName ==> mysql连接用户名
+  * password ==> mysql连接密码
+  * tableName ==> mysql表名称
+  * type ==> 表明维表的类型[hbase|mysql]
+  
+  * tableName ==> mysql 的表名称
+  * cache ==> 维表缓存策略(NONE/LRU)
+  
+      > * NONE: 不做内存缓存
+      > * LRU:
+      > > cacheSize ==> 缓存的条目数量
+      > > cacheTTLMs ==> 缓存的过期时间(ms)
+  
+  * partitionedJoin ==> 是否在維表join之前先根据 設定的key 做一次keyby操作(可以減少维表的数据缓存量)
+
+## 3.样例
+```
+create table sideTable(
+    channel String,
+    xccount int,
+    PRIMARY KEY(channel),
+    PERIOD FOR SYSTEM_TIME
+ )WITH(
+    type='mysql',
+    url='jdbc:mysql://172.16.8.104:3306/test?charset=utf8',
+    userName='dtstack',
+    password='abc123',
+    tableName='sidetest',
+    cache ='LRU',
+    cacheSize ='10000',
+    cacheTTLMs ='60000',
+    parallelism ='1',
+    partitionedJoin='false'
+ );
+
+
+```
+
+

docs/mysqlSink.md

@@ -0,0 +1,42 @@
+## 1.格式：
+```
+CREATE TABLE tableName(
+    colName colType,
+    ...
+    colNameX colType
+ )WITH(
+    type ='mysql',
+    url ='jdbcUrl',
+    userName ='userName',
+    password ='pwd',
+    tableName ='tableName',
+    parallelism ='parllNum'
+ );
+
+```
+
+## 2.参数：
+  * tableName ==> 在 sql 中使用的名称;即注册到flink-table-env上的名称
+  * colName ==> 列名称
+  * colType ==> 列类型 [colType支持的类型](colType.md)
+  * type ==> 表明 输出表类型[mysql|hbase|elasticsearch]
+  * url ==> 连接mysql数据库 jdbcUrl 
+  * userName ==> mysql连接用户名
+  * password ==> mysql连接密码
+  * tableName ==> mysql表名称
+  * parallelism ==> 并行度设置
+  
+## 3.样例：
+```
+CREATE TABLE MyResult(
+    channel VARCHAR,
+    pv VARCHAR
+ )WITH(
+    type ='mysql',
+    url ='jdbc:mysql://172.16.8.104:3306/test?charset=utf8',
+    userName ='dtstack',
+    password ='abc123',
+    tableName ='pv2',
+    parallelism ='1'
+ )
+ ```