ElasticSearch河JDBC MySQL不删除记录

Question

我正在使用 ElasticSearch 的 JDBC 插件来更新我的 MySQL 数据库。它拾取新的和更改的记录，但不会删除已从 MySQL 中删除的记录。它们保留在索引中。

这是我用来创建河流的代码：

curl -XPUT 'localhost:9200/_river/account_river/_meta' -d '{
    "type" : "jdbc",
    "jdbc" : {
        "driver" : "com.mysql.jdbc.Driver",
        "url" : "jdbc:mysql://localhost:3306/test",
        "user" : "test_user",
        "password" : "test_pass",
        "sql" : "SELECT `account`.`id` as `_id`, `account`.`id`, `account`.`reference`, `account`.`company_name`, `account`.`also_known_as` from `account` WHERE NOT `account`.`deleted`",
        "strategy" : "simple",
        "poll" : "5s",
        "versioning" : true,
        "digesting" : false,
        "autocommit" : true,
        "index" : "headphones",
        "type" : "Account"
    }
}'

在 OSX Mountain Lion 上通过 homebrew 安装了 ElasticSearch，没有错误或问题，一切都按预期响应。权限正常，日志中没有错误。

我已删除并包含（并设置为 true 和 false）我能想到的 autocommit、versioning 和 digesting 的每个组合。这是一个开发数据库，​​所以我确信记录被完全删除，没有缓存也没有软删除。如果我删除所有记录（即保持河流完好无损，只删除在 ES 上索引的内容），下次河流更新时不会重新添加记录，这让我相信我错过了有关版本控制和删除的一些内容.

请注意，我还尝试了各种方法来指定 _id 列，并通过 JSON on call 检查了它是否具有值。

干杯。

Answer 1

自从问了这个问题后，参数发生了很大的变化，不推荐使用versioning和digesting，并且poll已经被schedule取代，这将采取cron表达式关于多久重新运行一次river（下面是scheduled to run每 5 分钟）

    curl -XPUT 'localhost:9200/_river/account_river/_meta' -d '{
        "type" : "jdbc",
        "jdbc" : {
            "driver" : "com.mysql.jdbc.Driver",
            "url" : "jdbc:mysql://localhost:3306/test",
            "user" : "test_user",
            "password" : "test_pass",
            "sql" : "SELECT `account`.`id` as `_id`, `account`.`id`, `account`.`reference`, `account`.`company_name`, `account`.`also_known_as` from `account` WHERE NOT `account`.`deleted`",
            "strategy" : "simple",
            "schedule": "0 0/5 * * * ?" ,
            "autocommit" : true,
            "index" : "headphones",
            "type" : "Account"
        }
    }'

但是对于主要问题，我从开发人员那里得到的答案是这样的 https://github.com/jprante/elasticsearch-river-jdbc/issues/213

不再检测到删除行。

我尝试使用版本控制进行内务管理，但效果不佳 以及增量更新和添加行。

一个好的方法是窗口索引。每个时间范围（可能一次 每天或每周）为河流创建一个新索引，并添加 给一个别名。一段时间后将删除旧索引。这 维护类似于logstash索引，但它不在 河流的范围。

我目前用作研究别名的方法是每晚重新创建索引和河流，并安排河流每隔几个小时运行一次。它确保将在当天为输入的新数据编制索引，并且每 24 小时反映一次删除情况

Answer 2

我对弹性还是比较陌生，并且一直在我的项目中使用 jdbc River。 如果我理解正确，不一定是这种情况，这就是它的工作原理：

1. 从以下位置获取所有行（由 River 中的 SQL 语句指定） 数据库。
2. 从所有提取行的（id、type 和 index）计算摘要（如果添加了新行或删除了行，这应该 改变）。
3. 为所有行重新索引文档。这会自动 增加每个文档的版本。
4. 存储在 _river 索引中的河流的增量版本（自定义）
5. 如果 #3 中计算的摘要不同于 然后存储在 _river 索引中：
   • 储存起来
   • 运行管家功能（删除所有版本号较低的文档）。

因此，考虑到您希望运行内务管理，您需要将版本控制设置为 true，随后这意味着 digesting 也应设置为 true。

话虽如此，你的河流应该是这样的：

curl -XPUT 'localhost:9200/_river/account_river/_meta' -d '{
    "type" : "jdbc",
    "jdbc" : {
        "driver" : "com.mysql.jdbc.Driver",
        "url" : "jdbc:mysql://localhost:3306/test",
        "user" : "test_user",
        "password" : "test_pass",
        "sql" : "SELECT `account`.`id` as `_id`, `account`.`id`, `account`.`reference`, `account`.`company_name`, `account`.`also_known_as` from `account` WHERE NOT `account`.`deleted`",
        "strategy" : "simple",
        "poll" : "5s",
        "autocommit" : true,
        "index": {
          "index" : "headphones",
          "type" : "Account",
          "versioning" : true,
          "digesting" : true
        }
    }
}'

注意versioning 和digesting 应该是index 定义的一部分，而不是jdbc 定义的一部分