企业官网建设流程全解析

SeaTunnel Elasticsearch Source 连接器实战指南：多索引同步、认证、SCROLL/PIT 分页与运行时字段

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel

SeaTunnel 的 Elasticsearch Source 连接器（插件名Elasticsearch）用于从 Elasticsearch 集群批量读取数据，支持 ES 2.x 到 8.x 之间各主要版本，可作为批处理管道的入口将索引数据同步到任意目标端。读完本文你将掌握该连接器的全部配置参数、三种查询方式（DSL / SQL）、两种分页 API（SCROLL / PIT）的选型原则，以及多索引并行同步、切片加速、TLS 认证与运行时字段等生产级用法。

简介与能力边界

Elasticsearch Source 连接器在 connector-elasticsearch 模块 中实现，是一个有界（BOUNDED）批处理数据源。从源码看，ElasticsearchSource.java 实现了SeaTunnelSource、SupportParallelism、SupportColumnProjection三个核心接口，这与下表所列特性一一对应。

版本兼容说明：连接器支持读取 Elasticsearch 2.x 至 8.x 之间各版本的数据。个别高级能力对版本有硬性要求，例如 SCROLL 切片（sliced scroll）要求 ES ≥ 5.0，PIT 要求 ES ≥ 7.10，runtime_fields要求 ES ≥ 7.11，详见下文对应小节。

读取流程：从 Split 枚举到数据行输出

理解底层执行模型有助于正确配置并行度和切片参数。该连接器严格遵循 SeaTunnel 的 Source/Split 模型，核心链路如下：

1. ElasticsearchSource 解析配置：读取index或index_list生成ElasticsearchConfig列表；如果配置了source字段会调用EsRestClient.getFieldTypeMapping拉取索引映射（mapping）推导 SeaTunnel 字段类型；若为 SQL 模式则调用getSqlMapping从 SQL 结果推导类型。
2. ElasticsearchSourceSplitEnumerator 枚举 Split：通过getIndexDocsCount获取每个索引的文档数，过滤掉空索引后，为每个(索引, 切片)组合生成一个 Split（slice_max > 1时 splitId 形如index#sliceId），再按assignCount % readerCount轮询（round-robin）分配给各并行子任务。
3. ElasticsearchSourceReader 消费数据：按search_type分支执行——SQL 走 X-Pack SQL 游标分页；DSL 模式再按search_api_type决定走 Scroll API 还是 PIT API，逐批拉取并用DefaultSeaTunnelRowDeserializer反序列化为SeaTunnelRow交给下游。

由于getBoundedness()返回BOUNDED，读取完毕会发送signalNoMoreElement结束任务，因此该连接器天然适合一次性全量/增量窗口同步，不适合持续监听写入。

配置参数总览

所有参数均在 ElasticsearchSourceOptions.java 与 ElasticsearchBaseOptions.java 中声明（含默认值），汇总如下：

基础连接与认证

hosts [array]（必填）

Elasticsearch 集群的 HTTP 地址数组，格式为host:port，支持一次配置多个节点实现负载均衡与故障转移。例如["host1:9200", "host2:9200"]。从 EsRestClient.java 的实现看，hosts 会被逐个解析为HttpHost构建底层RestClient，并为每个请求设置 10 秒连接请求超时与 5 分钟 Socket 超时。

auth_type [enum]

指定认证方式，由 AuthenticationProviderFactory 根据配置分发到对应 Provider。支持：

• basic（默认）：用户名 + 密码的 HTTP 基本认证（对应 x-pack 安全）；
• api_key：API Key 的 ID + 密钥认证；
• api_key_encoded：Base64 编码后的 API Key 认证。

未指定时默认使用basic以兼容旧版本配置。

基本认证（basic）

source { Elasticsearch { hosts = ["https://localhost:9200"] auth_type = "basic" username = "elastic" password = "your_password" index = "my_index" } }

API Key 认证

注意：auth.api_key_id+auth.api_key与auth.api_key_encoded只能二选一，同时配置时工厂校验会将其视为非法组合。

示例（分开配置 ID 和 key）：

source { Elasticsearch { hosts = ["https://localhost:9200"] auth_type = "api_key" auth.api_key_id = "your_api_key_id" auth.api_key = "your_api_key_secret" index = "my_index" } }

示例（使用编码 key）：

source { Elasticsearch { hosts = ["https://localhost:9200"] auth_type = "api_key_encoded" auth.api_key_encoded = "eW91cl9hcGlfa2V5X2lkOnlvdXJfYXBpX2tleV9zZWNyZXQ=" index = "my_index" } }

索引与字段投影配置

index [string]

Elasticsearch 索引名称，支持*通配符。例如存在索引index1、index2，可指定index*同时读取两个索引的数据。index与index_list至少配置一个：

• 单索引或索引通配场景使用index；
• 不同索引需要单独配置query、source、schema或分页参数时，使用index_list。

行为提示：从 ElasticsearchSource.java 的构造逻辑看，若index与index_list同时出现，会打印告警并只让index_list生效。

source [array]

要读取的索引字段列表，即列投影。你可以通过指定字段_id来获取文档 ID；如果要将_id写入其他索引，由于 Elasticsearch 的限制，需要为_id指定一个别名。如果未配置source，连接器会通过EsRestClient.getFieldTypeMapping自动从索引映射中获取全部字段及其类型。字段顺序即输出 SeaTunnel Row 的字段顺序。

array_column [map]

由于 Elasticsearch 中没有数组类型，映射推导无法识别数组字段，因此需要通过该参数显式声明。假设tags和phones是数组字段：

array_column = {tags = "array<string>", phones = "array<string>"}

在 ElasticsearchSource.java 中，array_column中命中的字段会直接按其声明的 SeaTunnel 类型（如array<string>、array<tinyint>）构建物理列，其余字段才走 ES 类型到 SeaTunnel 类型的自动转换（ElasticSearchTypeConverter）。

schema（已废弃）

早期版本使用schema显式声明字段类型，但源码中已明确打印警告：The schema config in ElasticSearch source/sink is deprecated, please use source config instead!。当前推荐做法是只配置source（字段列表）配合可选的array_column，让连接器根据索引映射自动推导类型。

查询与分页

query [json]

Elasticsearch 原生查询语句（DSL），用于控制读取哪些文档。不配置时默认值为{"match_all": {}}，即全量读取。例如：

{"range":{"firstPacket":{"gte":1669225429990,"lte":1669225429990}}}

search_type [enum] 与 sql_query [string]

• DSL（默认）：使用 Elasticsearch Query DSL 查询；
• SQL：使用 X-Pack SQL 查询，此时必须配置sql_query。

SQL 模式通过EsRestClient.searchBySql发起/_sql请求，并基于服务端返回的 cursor 游标持续翻页，结束后调用closeSqlCursor释放资源。

source { Elasticsearch { hosts = ["https://elasticsearch:9200"] username = "elastic" password = "elasticsearch" tls_verify_certificate = false tls_verify_hostname = false index = "st_index_sql" sql_query = "select * from st_index_sql where c_int>=10 and c_int<=20" search_type = "sql" } }

注意：SQL 查询不支持 map 和 array 类型字段；同时 SQL 查询不支持切片，slice_max配置会被忽略（源码中会打印告警并强制置回 1）。

scroll_time [string] 与 scroll_size [int]

底层使用滚动查询（Scroll API）拉取数据，因此需要：

• scroll_time：控制搜索上下文（search context）在 Elasticsearch 侧存活的时长，默认1m（1 分钟）。若任务处理较慢导致上下文过期，会触发SearchContextMissingException，此时应适当调大。
• scroll_size：每次滚动请求返回的最大文档数，默认100。

读取完成后连接器会调用clearScroll主动清理 scroll ID。

search_api_type [enum]：SCROLL 与 PIT 对比

• pit_keep_alive[long]：PIT 保持活动的时间量，单位毫秒，默认60000（1 分钟）。
• pit_batch_size[int]：每次 PIT 搜索请求返回的最大文档数，默认100，语义同scroll_size。

PIT 模式下 ElasticsearchSourceReader 会先创建 PIT，随后用search_after游标持续翻页直至hasMore为 false，最后在finally中删除 PIT 释放资源。

slice_max [int]：切片并行读取

将单个索引拆分为多个切片（slice）并行读取，仅对 SCROLL/PIT 生效，配置 > 1 时启用。

版本要求：

• SCROLL 切片（sliced scroll）需要 Elasticsearch 5.0 及以上版本；
• PIT 切片需要 Elasticsearch 7.10 及以上版本（PIT 于 7.10.0 引入）。

取舍说明：切片能显著提升吞吐，但可能降低跨切片的数据一致性。对一致性要求高的场景，建议使用 PIT（共享快照）或将slice_max = 1；对追加写或写入较少的场景，开启切片通常可以接受。

当search_type = "SQL"时，Elasticsearch SQL 查询不支持切片，slice_max会被忽略。

与并行度的配合：从 ElasticsearchSourceSplitEnumerator.java 的枚举逻辑可见，slice_max决定每个索引生成的 Split 数量（sliceId从 0 到sliceMax-1），而parallelism（通用选项）决定参与消费的 Reader 子任务数；所有 Split 会按轮询方式均匀分配到各 Reader。因此提升吞吐通常需要同时调大这两者。

runtime_fields：查询期动态计算字段

runtime_fields允许在查询时动态计算字段值，而无需重建索引，适合临时分析、字段试验与低频查询（Elasticsearch 7.11+）。每个 runtime field 需要包含：

• name：字段名
• type：数据类型，支持boolean、date、double、geo_point、ip、keyword、long
• script：Painless 脚本，用于计算字段值
• script_lang（可选）：脚本语言，默认painless
• script_params（可选）：脚本参数

在源码 ElasticsearchSource.java 的parseRuntimeFields方法中，该配置会被转换为 Elasticsearch 的runtime_mappings结构（{字段名: {type, script: {source, lang?, params?}}}）随查询请求一并下发。

runtime_fields = [ { name = "day_of_week" type = "keyword" script = "emit(doc['timestamp'].value.dayOfWeekEnum.toString())" }, { name = "total_price" type = "double" script = "emit(doc['quantity'].value * doc['price'].value)" } ]

性能与限制：

• 运行时字段在查询阶段计算，数据量大时会影响查询性能；
• 适合临时分析、字段试验与低频查询场景；
• 需要 Elasticsearch 7.11 及以上版本；
• 计算出的字段需加入source列表并（建议）在schema中声明类型后才可被下游使用。

TLS / SSL 加密连接配置

连接器基于 ElasticsearchBaseOptions.java 提供完整的 HTTPS 支持：

典型使用方式是 hosts 使用https://前缀并配合相应证书策略，完整示例见下文「使用案例」的案例三至案例五。

多索引同步（index_list）

index_list用于定义多索引同步任务。它是一个数组，每个元素包含单表同步所需的完整参数，如index、query、source/schema、scroll_size和scroll_time等。建议不要将index_list和query配置在同一层级，即query应放在index_list的每个条目内。

注意（源码注释披露的限制）：index_list内部每个条目的配置不会经过工厂的OptionRule校验（该校验只作用于顶层配置）。如果某个条目缺少index，或设置为search_type = SQL却缺少sql_query，会在运行时（解析阶段）才报错。因此多索引配置时务必逐个核对条目完整性。

common options（通用选项）

Source 插件常用参数，完整说明见 Source 常用选项：

重要提示：作业中使用plugin_output时，下游插件必须设置plugin_input才能消费该数据集。

使用案例

案例一：通配索引 + 字段投影 + 数组字段 + 范围查询

从满足seatunnel-*匹配的索引中按 query 读取数据，查询只返回文档的id、name、age、tags、phones字段；其中tags、phones通过array_column声明为数组类型，_id用于取回文档 ID。

Elasticsearch { hosts = ["localhost:9200"] index = "seatunnel-*" array_column = {tags = "array<string>",phones = "array<string>"} source = ["_id","name","age","tags","phones"] query = {"range":{"firstPacket":{"gte":1669225429990,"lte":1669225429990}}} }

案例二：多索引同步（index_list）

演示从read_index1和read_index2读取不同的数据。read_index1使用source指定字段、array_column声明数组字段并带 range 条件；read_index2使用match_all全量读取另一组字段。两个来源经同一管道由 Elasticsearch Sink 统一写出（示例 sink 目标索引为multi_source_write_test_index，index_type = "st"，采用CREATE_SCHEMA_WHEN_NOT_EXIST+APPEND_DATA的保存模式）。

source { Elasticsearch { hosts = ["https://elasticsearch:9200"] username = "elastic" password = "elasticsearch" tls_verify_certificate = false tls_verify_hostname = false index_list = [ { index = "read_index1" query = {"range": {"c_int": {"gte": 10, "lte": 20}}} source = [ c_map, c_array, c_string, c_boolean, c_tinyint, c_smallint, c_bigint, c_float, c_double, c_decimal, c_bytes, c_int, c_date, c_timestamp ] array_column = { c_array = "array<tinyint>" } } { index = "read_index2" query = {"match_all": {}} source = [ c_int2, c_date2, c_null ] } ] } } transform { } sink { Elasticsearch { hosts = ["https://elasticsearch:9200"] username = "elastic" password = "elasticsearch" tls_verify_certificate = false tls_verify_hostname = false index = "multi_source_write_test_index" index_type = "st" "schema_save_mode"="CREATE_SCHEMA_WHEN_NOT_EXIST" "data_save_mode"="APPEND_DATA" } }

案例三：SSL（禁用证书验证）

自签名证书环境下跳过证书链校验（不推荐用于生产）：

source { Elasticsearch { hosts = ["https://localhost:9200"] username = "elastic" password = "elasticsearch" tls_verify_certificate = false } }

案例四：SSL（禁用主机名验证）

证书合法但主机名不匹配时使用：

source { Elasticsearch { hosts = ["https://localhost:9200"] username = "elastic" password = "elasticsearch" tls_verify_hostname = false } }

案例五：SSL（启用证书验证）

通过密钥库（此处以 ES 安装目录下的http.p12为例）建立双向信任，生产环境推荐方式：

source { Elasticsearch { hosts = ["https://localhost:9200"] username = "elastic" password = "elasticsearch" tls_keystore_path = "${your elasticsearch home}/config/certs/http.p12" tls_keystore_password = "${your password}" } }

案例六：SQL 方式查询

使用 X-Pack SQL 语法查询。注意：SQL 查询不支持 map 和数组类型字段。

source { Elasticsearch { hosts = ["https://elasticsearch:9200"] username = "elastic" password = "elasticsearch" tls_verify_certificate = false tls_verify_hostname = false index = "st_index_sql" sql_query = "select * from st_index_sql where c_int>=10 and c_int<=20" search_type = "sql" } }

案例七：PIT 方式滚动查询

使用 DSL 查询 + Point in Time 分页，pit_keep_alive为 60000 毫秒（1 分钟），每次拉取 100 条：

source { Elasticsearch { hosts = ["https://elasticsearch:9200"] username = "elastic" password = "elasticsearch" tls_verify_certificate = false tls_verify_hostname = false index = "st_index" query = {"range": {"c_int": {"gte": 10, "lte": 20}}} # 使用 DSL 查询和 PIT API search_type = DSL search_api_type = PIT pit_keep_alive = 60000 # 1 minute in milliseconds pit_batch_size = 100 } }

案例八：Runtime Fields 查询期计算字段

在查询时计算字段值而无需重建索引。定义 4 个运行时字段（含条件分支与脚本参数），并声明输出字段与 schema 后写入 Console：

source { Elasticsearch { hosts = ["https://elasticsearch:9200"] username = "elastic" password = "elasticsearch" tls_verify_certificate = false tls_verify_hostname = false index = "sales_data" # 定义运行时字段 runtime_fields = [ { name = "total_amount" type = "double" script = "emit(doc['quantity'].value * doc['price'].value)" }, { name = "day_of_week" type = "keyword" script = "emit(doc['order_date'].value.dayOfWeekEnum.getDisplayName(TextStyle.FULL, Locale.ROOT))" }, { name = "order_category" type = "keyword" script = """ double amount = doc['quantity'].value * doc['price'].value; if (amount > 1000) { emit('high_value'); } else if (amount > 100) { emit('medium_value'); } else { emit('low_value'); } """ }, { name = "price_with_tax" type = "double" script = "emit(doc['price'].value * (1 + params.tax_rate))" script_params = { tax_rate = 0.13 } } ] source = [ "product_id", "quantity", "price", "order_date", "total_amount", "day_of_week", "order_category", "price_with_tax" ] schema = { fields { product_id = string quantity = int price = double order_date = timestamp total_amount = double day_of_week = string order_category = string price_with_tax = double } } } } sink { Console { } }

说明：虽然schema已标记为废弃，但示例中保留它是为了给运行时字段声明明确的输出类型；纯source模式无法推导运行时字段类型，因此在包含runtime_fields的场景中仍需以schema补齐类型信息。

案例九：PIT + slicing 并行读取

在 PIT 分页基础上开启slice_max = 2，将st_index拆分为 2 个切片并行拉取，提升吞吐：

source { Elasticsearch { hosts = ["https://elasticsearch:9200"] username = "elastic" password = "elasticsearch" tls_verify_certificate = false tls_verify_hostname = false index = "st_index" query = {"range": {"c_int": {"gte": 10, "lte": 20}}} search_type = DSL search_api_type = PIT pit_keep_alive = 60000 pit_batch_size = 100 # 开启切片并行读取 slice_max = 2 } }

版本演进与变更记录

该连接器自 SeaTunnel 2.2.0-beta 引入 ES Sink、2.3.0 正式支持 Source 以来持续演进，完整的逐版本变更见 connector-elasticsearch 变更日志，与本 Source 相关的重要里程碑包括：

• 2.3.8：支持多表源（multi-table source）特性，即index_list；
• 2.3.10：支持 Elasticsearch SQL Source（search_type = SQL）；
• 2.3.11：支持 PIT 分页（search_api_type = PIT）；
• 2.3.12：新增 API Key 认证支持（auth_type = api_key/api_key_encoded）。

配置建议小结

1. 连接：生产环境优先 HTTPS + 证书校验（案例五），测试环境可临时关闭校验（案例三/四）；启用安全认证的集群按需选择 basic 或 API Key。
2. 查询：默认 DSL +query即可满足绝大多数场景；需要复杂聚合/投影计算时改用 SQL，但要避开 map/array 字段并接受无法切片的限制。
3. 分页：ES ≥ 7.10 且对一致性有要求时优先 PIT（共享快照），老版本集群用 SCROLL 并将scroll_time调大到足以覆盖最慢批次的处理耗时。
4. 吞吐：大索引开启slice_max > 1配合调大parallelism；数据持续追加、可容忍近似一致性的场景收益最大。
5. 多索引：结构不同、条件不同的索引使用index_list逐条配置，避免把query放在顶层与index_list混用。

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel