update

Former-commit-id: 98a5b85745805470e3b25b3a395ee10c6d27c010
Former-commit-id: 0df7418d08c46f8e78d520cd79c89e56b86b17d6

labs/doris/conf/kafka-connect/kafka2elasticsearch.json

@@ -4,8 +4,17 @@
         "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
         "tasks.max": 1,
         "topics": "mysql.example.User",
-        "key.ignore": true,
+        "key.ignore": false,
+        "key.converter.schemas.enable": false,
+        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
         "connection.url": "http://elasticsearch:9200",
-        "type.name": "kafka-connect"
+        "type.name": "kafka-connect",
+        "write.method": "upsert",
+        "behavior.on.null.values": "delete",
+        "transforms": "key,ExtractFieldObject",
+        "transforms.key.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
+        "transforms.key.field": "Id",
+        "transforms.ExtractFieldObject.type": "org.apache.kafka.connect.transforms.ExtractField$Value",
+        "transforms.ExtractFieldObject.field": "after"
     }
 }

labs/doris/conf/kafka-connect/mysql2kafka.json

@@ -1,5 +1,5 @@
 {
-    "name": "msyql-source",
+    "name": "mysql-source",
     "config": {
         "connector.class": "io.debezium.connector.mysql.MySqlConnector",
         "tasks.max": "1",

labs/doris/conf/kafka-connect/start.sh

@@ -1,3 +1,7 @@
 #!/bin/bash
+echo 'init connectors start:'
+echo 'mysql2kafka'
 curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @mysql2kafka.json
-curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @kafka2elasticsearch.json
+echo 'kafka2elasticsearch'
+curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @kafka2elasticsearch.json
+echo 'init connectors end!'

labs/doris/readme.md

@@ -108,7 +108,33 @@ SHOW ALL ROUTINE LOAD;
 
 创建导入任务：
 
-CREATE ROUTINE LOAD example.job1 on User
+CREATE ROUTINE LOAD example.job2 on User
+--ALTER ROUTINE LOAD FOR example.job1
+columns (
+  `Id`,
+  `UserName`,
+  `SecurityStamp`,
+  `PasswordHash`,
+  `PasswordConfirmed`,
+  `Email`,
+  `EmailConfirmed`,
+  `PhoneNumber`,
+  `PhoneNumberConfirmed`,
+  `RealName`,
+  `IdentityNumber`,
+  `IdentityConfirmed`,
+  `NickName`,
+  `Avatar`,
+  `Sex`,
+  `Birthday`,
+  `LockoutEnabled`,
+  `AccessFailedCount`,
+  `LockoutEnd`,
+  `RowVersion`,
+  `Created`,
+  `Modified`,
+  `Deleted`
+  )
 PROPERTIES
 (
     "format"="json",