Confluent Platform: ksqlDB 实时数据 ETL

不念不忘少年蓝@ 2022-10-07 14:55 208阅读 0赞

### 文章目录 ###

*  1, 操作说明
 *   *  a, connector 故障排查
     *  b, 创建、查看已加载的连接器 (rest api)
     *  c, 指定connector 转换器
     *  d, 查看，创建，删除 schema (source connector 自动创建）
 *  2, 关系型数据库：之间数据流动
 *   *  a, pgsql -- > mysql
     *  b, mysql -- > mysql
     *  c, sqlserver -- > sqlserver
     *  d, sqlserver -- > mysql： query模式 (字段映射)
     *  e, sqlserver -- > oracle
     *  f, oracle -- > oracle
 *  3, 关系型数据库 ：导入hbase
 *   *  a, debezium-mysql-cdc -- > hbase: 实时数据
     *  b, oracle -- > hbase: 增量(时间字段)

# 1, 操作说明 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70]

数据源(pgsql,mysql,oracle,sqlserver) --> ksqldb ( kafka ) --> 目标端( mysql ,sqlserver, oracle,hbase )  
![在这里插入图片描述][20210618180945315.png]

## a, connector 故障排查 ##

// connector 故障排查1：
    ksql> describe connector `mysql-source-t1`; 
    Name                 : mysql-source-t1
    Class                : io.confluent.connect.jdbc.JdbcSourceConnector
    Type                 : source
    State                : RUNNING
    WorkerId             : 10.0.2.15:8083
    
     Task ID | State  | Error Trace                                                                                                                             
    ------------------------------------------------------------------------------------------------------------------------------------------------------------
     0       | FAILED | org.apache.kafka.connect.errors.ConnectException: Cannot make incremental queries using incrementing column id on `test`.`t1` because this column is nullable.
            at io.confluent.connect.jdbc.source.JdbcSourceTask.validateNonNullable(JdbcSourceTask.java:471)
            at io.confluent.connect.jdbc.source.JdbcSourceTask.start(JdbcSourceTask.java:165)
            at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:232)
            at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:189)
            at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:238)
            at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
            at java.util.concurrent.FutureTask.run(FutureTask.java:266)
            at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
            at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
            at java.lang.Thread.run(Thread.java:748)
    
    ------------------------------------------------------------------------------------------------------------------------------------------------------------
    
    // connector 故障排查2：
    $ curl -s "http://localhost:8083/connectors"
    ["jdbc_source_mysql_10"]
    #Check the status of the connector and its task[s]:
    $ curl -s "http://localhost:8083/connectors/jdbc_source_mysql_10/status"|jq '.'
    { 
      "name": "users_sink_mysql",
      "connector": { 
        "state": "RUNNING",
        "worker_id": "10.0.2.15:8083"
      },
      "tasks": [
        { 
          "id": 0,
          "state": "FAILED",
          "worker_id": "10.0.2.15:8083",
          "trace": "org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.\n\tat org.apache.kafka.connect.runtime....."
        }
      ],
      "type": "sink"
    }

## b, 创建、查看已加载的连接器 (rest api) ##

#创建连接器: rest api
    curl -s -X POST -H 'Content-Type: application/json'  http://localhost:8083/connectors  -d ' { "name": "jdbc-source", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "tasks.max": "1", "connection.url": "jdbc:sqlite:test.db", "mode": "incrementing", "incrementing.column.name": "id", "topic.prefix": "test-sqlite-jdbc-", "name": "jdbc-source" } }'
    
    #删除连接器: rest api
    $ curl -X DELETE localhost:8083/connectors/jdbc-source
    
    #查看加载的连接器插件: rest api
    $  curl -sS localhost:8083/connector-plugins | jq .[].class
    "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector"
    "io.confluent.connect.jdbc.JdbcSinkConnector"
    ...

## c, 指定connector 转换器 ##

*  `"key.converter" = 'org.apache.kafka.connect.storage.StringConverter'`,
 *  `"value.converter"= 'io.confluent.connect.avro.AvroConverter' //org.apache.kafka.connect.json.JsonConverter`,
 *  `"value.converter.schema.registry.url"='http://localhost:8081'`,

## d, 查看，创建，删除 schema (source connector 自动创建） ##

自定义topic schema 样例：[https://docs.confluent.io/platform/current/schema-registry/develop/using.html\#schemaregistry-using][https_docs.confluent.io_platform_current_schema-registry_develop_using.html_schemaregistry-using]

#create a new topic called test, edit schema:
    #avro Primitive Types:
    /* null: no value boolean: a binary value int: 32-bit signed integer long: 64-bit signed integer float: single precision (32-bit) IEEE 754 floating-point number double: double precision (64-bit) IEEE 754 floating-point number bytes: sequence of 8-bit unsigned bytes string: unicode character sequence */
    
    curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" http://localhost:8081/subjects/test-value/versions -d '
     {
      "type": "record",
      "name": "Payment",
      "namespace": "io.confluent.examples.clients.basicavro",
      "fields": [
        {
          "name": "id",
          "type": "string"
        },
        {
          "name": "amount",
          "type": "double"
        }
      ]
    } '
    
    
    #2, 查看schema: View the latest subject for transactions-value in Schema Registry:
    [root@hadoop01 /]# curl --silent -X GET http://localhost:8081/subjects/ |jq
    [
      "oracl-t1PER-value",
      "redo-log-topic-22-value",
      "oracle-redo-log-topicCloud-key"
    ]
    
    
    [root@hadoop01 /]# curl --silent -X GET http://localhost:8081/subjects/oracl-t1PER-value/versions/latest | jq .
    {
      "subject": "oracl-t1PER-value",
      "version": 1,
      "id": 2,
      "schema": "{\"type\":\"record\",\"name\":\"PER\",\"fields\":[{\"name\":\"ID\",\"type\":{\"type\":\"bytes\",\"scale\":0,\"precision\":38,\"connect.version\":1,\"connect.parameters\":{\"scale\":\"0\",\"connect.decimal.precision\":\"38\"},\"connect.name\":\"org.apache.kafka.connect.data.Decimal\",\"logicalType\":\"decimal\"}},{\"name\":\"NAME\",\"type\":[\"null\",\"string\"],\"default\":null}],\"connect.name\":\"PER\"}"
    }
    
    #3, 删除旧的schema
    curl -X DELETE http://localhost:8081/subjects/oracl-t1PER-value/versions/latest

# 2, 关系型数据库：之间数据流动 #

## a, pgsql – > mysql ##

参考：

*  [pgsql- - > elasticsearsh demo1 ][pgsql- - _ elasticsearsh demo1]
 *  [pgsql- - > elasticsearsh demo2 (mastering-kafka-streams-and-ksqldb/chapter-09 )][pgsql- - _ elasticsearsh demo2 _mastering-kafka-streams-and-ksqldb_chapter-09]
 *  jdbc连接器 doc: [https://docs.confluent.io/kafka-connect-jdbc/current/][https_docs.confluent.io_kafka-connect-jdbc_current]

source connector 参数说明：`poll.interval.ms=5000`, 默认5s一次查询新数据

#mysql> create database test;
    # create table test.per (id int PRIMARY KEY, title VARCHAR(120);
    #pgsql: docker run -d --name pgsql -p 5432:5432 -e POSTGRES_USER=root -e POSTGRES_PASSWORD=root -e POSTGRES_DB=root postgres:9.6.19-alpine
    #psql -U root -d root> CREATE TABLE titles ( id SERIAL PRIMARY KEY, title VARCHAR(120) );
    # INSERT INTO titles (title) values ('Stranger Things');
    # INSERT INTO titles (title) values ('Black Mirror');
    ksql>  CREATE SOURCE CONNECTOR `postgres-source` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSourceConnector',
        "connection.url"='jdbc:postgresql://192.168.56.117:5432/root?user=root&password=root',
        "mode"='incrementing',
        "incrementing.column.name"='id',
          "numeric.mapping"='best_fit',
        "topic.prefix"='',
        "table.whitelist"='titles',
        "key"='id');
    
    		#======================= 查看 kafka topic数据 ==================
    		ksql> set 'auto.offset.reset'='earliest';
    		ksql> print titles;
    		Key format: JSON or KAFKA_STRING
    		Value format: AVRO or KAFKA_STRING
    		rowtime: 2021/06/18 08:14:05.976 Z, key: 1, value: {
       "id": 1, "title": "Stranger Things"}
    		rowtime: 2021/06/18 08:14:05.976 Z, key: 2, value: {
       "id": 2, "title": "Black Mirror"}
    
    ksql>  CREATE sink CONNECTOR `mysql-sink` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSinkConnector',
        "connection.url"='jdbc:mysql://192.168.56.1:3306/test?user=root&password=root',
        "insert.mode"='upsert',
        "topics"='titles', 
        "table.name.format"='per',
        "pk.mode"='record_value', 
        "pk.fields"='id');

## b, mysql – > mysql ##

# mysql: CREATE TABLE per (id int PRIMARY KEY , title varchar(20))
    # mysql: CREATE TABLE per2(id int PRIMARY KEY , title varchar(20))
    ksql> 
    CREATE SOURCE CONNECTOR `mysql-source-per` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSourceConnector',
        "connection.url"='jdbc:mysql://192.168.56.1:3306/test?user=root&password=root',
        "mode"='incrementing',
        "incrementing.column.name"='id',
          "numeric.mapping"='best_fit',
        "topic.prefix"='mysql-source-',
        "table.whitelist"='per',
        "key"='id');
        
    		#查看源数据流
    		ksql> set 'auto.offset.reset'='earliest';
    		Successfully changed local property 'auto.offset.reset' to 'earliest'. Use the UNSET command to revert your change.
    		ksql> print `mysql-source-per`;
    		Key format: JSON or KAFKA_STRING
    		Value format: AVRO or KAFKA_STRING
    		rowtime: 2021/06/19 16:35:59.911 Z, key: 1, value: {
       "id": 1, "title": "Stranger Things"}
    		rowtime: 2021/06/19 16:35:59.913 Z, key: 2, value: {
       "id": 2, "title": "Black Mirror"}
    	
    CREATE sink CONNECTOR `mysql-sink-per2` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSinkConnector',
        "connection.url"='jdbc:mysql://192.168.56.1:3306/test?user=root&password=root',
        "insert.mode"='upsert',
        "topics"='mysql-source-per', 
        "table.name.format"='per2',
        "pk.mode"='record_value', 
        "pk.fields"='id');

## c, sqlserver – > sqlserver ##

# sqlserver: CREATE TABLE t1(id int PRIMARY KEY ,name varchar(10), job varchar(20))
    # sqlserver: CREATE TABLE t2(id int PRIMARY KEY ,name varchar(10), job varchar(20))
    ksql> 
    CREATE SOURCE CONNECTOR `sqlserver-source-t11` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSourceConnector',
         "connection.url"='jdbc:sqlserver://192.168.56.117:1434;databaseName=master',
         "connection.user"='sa',
         "connection.password"='abc123_123',
        "mode"='incrementing',
        "incrementing.column.name"='id',
          "numeric.mapping"='best_fit',
        "topic.prefix"='sqlserver-',
        "table.whitelist"='t1',
        "key"='id');
    
    CREATE sink CONNECTOR `sqlserver-sink-t2` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSinkConnector',
        "connection.url"='jdbc:sqlserver://192.168.56.117:1434;databaseName=master',
         "connection.user"='sa',
         "connection.password"='abc123_123',
        "insert.mode"='upsert',
        "topics"='sqlserver-t1', 
        "table.name.format"='t2',
        "pk.mode"='record_value', 
        "pk.fields"='id');

## d, sqlserver – > mysql： query模式 (字段映射) ##

# sqlserver: CREATE TABLE t1(id int PRIMARY KEY ,name varchar(10), job varchar(20))
    # mysql: CREATE TABLE per2(id int PRIMARY KEY ,name2 varchar(10), title varchar(20)) 
    ksql> 
    // query 模式
    CREATE SOURCE CONNECTOR `sqlserver2mysql-source-t11` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSourceConnector',
         "connection.url"='jdbc:sqlserver://192.168.56.117:1434;databaseName=master',
         "connection.user"='sa',
         "connection.password"='abc123_123',
        "mode"='incrementing',
        "incrementing.column.name"='id',
          "numeric.mapping"='best_fit',
        "topic.prefix"='sqlserver2mysql',
        "query"='select id ,name as name2, job as title from t1',
        "key"='id');
                                                                         
    CREATE sink CONNECTOR `sqlserver2mysql-sink-per2` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSinkConnector',
        "connection.url"='jdbc:mysql://192.168.56.1:3306/test?user=root&password=root',
        "insert.mode"='upsert',
        "topics"='sqlserver2mysql', 
        "table.name.format"='per2',
        "pk.mode"='record_value', 
        "pk.fields"='id');

## e, sqlserver – > oracle ##

# sqlserver: CREATE TABLE t1(id int PRIMARY KEY ,name varchar(10), job varchar(20))
    # oracle: CREATE TABLE t2(ID int PRIMARY KEY ,NAME varchar2(10), TITLE varchar2(20)) 
    ksql> 
    CREATE SOURCE CONNECTOR `sqlserver2oracl-source-t1` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSourceConnector',
        "connection.url"='jdbc:sqlserver://192.168.56.117:1434;databaseName=master',
        "connection.user"='sa',
        "connection.password"='abc123_123',
        "mode"='incrementing',
        "incrementing.column.name"='ID',
           "numeric.mapping"='best_fit',
        "topic.prefix"='sqlserver2oracl-t1',
        "query"='select id as ID ,name as NAME , job as TITLE from t1',
        "key"='ID');
        
    CREATE sink CONNECTOR `sqlserver2oracl-sink-t2` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSinkConnector',
        "connection.url"='jdbc:oracle:thin:@192.168.1.1:1521:test',
        "connection.user"='test1',
        "connection.password"='test1',
        "insert.mode"='upsert',
        "topics"='sqlserver2oracl-t1', 
        "table.name.format"='T2',
        "pk.mode"='record_value', 
        "pk.fields"='ID');

## f, oracle – > oracle ##

*  注意事项：schema.pattern 可以防止扫描所有的表（oracle 一般是一个数据库, 对应多个schema （模式空间,用户名称空间) )  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 1]

# oracle: CREATE TABLE t1(id int PRIMARY KEY ,name varchar(10), job varchar(20))
    # oracle: CREATE TABLE t2(ID int PRIMARY KEY ,NAME varchar2(10),job varchar(20), TITLE varchar2(20)) 
    ksql> 
    CREATE SOURCE CONNECTOR `oracl-source-t1` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSourceConnector',
        "connection.url"='jdbc:oracle:thin:@192.168.56.1:1521:orcl',
        "connection.user"='test1',
        "connection.password"='test1',
        "mode"='incrementing',
        "incrementing.column.name"='ID',
          "numeric.mapping"='best_fit',
        "topic.prefix"='oracl-t1',
        "query"='select id ,name , job from t1',
        "schema.pattern"='TEST1',
        "key"='ID');
        
        	#查看源数据流：
        	ksql> show topics;
    		Kafka Topic                 | Partitions | Partition Replicas
    		---------------------------------------------------------------
    		default_ksql_processing_log | 1          | 1
    		oracl-t1                    | 1          | 1		
    		#
    		ksql> print `oracl-t1`;
    		Key format: JSON or KAFKA_STRING
    		Value format: AVRO or KAFKA_STRING
    		rowtime: 2021/06/25 03:57:43.484 Z, key: 1, value: {
       "ID": "\u0001", "NAME": "a", "JOB": "java"}
    		rowtime: 2021/06/25 03:57:43.485 Z, key: 2, value: {
       "ID": "\u0002", "NAME": "b", "JOB": "py"}
    
    CREATE sink CONNECTOR `oracl-sink-t2` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSinkConnector',
        "connection.url"='jdbc:oracle:thin:@192.168.56.1:1521:orcl',
        "connection.user"='test1',
        "connection.password"='test1',
        "insert.mode"='upsert',
        "topics"='oracl-t1', 
        "table.name.format"='T2',
        "pk.mode"='record_value', 
        "pk.fields"='ID');

# 3, 关系型数据库 ：导入hbase #

kafka-hbase-sink插件zip包 (已改字段类型乱码问题)：[链接：https://pan.baidu.com/s/1f75ECo-i\_iBRtfcF5vYRRA 提取码：abcd ][https_pan.baidu.com_s_1f75ECo-i_iBRtfcF5vYRRA _abcd]  
源代码：[https://github.com/tayalnishu/kafka-connect-hbase][https_github.com_tayalnishu_kafka-connect-hbase]

*  修改1：源代码中JsonEventParser或 AvroEventParser, 类中区分了字段的数据类型，如果写入hbase的rowkey是这几种的话，就会导致`scan 'hbaseTab'`时乱码，所以需要统一改成String类型
 *  修改2：把avroData.fromConnectData(schema, value)改为 (Struct) value；并且使用schema.name()补充判断字段类型\[bytes, long\]
 *  修改3：hbase表单列族时，也能支持选择插入的列名（ hbase.destTable.d.columns=d1,d2）
 *  修改4：解析op\_type字段以识别delete操作，并删除hbase数据  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 2]  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 3]

## a, debezium-mysql-cdc – > hbase: 实时数据 ##

*  下载zip包 (选择0.9.4版本): [https://www.confluent.io/hub/debezium/debezium-connector-mysql][https_www.confluent.io_hub_debezium_debezium-connector-mysql]
 *  confluent-hub install debezium-debezium-connector-mysql-0.9.4.zip

doc:  
[https://docs.confluent.io/debezium-connect-mysql-source/current/mysql\_source\_connector\_config.html\#mysql-source-connector-config][https_docs.confluent.io_debezium-connect-mysql-source_current_mysql_source_connector_config.html_mysql-source-connector-config],  
[https://debezium.io/documentation/reference/0.9/tutorial.html][https_debezium.io_documentation_reference_0.9_tutorial.html]

# 安装hbase-sink插件: 
    # shell $> confluent-hub install ~/nishutayal-kafka-connect-hbase-1.0.1.zip
    # hbase: create 'www', 'f'
    # mysql: create table customers ( id int primary key, first_name varchar(100), last_name varchar(100), email varchar(100) );
    /* my.cnf配置如下：（启用bin-log) server-id = 1 log_bin = mysql-bin binlog_format = row binlog_row_image = full expire_logs_days = 10 */
    ksql> CREATE SOURCE CONNECTOR `mysql-dbz-source` WITH(
        "connector.class"='io.debezium.connector.mysql.MySqlConnector', 
         "transforms"='unwrap',
         "transforms.unwrap.type"='io.debezium.transforms.UnwrapFromEnvelope',
         "transforms.unwrap.drop.tombstones"=true,
         "transforms.unwrap.delete.handling.mode"='rewrite',
    
         "tasks.max"='1',
         "database.hostname"='192.168.56.117',
         "database.port"='32768',
         "database.server.name"='dbz3',
    
         "database.user"='root',
         "database.password"='123456',
         "database.server.id"='1',
         
         "database.whitelist"='test',
         "database.history.kafka.bootstrap.servers"='localhost:9092',
         "database.history.kafka.topic"='schema-changes.inventory'
    );
    		#查看源数据流
    		ksql> print `dbz3.inventory.customers`;
    		Key format: HOPPING(KAFKA_STRING) or TUMBLING(KAFKA_STRING) or KAFKA_STRING
    		Value format: AVRO
    		rowtime: 2021/06/28 03:59:38.246 Z, key: [Struct{
       @7594262153124655485/-], value: {
       "id": 1001, "first_name": "Sally", "last_name": "Thomas", "email": "sally.thomas@acme.com", "__deleted": "false"}
    		rowtime: 2021/06/28 03:59:38.246 Z, key: [Struct{
       @7594262153124655741/-], value: {
       "id": 1002, "first_name": "George", "last_name": "Bailey", "email": "gbailey@foobar.com", "__deleted": "false"}
    
    /* hbase sink 配置参数： name=kafka-cdc-hbase connector.class=io.svectors.hbase.sink.HBaseSinkConnector tasks.max=1 topics=test zookeeper.quorum=localhost:2181 event.parser.class=io.svectors.hbase.parser.AvroEventParser hbase.table.name=destTable hbase.destTable.rowkey.columns=id hbase.destTable.rowkey.delimiter=| hbase.destTable.family=c,d hbase.destTable.c.columns=c1,c2 hbase.destTable.d.columns=d1,d2 */
    CREATE sink CONNECTOR `www-hbase-sink2` WITH(
        "connector.class"='io.svectors.hbase.sink.HBaseSinkConnector',
        "zookeeper.quorum"='192.168.56.161:2181',
        "event.parser.class"='io.svectors.hbase.parser.AvroEventParser',
        "hbase.table.name"='www',
        "hbase.www.rowkey.columns"='id',
        "hbase.www.family"='f',
        "topics"='dbz3.inventory.customers'
        );

## b, oracle – > hbase: 增量(时间字段) ##

*  注意jdbc 连接使用的时区（默认UTC) :`"db.timezone": "Asia/Shanghai"`
 *  默认最初的时间是1970年，可以设置为当前时间： `"timestamp.initial": -1,` (必须为数值)  
    获取自定义时间的时间戳数值如下:

[root@c7-docker logs]# date -d '2 minute ago'
    Mon Jul 19 09:15:25 CST 2021
    [root@c7-docker logs]# date -d '2 minute ago' +%s
    1626657334
    [root@c7-docker logs]# date -d '10 minute ago' +%s
    1626656860
    
    [root@c7-docker logs]# date -d '1 hour ago'
    Mon Jul 19 08:17:08 CST 2021
    [root@c7-docker logs]# date -d '2 hour ago'
    Mon Jul 19 07:17:11 CST 2021
    
    [root@c7-docker logs]# date -d '2 day ago'
    Sat Jul 17 09:17:15 CST 2021
    [root@c7-docker logs]# date -d '1 month ago' +%s
    1624065481
    [root@c7-docker logs]# date -d '1 year ago' +%s
    1595121487

/* 方式1： 嵌套sql，时间格式转换后形成临时表 [root@c7-docker ~]# cat a.json { "name": "oracl-source-t1", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "tasks.max": "1", "connection.url": "jdbc:oracle:thin:@192.168.56.117:1523:helowin", "connection.user": "scott", "connection.password": "scott", "mode": "timestamp", "timestamp.column.name": "RECORD_TIME", "topic.prefix": "oracl-source-t1", "numeric.mapping": "best_fit", "validate.non.null": false, "query": "SELECT * FROM ( SELECT id, age, TO_DATE(COLUMN1, 'yyyy-mm-dd hh24:mi:ss') AS record_time FROM T1 )", "schema.pattern": "SCOTT", "db.timezone": "Asia/Shanghai", "key": "ID", "name": "oracl-source-t1" } } [root@c7-docker ~]# curl -s -X POST -H 'Content-Type: application/json' http://localhost:8083/connectors -d @a.json */
    
    # 方式2： 通过视图来获取格式转换后的时间字段
    #oracle1: CREATE TABLE patient( id varchar(10), name varchar(10),visit_time varchar(20) );
    # CREATE VIEW PATIENT_view AS select id as id2 ,name as name2, TO_DATE(VISIT_TIME,'yyyy-mm-dd hh24:mi:ss') AS VISIT_TIME from patient 
    #oracle2: CREATE TABLE patient2( id2 varchar(10), name2 varchar(10),visit_time varchar(20) );
    
    #validate.non.null: By default, the JDBC connector will validate that all incrementing and timestamp tables have NOT NULL 
    # set for the columns being used as their ID/timestamp. If the tables don’t, JDBC connector will fail to #start.
    # Setting this to false will disable these checks.
    ksql>
    CREATE SOURCE CONNECTOR `oracl-source-patient` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSourceConnector',
        "connection.url"='jdbc:oracle:thin:@192.168.56.161:1521:orcl',
        "connection.user"='test1',
        "connection.password"='test1',
         "table.types"='VIEW',
         "table.whitelist"='PATIENT_VIEW',
         "schema.pattern"= 'TEST1',
         "mode"='timestamp',
            "numeric.mapping"='best_fit',
         "timestamp.column.name"='VISIT_TIME',    
         "db.timezone"= 'Asia/Shanghai',
        
        "validate.non.null"=false,
        "TIMESTAMP"='VISIT_TIME', 
        "TIMESTAMP_FORMAT"='yyyy-mm-dd hh24:mi:ss' ,
        "topic.prefix"='oracl-patient',   
        "key"='ID2');
            
    	    #查看源数据流
    	    #INSERT INTO TEST1.PATIENT(ID, NAME, VISIT_TIME) VALUES('000123', '张三', '2020-01-01 12:23');
    	    ksql> print `oracl-patientPATIENT_VIEW`;
    		Key format: KAFKA_STRING
    		Value format: AVRO
    		rowtime: 2021/06/25 07:45:21.103 Z, key: 000123, value: {
       "ID2": "000123", "NAME2": "张三", "TIME": 1577881340000}
    		rowtime: 2021/06/25 07:45:21.103 Z, key: b123axa, value: {
       "ID2": "b123axa", "NAME2": "测试1", "TIME": 1577881340000}
    		
    CREATE sink CONNECTOR `www-hbase-sink2` WITH(
        "connector.class"='io.svectors.hbase.sink.HBaseSinkConnector',
        "zookeeper.quorum"='192.168.56.161:2181',
        "event.parser.class"='io.svectors.hbase.parser.AvroEventParser',
        "hbase.table.name"='www',
        "hbase.www.rowkey.columns"='ID2',
        "hbase.www.family"='f',
        "topics"='oracl-patientPATIENT_VIEW'
        );

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70]: /images/20221005/6448be91c0aa40419ebe26c3d22dcd0d.png
[20210618180945315.png]: /images/20221005/528db0853a604d118eb9768275499a1f.png
[https_docs.confluent.io_platform_current_schema-registry_develop_using.html_schemaregistry-using]: https://docs.confluent.io/platform/current/schema-registry/develop/using.html#schemaregistry-using
[pgsql- - _ elasticsearsh demo1]: https://github.com/confluentinc/demo-scene/tree/master/connect-jdbc
[pgsql- - _ elasticsearsh demo2 _mastering-kafka-streams-and-ksqldb_chapter-09]: https://github.com/mitch-seymour/mastering-kafka-streams-and-ksqldb/tree/master/chapter-09
[https_docs.confluent.io_kafka-connect-jdbc_current]: https://docs.confluent.io/kafka-connect-jdbc/current/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 1]: /images/20221005/cb7c077f141940a1a50866cf8aa60c9c.png
[https_pan.baidu.com_s_1f75ECo-i_iBRtfcF5vYRRA _abcd]: https://pan.baidu.com/s/1f75ECo-i_iBRtfcF5vYRRA
[https_github.com_tayalnishu_kafka-connect-hbase]: https://github.com/tayalnishu/kafka-connect-hbase
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 2]: /images/20221005/25113653f8d94a76a5068562b617a4f2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 3]: /images/20221005/ff3c4e344f8d4f3eb487a803ec7858ee.png
[https_www.confluent.io_hub_debezium_debezium-connector-mysql]: https://www.confluent.io/hub/debezium/debezium-connector-mysql
[https_docs.confluent.io_debezium-connect-mysql-source_current_mysql_source_connector_config.html_mysql-source-connector-config]: https://docs.confluent.io/debezium-connect-mysql-source/current/mysql_source_connector_config.html#mysql-source-connector-config
[https_debezium.io_documentation_reference_0.9_tutorial.html]: https://debezium.io/documentation/reference/0.9/tutorial.html