mysql 实时同步到 es

CREATE TABLE from_mysql (  id int,  cid int NOT NULL,  gid bigint NOT NULL,  content varchar,  create_time TIMESTAMP(3)  ,  PRIMARY KEY (id) NOT ENFORCED) WITH (  'connector' = 'mysql-cdc',  'hostname' = 'mysql-ip',  'port' = '3306',  'username' = 'mysqluser',  'password' = 'mysqlpwd',  'database-name' = 'mysqldb',  'debezium.snapshot.locking.mode' = 'none',  'table-name' = 'tb_test[0-9]?[0-9]',  'server-id' = '100-110',  'server-time-zone' = 'Asia/Shanghai',  'debezium.skipped.operations' = 'd',  'debezium.snapshot.mode' = 'schema_only',  'debezium.min.row.count.to.stream.results' = '50000');
CREATE TABLE to_es (  id string,  tableid int,  tablename string,  cid int NOT NULL,  gid string NOT NULL,  content string,  create_time string,  PRIMARY KEY (id,companyId) NOT ENFORCED) WITH (   'connector.type' = 'elasticsearch',    'connector.version' = '7',    'connector.hosts' = 'http://ip:9200',   'connector.index' = 'myindex',   'connector.document-type' = '_doc',   'connector.username' = 'elastic',   'connector.password' = 'password123',   'update-mode' = 'upsert',   'connector.key-delimiter' = '$',   'connector.key-null-literal' = 'n/a',   'connector.failure-handler' = 'retry-rejected',   'connector.flush-on-checkpoint' = 'true',   'connector.bulk-flush.max-actions' = '10000',   'connector.bulk-flush.max-size' = '2 mb',   'connector.bulk-flush.interval' = '2000',   'connector.connection-max-retry-timeout' = '300',   'format.type' = 'json');
INSERT INTO to_esSELECTconcat(CAST(id as string),'-',CAST(mod(cid,100) AS VARCHAR)) as id, id tableid,tablename,cid,gid,content,DATE_FORMAT(create_time, 'yyyy-MM-dd HH:mm:ss') as create_timefrom from_mysql

PUT myindex-20230314/{   "mappings": {    "properties": {      "id":{        "type": "text"      },      "tableid":{        "type": "long"      },      "cid":{        "type": "long"      },      "gid":{        "type": "text",    "analyzer": "my_analyzer"      },      "content":{        "type": "text",    "analyzer": "my_analyzer"      },            "create_time" : {          "type" : "keyword"        }    }  },  "settings": {    "index":{      "number_of_shards": "10",      "number_of_replicas": "1",      "refresh_interval" : "1s",      "translog": {        "sync_interval": "30s",        "durability": "async"      },      "codec": "best_compression",       "analysis": {      "analyzer": {        "my_analyzer": {          "tokenizer": "my_tokenizer",          "filter": [            "lowercase"          ]        }      },      "tokenizer": {        "my_tokenizer": {          "type": "ngram",          "min_gram": 1,          "max_gram": 2,          "token_chars": [            "letter",            "digit","whitespace","punctuation","symbol"          ]        }      }    }    }  }}

 POST /_aliases{    "actions": [        { "add":    { "index": "myindex-20230314", "alias": "myindex" }}    ]}