Skip to content

Latest commit

 

History

History
161 lines (140 loc) · 10.4 KB

README.md

File metadata and controls

161 lines (140 loc) · 10.4 KB
Raw

elasticsearch-cdc 插件支持将elasticsearch 的cdc 数据异步同步到kafka

elasticsearch-cdc 插件设计

image 为了较大的吞吐量，采用异步的方式向kafka发送消息

安装

1. 编译打包
mvn clean package
2. 移除之前版本(如果之前安装过)
${elasticsearch_home}/bin/elasticsearch-plugin remove elasticsearch-cdc
3. 安装
${elasticsearch_home}/bin/elasticsearch-plugin install file:///${elasticsearch_cdc_home_dir}/target/releases/elasticsearch-cdc-1.0-SNAPSHOT.zip
4. 配置es的javax相关权限
/etc/elasticsearch/jvm.options 文件加入相关java配置
-Djava.security.policy=/your_elasticsearch_home/plugins/elasticsearch-cdc/plugin-security.policy
5. 重启 elasticsearch
service restart elasticsearch

注：对于es分布式集群，每一个节点都应该安装插件

使用

设置集群属性

PUT _cluster/settings
{
  "persistent": {
    "indices.cdc.request.timeout.ms": 30000,
    "indices.cdc.send.buffer.bytes": 131072,
    "indices.cdc.acks": "all",
    "indices.cdc.compression.type": "none",
    "indices.cdc.receive.buffer.bytes": 32768,
    "indices.cdc.batch.size": 16384,
    "indices.cdc.linger.ms": 1000,
    "indices.cdc.buffer.memory": 33554432,
    "indices.cdc.bootstrap.servers": "your_kafka_server1:9092,your_kafka_server2:9092,your_kafka_server3:9092",
    "indices.cdc.max.request.size": 1048576,
    "indices.cdc.max.in.flight.requests.per.connection": 10,
    "indices.cdc.retry.backoff.ms": 100,
    "indices.cdc.retries": 100,
    "indices.cdc.max.block.ms": 86400000
  }
}

创建索引时，enable cdc并初始化相关属性

PUT /index
{
  "settings": {
    "number_of_shards": 2,
    "number_of_replicas": 1,
    "index.cdc.enabled": true,
    "index.cdc.topic": "cdc_test",
    "index.cdc.pk.column": "doc_id",
    "index.refresh_interval": "100s"
  },
  "mappings": {
    "properties": {
      "content": {
        "type": "text"
      },
      "doc_id": {
        "type": "integer"
      },
      "age": {
        "type": "integer"
      },
      "address": {
        "type": "keyword"
      },
      "keywords": {
        "type": "nested",
        "properties": {
          "keyword": {
            "type": "keyword"
          },
          "frequency": {
            "type": "integer"
          }
        }
      }
    }
  }
}

索引级别属性说明

属性值 默认值 说明
index.cdc.enabled false 是否启用cdc 监听
index.cdc.topic "" 指定cdc数据写入的topic 名称
index.cdc.pk.column "" 主键列
index.cdc.exclude.columns "" 排除列，多列使用逗号分割
index.cdc.alias "" index的别名，当设置时，cdc 会使用该别名

集群级别配置属性

属性值 默认值 说明
indices.cdc.bootstrap.servers "" 同 producer bootstrap.servers
indices.cdc.batch.size 16384 同 producer batch.size
indices.cdc.acks all 同 producer acks
indices.cdc.buffer.memory 32 * 1024 * 1024 同 producer buffer.memory
indices.cdc.compression.type none 同 producer compression.type
indices.cdc.request.timeout.ms 30 * 1000 同 producer request.timeout.ms
indices.cdc.max.request.size 1024 * 1024 同 producer max.request.size
indices.cdc.retries 0 同 producer retries
indices.cdc.retry.backoff.ms 100 同 producer retry.backoff.ms
indices.cdc.send.buffer.bytes 131072 同 producer send.buffer.bytes
indices.cdc.receive.buffer.bytes 32768 同 producer receive.buffer.bytes
indices.cdc.linger.ms 0 同 producer linger.ms
indices.cdc.max.in.flight.requests.per.connection 5 同 producer max.in.flight.requests.per.connection
indices.cdc.max.block.ms 0 同 producer max.block.ms
indices.cdc.producer.nums es server 的core数 每一个节点kafka producer的线程数

注意

action 值：delete - 0，insert - 1，update - 2

测试

操作 请求数据 kafka key kafka value
insert with POST POST /index/_create/1
{"doc_id":1,"content":"peace and hope","age":12,"address":"bj","keywords":[{"keyword":"peace","frequency":10},{"keyword":"hope","frequency":5}]}		 1 {"op":1,"index":"index","content":{"address":"bj","keywords":[{"keyword":"peace","frequency":10},{"keyword":"hope","frequency":5}],"doc_id":1,"content":"peace and hope","age":12},"ts":1629530725380}
insert with PUT PUT /index/_doc/2
{"doc_id": 2,"content":"peace and hope 2", "age": 12, "address": "bj", "keywords": [{"keyword": "peace", "frequency": 1}, {"keyword": "hope", "frequency": 1}]}		 2 {"op":1,"index":"index","content":{"address":"bj","keywords":[{"keyword":"peace","frequency":1},{"keyword":"hope","frequency":1}],"doc_id":2,"content":"peace and hope 2","age":12},"ts":1629531058056}
bulk insert PUT /_bulk
{"index":{"_index":"index","_id":"3"}}
{"doc_id": 3,"content":"peace and hope 3", "age": 13, "address": "bj", "keywords": [{"keyword": "peace", "frequency": 1}, {"keyword": "hope", "frequency": 1}]}
{"index":{"_index":"index","_id":"4"}}
{"doc_id": 4,"content":"peace and hope 4", "age": 14, "address": "bj", "keywords": [{"keyword": "peace", "frequency": 1}, {"keyword": "hope", "frequency": 1}]}
{"index":{"_index":"index","_id":"5"}}
{"doc_id": 5,"content":"peace and hope 5", "age": 15, "address": "bj", "keywords": [{"keyword": "peace", "frequency": 1}, {"keyword": "hope", "frequency": 1}]}		 三条记录key分别是5,4,3 与key对应的value分别是：
{"op":1,"index":"index","content":{"address":"bj","keywords":[{"keyword":"peace","frequency":1},{"keyword":"hope","frequency":1}],"doc_id":5,"content":"peace and hope 5","age":15},"ts":1629531206042}
{"op":1,"index":"index","content":{"address":"bj","keywords":[{"keyword":"peace","frequency":1},{"keyword":"hope","frequency":1}],"doc_id":4,"content":"peace and hope 4","age":14},"ts":1629531205887}
{"op":1,"index":"index","content":{"address":"bj","keywords":[{"keyword":"peace","frequency":1},{"keyword":"hope","frequency":1}],"doc_id":3,"content":"peace and hope 3","age":13},"ts":1629531205885}
simple update POST /index/_update/1
{"doc":{"doc_id":1,"address":"sjz"}}		 1 {"op":2,"index":"index","content":{"address":"sjz","keywords":[{"keyword":"peace","frequency":10},{"keyword":"hope","frequency":5}],"doc_id":1,"content":"peace and hope","age":12},"ts":1629531453903}
update by query and just update non-nested field POST /index/_update_by_query
{"script":{"source":"ctx._source.age++","lang":"painless"},"query":{"term":{"doc_id":1}}}		 1 {"op":2,"index":"index","content":{"address":"sjz","keywords":[{"keyword":"peace","frequency":10},{"keyword":"hope","frequency":5}],"doc_id":1,"content":"peace and hope","age":13},"ts":1629531523775}
update by query --update nested field POST /index/_update/1
{"script":{"source":"for(e in ctx._source.keywords){if (e.frequency == 5) {e.frequency = 10;}}"}}		 1 {"op":2,"index":"index","content":{"address":"sjz","keywords":[{"keyword":"peace","frequency":10},{"keyword":"hope","frequency":10}],"doc_id":1,"content":"peace and hope","age":13},"ts":1629531711169}
update by query -- delete nested field POST /index/_update/1
{"script":{"source":"ctx._source.keywords.removeIf(it -> it.frequency == 10);"}}		 1 {"op":2,"index":"index","content":{"address":"sjz","keywords":[],"doc_id":1,"content":"peace and hope","age":13},"ts":1629531763492}
simple delete DELETE /index/_doc/1
 1 {"op":0,"index":"index","content":{"doc_id":"1"},"ts":1629531859497}
delete by query POST /index/_delete_by_query
{"query":{"match":{"doc_id":2}}}		 2 {"op":0,"index":"index","content":{"doc_id":"2"},"ts":1629531944820}

变更cdc 配置 测试

PUT index/_settings
{
  "index.cdc.enabled":false
}

PUT index/_settings
{
  "index.cdc.enabled":true,
  "index.cdc.exclude.columns": "age,content"
}

POST /index/_create/6
{"doc_id": 6,"content":"peace and hope", "age": 12, "address": "bj", "keywords": [{"keyword": "peace", "frequency": 10}, {"keyword": "hope", "frequency": 5}]}


生成的 cdc 消息，在cdc_test中有如下消息：
{"op":1,"index":"index","content":{"address":"bj","keywords":[{"keyword":"peace","frequency":10},{"keyword":"hope","frequency":5}],"doc_id":6},"ts":1629532594468}
已经排除了 content 和 age 的信息