es过滤数据归档的delete

需要自己写一个plugin来过滤，默认不支持

原回答者GitHub用户agapple

根据您的描述，您正在使用 Canal ETL 将 MySQL 数据库中的数据同步到 Elasticsearch 中，并计划定期归档数据库中的历史数据。但是，您希望在归档数据库中的数据时，不会影响已经存储在 Elasticsearch 中的全部数据，并且过滤掉一些 delete 操作。针对这个问题，您可以考虑以下几个方案：

1、使用 Elasticsearch 的索引别名

您可以使用 Elasticsearch 的索引别名功能来实现定期归档历史数据的需求。具体来说，您可以创建一个主索引和一个别名，然后将别名指向主索引。在定期归档历史数据时，您可以创建一个新的索引，将数据导入该索引中，并将别名指向新的索引，这样就可以保留全部数据，并防止新数据和历史数据冲突。

例如，您可以按照以下步骤创建索引别名：

创建主索引： PUT /my_index { "settings": { "number_of_shards": 3, "number_of_replicas": 2 }, "mappings": { "properties": { "id": {"type": "long"}, "name": {"type": "text"} } } }

创建别名：

POST /_aliases { "actions" : [ { "add" : { "index" : "my_index", "alias" : "my_alias" } } ] }

查询别名：

GET /_cat/aliases/my_alias

修改别名：

POST /_aliases { "actions" : [ { "remove" : { "index" : "my_index", "alias" : "my_alias" } }, { "add" : { "index" : "new_index", "alias" : "my_alias" } } ] } 在上述示例中，主索引名是 my_index，别名是 my_alias。您可以根据实际情况修改索引和别名的名称。当您需要归档历史数据时，可以创建一个新的索引，例如 new_index，将数据导入该索引中，并将别名指向新的索引，这样就可以保留全部数据，并防止新数据和历史数据冲突。

2、在 Canal ETL 配置中过滤 delete 操作

您可以在 Canal ETL 配置文件中添加过滤规则，以过滤掉 delete 操作。具体来说，您可以使用如下语句过滤 delete 操作：

{ "type": "es", "url": "http://localhost:9200", "index": "my_index", "params": { "es.mapping.id": "id", "es.index.auto.create": "true", "es.action": "upsert", "es.upsert.script": "ctx.op = 'none'", "es.retry": "3", "es.bulk.flush.interval": "1000", "es.bulk.flush.max.actions": "5000", "es.filter.regex": "^((?!delete).)$" } } 在上述配置文件中，您可以看到 es.filter.regex 参数，它指定了一个正则表达式，用于过滤 delete 操作。具体来说，正则表达式 "^((?!delete).)$" 表示过滤掉任何包含 delete 子串的操作。

需要注意的是，使用 Canal ETL 进行数据同步时，如果数据源是 MySQL 数据库，可以在 Canal ETL 配置文件中添加 binlog.filter.regex 参数，以实现对操作的过滤。例如，您可以使用如下语句过滤 delete 操作：

{ "type": "mysql", "url": "jdbc:mysql://localhost:3306/mydb", "user": "root", "password": "123456", "binlog": { "name": "mysql-bin.000001", "position": 4, "filter": { "name": "", "regex": "^((?!delete).)*$", "black": false } } }