Merge branch '1.8_release_3.10.x' into 1.8_release_3.10.x_jingdongfang

# Conflicts:
#	flinkx-rdb/flinkx-rdb-reader/src/main/java/com.dtstack.flinkx.rdb.inputformat/JdbcInputFormat.java

Author: kanata163

.gitignore

@@ -1,11 +1,14 @@
 # Created by .ignore support plugin (hsz.mobi)
 .idea/
 plugins/
+syncplugins/
 *.iml
 target/
 lib/
 jobs/
 nohup.out
 flinkconf/
 hadoopconf/
-/default_task_id_output
+/default_task_id_output
+/syncplugins
+flinkx-test/

.gitlab-ci.yml

@@ -0,0 +1,11 @@
+build:
+  stage: test
+  script:
+    - sh ci/install_jars.sh
+    - mvn clean org.jacoco:jacoco-maven-plugin:0.7.8:prepare-agent package -Dmaven.test.failure.ignore=true -q
+    - mvn sonar:sonar -Dsonar.projectKey="dt-insight-engine/flinkx" -Dsonar.host.url=http://172.16.100.198:9000 -Dsonar.jdbc.url=jdbc:postgresql://172.16.100.198:5432/sonar -Dsonar.java.binaries=target/sonar -Dsonar.login=11974c5e9a29625efa09fdc3c3fdc031efb1aab1
+    - sh ci/sonar_notify.sh
+  only:
+    - 1.8_dev
+  tags:
+    - dt-insight-engine

README.md

@@ -25,6 +25,8 @@
 * 单机模式：对应Flink集群的单机模式
 * standalone模式：对应Flink集群的分布式模式
 * yarn模式：对应Flink集群的yarn模式
+* yarnPer模式: 对应Flink集群的Per-job模式
+
 
 ### 3.2 执行环境
 
@@ -52,6 +54,7 @@ mvn clean package -Dmaven.test.skip
     * local: 本地模式
     * standalone: 独立部署模式的flink集群
     * yarn: yarn模式的flink集群，需要提前在yarn上启动一个flink session，使用默认名称"Flink session cluster"
+    * yarnPer: yarn模式的flink集群，单独为当前任务启动一个flink session，使用默认名称"Flink per-job cluster"
   * 必选：否
   * 默认值：local
 
@@ -61,15 +64,15 @@ mvn clean package -Dmaven.test.skip
   * 必选：是
   * 默认值：无
 
-* **plugin**
+* **pluginRoot**
 
-  * 描述：插件根目录地址，也就是打包后产生的plugins目录。
+  * 描述：插件根目录地址，也就是打包后产生的pluginRoot目录。
   * 必选：是
   * 默认值：无
 
 * **flinkconf**
 
-  * 描述：flink配置文件所在的目录（单机模式下不需要），如/hadoop/flink-1.4.0/conf
+  * 描述：flink配置文件所在的目录（单机模式下不需要），如/opt/dtstack/flink-1.8.1/conf
   * 必选：否
   * 默认值：无
 
@@ -78,25 +81,79 @@ mvn clean package -Dmaven.test.skip
   * 描述：Hadoop配置文件（包括hdfs和yarn）所在的目录（单机模式下不需要），如/hadoop/etc/hadoop
   * 必选：否
   * 默认值：无
+  
+* **flinkLibJar**
+  
+  * 描述：flink lib所在的目录（单机模式下不需要），如/opt/dtstack/flink-1.8.1/lib
+  * 必选：否
+  * 默认值：无
+  
+* **confProp**
+  
+  * 描述：flink相关参数，如{\"flink.checkpoint.interval\":200000}
+  * 必选：否
+  * 默认值：无
+     
+* **queue**
+  
+  * 描述：yarn队列，如default
+  * 必选：否
+  * 默认值：无
+  
+* **pluginLoadMode**
+  
+  * 描述：yarnPer模式插件加载方式：
+    * classpath：提交任务时不上传插件包，需要在yarn-node节点pluginRoot目录下部署插件包，但任务启动速度较快
+    * shipfile：提交任务时上传pluginRoot目录下部署插件包的插件包，yarn-node节点不需要部署插件包，任务启动速度取决于插件包的大小及网络环境
+  * 必选：否
+  * 默认值：classpath        
 
 #### 3.4.2 启动数据同步任务
 
 * **以本地模式启动数据同步任务**
 
 ```
-bin/flinkx -mode local -job /Users/softfly/company/flink-data-transfer/jobs/task_to_run.json -pluginRoot /Users/softfly/company/flink-data-transfer/plugins -confProp "{"flink.checkpoint.interval":60000,"flink.checkpoint.stateBackend":"/flink_checkpoint/"}" -s /flink_checkpoint/0481473685a8e7d22e7bd079d6e5c08c/chk-*
+bin/flinkx -mode local \
+            -job /Users/softfly/company/flink-data-transfer/jobs/task_to_run.json \
+            -plugin /Users/softfly/company/flink-data-transfer/plugins \
+            -confProp "{"flink.checkpoint.interval":60000,"flink.checkpoint.stateBackend":"/flink_checkpoint/"}" \
+            -s /flink_checkpoint/0481473685a8e7d22e7bd079d6e5c08c/chk-*
 ```
 
 * **以standalone模式启动数据同步任务**
 
 ```
-bin/flinkx -mode standalone -job /Users/softfly/company/flink-data-transfer/jobs/oracle_to_oracle.json  -pluginRoot /Users/softfly/company/flink-data-transfer/plugins -flinkconf /hadoop/flink-1.4.0/conf -confProp "{"flink.checkpoint.interval":60000,"flink.checkpoint.stateBackend":"/flink_checkpoint/"}" -s /flink_checkpoint/0481473685a8e7d22e7bd079d6e5c08c/chk-*
+bin/flinkx -mode standalone \
+            -job /Users/softfly/company/flink-data-transfer/jobs/oracle_to_oracle.json \
+            -plugin /Users/softfly/company/flink-data-transfer/plugins \
+            -flinkconf /hadoop/flink-1.4.0/conf \
+            -confProp "{"flink.checkpoint.interval":60000,"flink.checkpoint.stateBackend":"/flink_checkpoint/"}" \
+            -s /flink_checkpoint/0481473685a8e7d22e7bd079d6e5c08c/chk-*
 ```
 
 * **以yarn模式启动数据同步任务**
 
 ```
-bin/flinkx -mode yarn -job /Users/softfly/company/flinkx/jobs/mysql_to_mysql.json  -pluginRoot /opt/dtstack/flinkplugin/syncplugin -flinkconf /opt/dtstack/myconf/conf -yarnconf /opt/dtstack/myconf/hadoop -confProp "{"flink.checkpoint.interval":60000,"flink.checkpoint.stateBackend":"/flink_checkpoint/"}" -s /flink_checkpoint/0481473685a8e7d22e7bd079d6e5c08c/chk-*
+bin/flinkx -mode yarn \
+            -job /Users/softfly/company/flinkx/jobs/mysql_to_mysql.json \
+            -plugin /opt/dtstack/flinkplugin/syncplugin \
+            -flinkconf /opt/dtstack/myconf/conf \
+            -yarnconf /opt/dtstack/myconf/hadoop \
+            -confProp "{"flink.checkpoint.interval":60000,"flink.checkpoint.stateBackend":"/flink_checkpoint/"}" \
+            -s /flink_checkpoint/0481473685a8e7d22e7bd079d6e5c08c/chk-*
+```
+
+* **以perjob模式启动数据同步任务**
+
+```
+bin/flinkx -mode yarnPer \
+            -job /test.json \
+            -pluginRoot /opt/dtstack/syncplugin \
+            -flinkconf /opt/dtstack/flink-1.8.1/conf \
+            -yarnconf /opt/dtstack/hadoop-2.7.3/etc/hadoop \
+            -flinkLibJar /opt/dtstack/flink-1.8.1/lib \
+            -confProp {\"flink.checkpoint.interval\":200000} \
+            -queue c -pluginLoadMode classpath
 ```
 
 ## 4 数据同步任务模版
@@ -187,6 +244,25 @@ setting包括speed、errorLimit和dirty三部分，分别描述限速、错误
 
 restore配置请参考[断点续传](docs/restore.md)
 
+#### 4.1.5  log
+
+```
+"log" : {
+        "isLogger": true,
+        "level" : "warn",
+        "path" : "/opt/log/",
+        "pattern":""
+      }
+```
+* isLogger: 日志是否保存到磁盘, `true`: 是; `false`(默认): 否;
+* level: 日志输出级别, `trace`, `debug`, `info`(默认), `warn`, `error`;
+* path: 日志保存路径, 默认为`/tmp/dtstack/flinkx/`, 日志名称为当前flink任务的jobID，如: `97501729f8c44c260d889d099968cc74.log`
+* pattern: 日志输出格式
+    * log4j默认格式为: `%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{60} %X{sourceThread} - %msg%n`; 
+    * logback默认格式为: `%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n`
+
+注意：该日志记录功能只会记录`com.dtstack`包下的输出日志, 如需变更，可修改类参数`DtLogger.LOGGER_NAME`。
+
 ### 4.2 content
 
 ```
@@ -221,7 +297,7 @@ reader和writer包括name和parameter，分别表示插件名称和插件参数
 
 ### 5.1 读取插件
 
-* [关系数据库读取插件](docs/rdbreader.md)
+* [关系数据库读取插件(Mysql,Oracle,Sqlserver,Postgresql,Db2,Gbase)](docs/rdbreader.md)
 * [分库分表读取插件](docs/rdbdreader.md)
 * [HDFS读取插件](docs/hdfsreader.md)
 * [HBase读取插件](docs/hbasereader.md)
@@ -234,11 +310,14 @@ reader和writer包括name和parameter，分别表示插件名称和插件参数
 * [MySQL binlog读取插件](docs/binlog.md)
 * [KafKa读取插件](docs/kafkareader.md)
 * [Kudu读取插件](docs/kudureader.md)
+* [Emqx读取插件](docs/emqxreader.md)
+* [Oracle实时采集插件](docs/logminer.md)
+* [SqlServerCdc实时采集插件](docs/sqlservercdc.md)
 
 
 ### 5.2 写入插件
 
-* [关系数据库写入插件](docs/rdbwriter.md)
+* [关系数据库写入插件(Mysql,Oracle,Sqlserver,Postgresql,Db2,Gbase)](docs/rdbwriter.md)
 * [HDFS写入插件](docs/hdfswriter.md)
 * [HBase写入插件](docs/hbasewriter.md)
 * [Elasticsearch写入插件](docs/eswriter.md)
@@ -251,6 +330,8 @@ reader和writer包括name和parameter，分别表示插件名称和插件参数
 * [Kafka写入插件](docs/kafkawriter.md)
 * [Hive写入插件](docs/hivewriter.md)
 * [Kudu写入插件](docs/kuduwriter.md)
+* [Emqx写入插件](docs/emqxwriter.md)
+
 
 [断点续传和实时采集功能介绍](docs/restore.md)
 
@@ -269,4 +350,4 @@ reader和writer包括name和parameter，分别表示插件名称和插件参数
 
 ## 7.招聘信息
 
- 1.大数据平台开发工程师，想了解岗位详细信息可以添加本人微信号ysqwhiletrue,注明招聘，如有意者发送简历至[email protected]。
+ 1.大数据平台开发工程师，想了解岗位详细信息可以添加本人微信号ysqwhiletrue,注明招聘，如有意者发送简历至[email protected]。

bin/install_jars.bat

@@ -0,0 +1,8 @@
+call mvn install:install-file -DgroupId=com.ibm.db2 -DartifactId=db2jcc -Dversion=3.72.44 -Dpackaging=jar -Dfile=../jars/db2jcc-3.72.44.jar
+
+call mvn install:install-file -DgroupId=com.github.noraui -DartifactId=ojdbc8 -Dversion=12.2.0.1 -Dpackaging=jar -Dfile=../jars/ojdbc8-12.2.0.1.jar
+
+call mvn install:install-file -DgroupId=com.esen.jdbc -DartifactId=gbase -Dversion=8.3.81.53 -Dpackaging=jar -Dfile=../jars/gbase-8.3.81.53.jar
+
+call mvn install:install-file -DgroupId=dm.jdbc.driver -DartifactId=dm7 -Dversion=18.0.0 -Dpackaging=jar -Dfile=../jars/Dm7JdbcDriver18.jar
+

bin/install_jars.sh

@@ -0,0 +1,13 @@
+#!/usr/bin/env bash
+
+## db2 driver
+mvn install:install-file -DgroupId=com.ibm.db2 -DartifactId=db2jcc -Dversion=3.72.44 -Dpackaging=jar -Dfile=../jars/db2jcc-3.72.44.jar
+
+## oracle driver
+mvn install:install-file -DgroupId=com.github.noraui -DartifactId=ojdbc8 -Dversion=12.2.0.1 -Dpackaging=jar -Dfile=../jars/ojdbc8-12.2.0.1.jar
+
+## gbase driver
+mvn install:install-file -DgroupId=com.esen.jdbc -DartifactId=gbase -Dversion=8.3.81.53 -Dpackaging=jar -Dfile=../jars/gbase-8.3.81.53.jar
+
+## dm driver
+mvn install:install-file -DgroupId=dm.jdbc.driver -DartifactId=dm7 -Dversion=18.0.0 -Dpackaging=jar -Dfile=../jars/Dm7JdbcDriver18.jar

ci/install_jars.sh

@@ -0,0 +1,13 @@
+#!/usr/bin/env bash
+
+## db2 driver
+mvn install:install-file -DgroupId=com.ibm.db2 -DartifactId=db2jcc -Dversion=3.72.44 -Dpackaging=jar -Dfile=jars/db2jcc-3.72.44.jar
+
+## oracle driver
+mvn install:install-file -DgroupId=com.github.noraui -DartifactId=ojdbc8 -Dversion=12.2.0.1 -Dpackaging=jar -Dfile=jars/ojdbc8-12.2.0.1.jar
+
+## gbase driver
+mvn install:install-file -DgroupId=com.esen.jdbc -DartifactId=gbase -Dversion=8.3.81.53 -Dpackaging=jar -Dfile=jars/gbase-8.3.81.53.jar
+
+## dm driver
+mvn install:install-file -DgroupId=dm.jdbc.driver -DartifactId=dm7 -Dversion=18.0.0 -Dpackaging=jar -Dfile=jars/Dm7JdbcDriver18.jar

ci/sonar_notify.sh

@@ -0,0 +1,14 @@
+#!/bin/bash
+#参考钉钉文档 https://open-doc.dingtalk.com/microapp/serverapi2/qf2nxq
+ sonarreport=$(curl -s http://172.16.100.198:8082/?projectname=dt-insight-engine/flinkx)
+ curl -s "https://oapi.dingtalk.com/robot/send?access_token=e2718f7311243d2e58fa2695aa9c67a37760c7fce553311a32d53b3f092328ed" \
+   -H "Content-Type: application/json" \
+   -d "{
+     \"msgtype\": \"markdown\",
+     \"markdown\": {
+         \"title\":\"sonar代码质量\",
+         \"text\": \"## sonar代码质量报告: \n
+> [sonar地址](http://172.16.100.198:9000/dashboard?id=dt-insight-engine/flinkx) \n
+> ${sonarreport} \n\"
+     }
+ }"

docs/binlog.md

@@ -1,6 +1,11 @@
 # MySQL binlog读取插件（*reader）
 
-## 1. 配置样例
+## 1.首先给莫个用户赋权，有读binglog的权限
+
+     GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT
+     ON *.* TO 'xxx'@'%' IDENTIFIED BY 'xxx';
+
+## 2. 配置样例
 
 ```json
 {
@@ -9,7 +14,7 @@
             "reader": {
                 "parameter": {
                     "jdbcUrl" : "jdbc:mysql://127.0.0.1:3306/test?charset=utf8",
-                    "username" : "username"，
+                    "username" : "username",
                     "password" : "password",
                     "host" : "127.0.0.1",
                     "port": 3306,
@@ -164,10 +169,10 @@
              "table":"tb1",
              "ts":1231232,
              "ingestion":123213,
-             "before_id":{
+             "before":{
                  "id":1
              },
-             "after_id":{
+             "after":{
                  "id":2
              }
         }