elasticsearchreader

1 快速介绍

Elasticsearch Scroll API 是一项突破性的技术解决方案，专为高效处理大数据集而设计。

相较于传统的分页查询机制，Scroll API 能够克服其在数据量庞大时性能下降、资源消耗增大的局限性。

在DataX数据迁移场景中，ElasticsearchReader巧妙地运用了Scroll API的全量检索能力，通过一次性设定恰当的查询语句，能够无缝遍历并抽取指定索引下的所有数据记录。

在此过程中，ElasticsearchReader会将获取到的海量原始数据实时且高效地转化为符合DataX传输规范的数据格式，进而确保Writer组件能够流畅接收与处理这些数据，实现大规模数据在不同系统间的快速迁移和同步。

2 功能与限制

基于Elasticsearch的Scroll API，ElasticsearchReader实现了以下功能。

得益于Elasticsearch的Scroll API，ElasticsearchReader高效实现了以下核心功能：

1.全量数据精确抓取：通过精心设计的query语句，ElasticsearchReader利用Scroll API的独特优势，能够一次性从指定索引中检索并获取全部相关数据，从而克服了传统分页查询在处理大量数据时的性能瓶颈。

3.高效数据流处理：ElasticsearchReader借助Scroll API的持久化游标机制，实现了一种持续、流畅的数据流读取方式。这不仅减少了网络开销，还确保了在处理大量数据时系统的稳定性和响应速度。

4.资源管理优化：通过 Scroll API 的上下文管理，ElasticsearchReader能够有效控制和清理检索过程中的临时资源，避免了因长时间执行查询而可能导致的系统资源浪费或堆积问题。

5.与DataX生态无缝集成：将从Elasticsearch获取到的全量数据转换为DataX传输协议格式后，ElasticsearchReader可直接与DataX生态下的Writer组件对接，实现了跨系统的高效数据迁移和同步任务。

我们暂时不能做到：

1. 增量查询数据。原因：
   在Elasticsearch中，当使用Scroll API进行深度分页或批量检索数据时，它所基于的“快照”并非传统意义上的数据库备份快照，而是指Scroll API在首次执行搜索请求时获取的一次性索引视图。

2. 考虑到稳定性，elasticsearchreader采用单线程工作方式。

3. 考虑到配置大于约定的原则，不动态mapping映射数据结构，最大的基于配置灵活性。

这个视图包含当前搜索请求定义下所有匹配文档的一个静态集合。

具体来说：

• a. 初始化Scroll：当你发起一个带有scroll参数的搜索请求时，Elasticsearch会冻结当前索引的状态，并为此次搜索创建一个上下文（context）。

• b. 快照视图：这个上下文包含了满足搜索条件的所有文档的一个瞬时视图，可以理解为对索引状态的一个“快照”。这意味着在这个快照生成后，即使索引中的文档发生增删改等变化，Scroll API后续返回的结果也不会受到这些变更的影响。

• c. 滚动查询：在首次搜索之后，你可以通过发送带有先前响应中返回的scroll ID的请求来继续滚动查询。每次滚动都会从快照视图中返回下一批文档，直到所有文档都被遍历完毕或者scroll超时。

因此，对于Scroll API而言，“快照”意味着一次搜索请求结果集的固定版本，它不支持增量更新和实时变化的数据同步。这种设计允许用户在一段时间内处理大量数据而不用担心在此期间索引数据的变动问题。

3 功能说明

全量查询Elasticsearch某索引的所有数据,适合全量同步数据，适用于大量数据的迁移场景。

3.1 配置样例

{
  "setting": {},
  "job": {
    "setting": {
      "speed": {
        "channel": 2
      }
    },
    "content": [
      {
        "reader": {
          "name": "elasticsearchreader",
          "parameter": {
            "hosts": "http://127.0.0.1:9200",
            "username": "",
            "password": "",
            "connection_timeout": "10000",
            "socket_timeout": "60000",
            "io_thread_count": "5",
            "index": "abc-*",
            "type": "_doc",
            "scroll": "10m",
            "size": 100,
            "dsl": {
              "query": {
                "term": {
                  "field1": {
                    "value": "1"
                  }
                }
              },
              "_source": [
                "field1"
              ]
            },
            "column": [
              {
                "field": "field1",
                "type": "string",
                "defaultValue": "",
                "dataFormat": ""
              }
            ]
          }
        },
        "writer": {
          "name": "streamwriter",
          "parameter": {
            "encoding": "UTF-8",
            "print": true
          }
        }
      }
    ]
  }
}

3.2 参数说明

• hosts

• 描述：Elasticsearch服务的访问地址列表。多个地址之间使用英文逗号分隔。

• 必选：是

• 默认值：无

• username

• 描述：用于连接Elasticsearch集群的HTTP基本认证用户名。如果集群启用了身份验证，则需要提供此信息。

• 必选：视情况而定（取决于集群是否要求身份验证）

• 默认值：空字符串

• password

• 描述：与username对应的HTTP基本认证密码。

• 必选：视情况而定（同上，取决于集群是否要求密码认证）

• 默认值：空字符串

• connection_timeout

• 描述：客户端尝试连接到Elasticsearch服务器的超时时间（单位：毫秒）。

• 必选：否

• 默认值：10000毫秒（即10秒）

• socket_timeout

• 描述：客户端与Elasticsearch服务器之间的网络读写操作超时时间（单位：毫秒）。

• 必选：否

• 默认值：60000毫秒（即1分钟）

• io_thread_count

• 描述：用于处理与Elasticsearch服务器网络I/O操作的线程数。

• 必选：否

• 默认值：5个线程

• index

• 描述：满足elasticsearch索引检索规则。

  • 单个索引名称： 指定要在其中搜索的单个索引名称。例如：”index”: “my_index”。

  • 逗号分隔的多个索引名称： 可以指定逗号分隔的多个索引名称，以在这些索引中执行搜索。例如：”index”: “index1,index2”。

  • 通配符表达式： 可以使用通配符表达式来匹配多个索引名称。常用的通配符包括和?。例如：”index”: “prefix“或”index”: “prefix?,suffix*”。

  • 索引模式匹配： 除了通配符表达式外，还可以使用索引模式来匹配索引名称。例如，使用<index-{now/d}>
    这样的模式可以匹配形如index-2024-02-26的索引名称，其中{now/d}是一个时间格式的占位符，表示当天的日期。

  • 索引别名： 除了直接指定索引名称外，还可以指定索引别名，搜索将在该别名所关联的索引上执行。例如：”index”: “my_alias”。

• 必选：是（在执行索引相关操作时）

• 默认值：无

• type

• 描述：在旧版本Elasticsearch中代表文档类型的字段，在7.x及以上版本已弃用，默认为”_doc”。

• 必选：视情况而定（对于兼容旧版Elasticsearch的情况）

• 默认值：”_doc”

• scroll

• 描述：Scroll API的保持存活时间，在这个时间段内可以持续滚动获取更多数据（单位：分钟）。

• 必选：当使用Scroll API时必需

• 默认值：根据设置（此处设定为 “10m” 即10分钟）

• size

• 描述：每次Scroll请求返回的结果集大小，即一次获取多少条文档记录。

• 必选：在使用Scroll API时通常需要设置

• 默认值：根据上下文需求决定（此处设定了 “100” 条记录）

• dsl

• 描述：包含Elasticsearch查询DSL（Domain Specific Language）的JSON对象，用于定义复杂的搜索和过滤条件。

• 必选：当需要执行特定查询时必需,并且包含query和_source。

• 默认值：无（此处给出了一个示例查询）

• column

• 描述：字段映射

{
  "field": "field1",
  "type": "int/long/double/date/bool/string",
  "defaultValue": "默认值",
  "dataFormat": "时间格式"
}

type类型支持int,long,double,date,bool,string,大小写不敏感。是data类型的时候需要指定dataFormat.

Java中常见的日期格式主要包括以下几种：

• 完整日期时间（包括年月日时分秒）：

yyyy-MM-dd HH:mm:ss：表示四位数的年、两位数的月和日，24小时制的小时、分钟和秒。例如：2024-02-26 14:30:45。

yyyy-MM-dd hh:mm:ss a 或 yyyy-MM-dd HH:mm:ss a：与上面类似，但添加了AM/PM指示符区分上午和下午，其中hh是12小时制小时数。

• 仅日期部分：

yyyy-MM-dd：只包含年月日信息，不包含时间。例如：2024-02-26。

• 仅时间部分：

HH:mm:ss：24小时制的时间，不包含日期。例如：14:30:45。

hh:mm:ss a：12小时制的时间，并带有AM/PM标记。

• 精确到毫秒：

yyyy-MM-dd HH:mm:ss.SSS：表示日期时间至毫秒级别，例如：2024-02-26 14:30:45.123。

Java 8中的Java.time包支持更精细的时间格式：

yyyy-MM-dd HH:mm:ss.SSSSSS：使用java.time.LocalDateTime和DateTimeFormatter可以达到微秒级别的格式化，但实际上Java获取的是系统当前时间的毫秒数，微秒部分补零。

• 必选：和_soured字段映射

• 默认值：无

3.3 类型转换

Datax的数据结构包括：BAD, NULL, INT, LONG, DOUBLE, STRING, BOOL, DATE, BYTES。
Elasticsearch Mapping 类型及其可能的DataX映射：

1. 文本（Text）

   • 对应DataX：字符串类型（String）

2. 关键字（Keyword）

   • 对应DataX：字符串类型（String），适用于索引和排序的精确值场景。

3. 整数（Integer）

   • byte, short, integer, long
   • 对应DataX：整数类型（LONG）

4. 浮点数（Floating Point）

   • float, double
   • 对应DataX：浮点数类型（Double）

5. 日期（Date）

   • 对应DataX：日期类型（Date）

6. 布尔（Boolean）

   • 对应DataX：布尔类型（Boolean）

7. 二进制（Binary）

   • 对应DataX：通常通过Base64编码转换为字符串传输，因此在DataX中可以表现为字符串类型（String）

8. 数组（Array）

   • 对应DataX：字符串类型（String）

9. 地理位置（Geo-point）

   • 对应DataX：可以转换为经纬度坐标对字符串类型（String）

10. 嵌套对象（Nested Object）

    • 对应DataX：复杂类型，通常需要转换为JSON字符串或其他可序列化的结构，在DataX中用字符串类型（String）

      插件地址

      链接：https://pan.baidu.com/s/1nF1SBOwlBBvQ-UF7_EWHww
      提取码：pgwk