使用LogStash将数据同步至PolarSearch

相关参数说明

• input源库中的index字段支持使用通配符*以同步多个索引。然而，不建议使用全通配逻辑*来复制所有索引，因为这可能会导致不必要的内部索引复制。

• input源库中的docinfo字段，可以获取原始索引名和文档ID。

• output目标库中index名字等可以通过记录的metadata读取，从而保持索引名等不变或在原有名字等元信息基础上进行定制。

• output目标库中可以增加stdout调试选项可输出调试信息。

更多信息，请参见同步Logstash事件至OpenSearch。

调试输出

如下所示，通过该输出可以观察到记录的_index等metadata及其结构层次。例如，_index字段的结构层次表明其在output等后续流程中的提取逻辑为：[@metadata][input][opensearch][_index]。

[2025-10-23T01:36:52,765][INFO ][logstash.inputs.opensearch][main][bb6f7ddd51cb0a42a08aa287556a451e8939ecbc0ce17c51ddffe0ae0b0d217e] Slice complete {:slice_id=>2, :slices=>4}
[2025-10-23T01:36:52,787][INFO ][logstash.inputs.opensearch][main][bb6f7ddd51cb0a42a08aa287556a451e8939ecbc0ce17c51ddffe0ae0b0d217e] Slice complete {:slice_id=>0, :slices=>4}
{
      "@timestamp" => 2025-10-22T17:36:52.780903870Z,
           "price" => 29.99,
        "@metadata" => {
        "input" => {
            "opensearch" => {
                "_index" => "lptest",
                   "_id" => "sQXXDJoBvnBnbbz3pKb9",
                 "_type" => nil
            }
        }
    },
        "@version" => "1",
     "description" => "这是一个示例文档",
           "title" => "示例标题",
       "timestamp" => "2024-01-15T10:30:00Z"
}
[2025-10-23T01:36:57,204][INFO ][logstash.javapipeline    ][main] Pipeline terminated {"pipeline.id"=>"main"}
[2025-10-23T01:36:57,263][INFO ][logstash.pipelinesregistry] Removed pipeline from registry successfully {:pipeline_id=>:main}

配置文件示例

# synchronization.conf
# 一个从 Elasticsearch/OpenSearch 向 PolarSearch 同步数据的完整配置示例。

input {
  # 如果源集群是 Elasticsearch，请将 'opensearch' 替换为 'elasticsearch'。
  # 两个插件的配置参数基本相同，但元数据路径可能存在差异。
  opensearch {
    # 【必填】源集群的连接地址，建议使用 HTTPS 协议。
    hosts => ["https://source-cluster-endpoint:9200"]
    # 【必填】源集群的认证凭据。
    user => "your_source_user"
    password => "your_source_password"

    # 【必填】指定需要同步的索引，支持通配符。
    # 为避免同步不必要的内部索引（如 .kibana），不建议使用 "*" 或 ".*"。
    index => "your-business-logs-*"

    # --- 安全配置 ---
    # 如果源集群启用了 SSL/TLS，请设置为 true。
    ssl => true
    # 如果源集群使用自签名证书，取消注释并指定 CA 证书路径。
    # cacert => "/path/to/source_ca.crt"

    # --- 性能与元数据 ---
    # 开启此选项以获取原始索引名和文档 ID。
    docinfo => true
    # 设置并发读取数，建议设置为源索引的主分片数量以最大化读取性能。
    slices => 4
    # 每次批量获取的文档数量。
    size => 1000
    # 滚动查询的存活时间，确保长任务不会因超时而中断。
    scroll => "5m"
  }
}

output {
  opensearch {
    # 【必填】目标 PolarSearch 集群的连接地址。
    hosts => ["https://polarsearch-cluster-endpoint:9200"]
    # 【必填】目标集群的认证凭据。
    user => "your_target_user"
    password => "your_target_password"

    # --- 安全配置 ---
    ssl => true
    # 如果目标集群使用自签名证书，取消注释并指定 CA 证书路径。
    # cacert => "/path/to/polarsearch_ca.crt"

    # --- 索引与文档 ID ---
    # 从元数据中动态读取并设置索引名，以保持与源端一致。
    # 注意：元数据路径因 input 插件而异，需通过调试输出确认。
    index => "%{[@metadata][input][opensearch][_index]}"
    # 从元数据中读取并设置文档 ID，以保持文档的唯一性。
    document_id => "%{[@metadata][input][opensearch][_id]}"
  }

  # --- 调试输出（可选） ---
  # 在开发和测试阶段，取消此段注释可在控制台打印数据流信息。
  # 正式同步时，注释掉此段以获得最佳性能。
  # stdout {
  #   codec => rubydebug {
  #     metadata => true
  #   }
  # }
}

参考写入速率

配置文件示例

# synchronization.conf
# 一个从 Elasticsearch/OpenSearch 向 PolarSearch 增量同步数据的完整配置示例。
# 通过定时执行带时间范围过滤的查询，每个调度周期仅拉取指定时间窗口内
# 新写入或更新的文档，实现增量数据同步。

input {
  # 如果源集群是 Elasticsearch，请将 'opensearch' 替换为 'elasticsearch'。
  elasticsearch {
    # 【必填】源集群的连接地址。
    hosts => ["https://source-cluster-endpoint:9200"]
    # 【必填】源集群的认证凭据。
    user     => "your_source_user"
    password => "your_source_password"

    # 【必填】指定需要增量同步的索引，支持通配符。
    index => "your-business-logs-*"

    # --- 增量同步核心配置 ---
    # 时间范围查询：仅拉取最近 5 分钟内写入的文档。
    # gte: 大于等于当前时间减去 5 分钟。
    # lte: 小于等于当前分钟整点（now/m 表示向下取整到分钟，用于避免拉取尚未写入完毕的数据）。
    # 时间窗口大小需与 schedule 调度周期保持匹配，防止两次调度之间出现数据空档。
    query => '{"query":{"range":{"@timestamp":{"gte":"now-5m","lte":"now/m"}}}}'

    # 定时调度：使用 Cron 语法定义执行周期，以下配置为每分钟执行一次。
    # 常见示例：
    #   "* * * * *"     每分钟执行
    #   "*/5 * * * *"   每 5 分钟执行
    #   "0 * * * *"     每小时整点执行
    schedule => "* * * * *"

    # 滚动查询的存活时间，确保分批拉取时上下文不会超时中断。
    scroll => "5m"
    # 开启此选项以获取原始索引名和文档 ID。
    docinfo => true
    # 设置并发读取数，建议设置为源索引的主分片数量以最大化读取性能。
    slices => 4
    # 每次批量获取的文档数量。
    size => 5000

    # --- 安全配置 ---
    # ssl => true
    # cacert => "/path/to/source_ca.crt"
  }
}

# filter(可选): 移除 pipeline 附加的@timestamp字段
# filter {
#   mutate {
#     remove_field => ["@timestamp"]
#   }
# }

output {
  opensearch {
    # 【必填】目标 PolarSearch 集群的连接地址。
    hosts    => ["https://polarsearch-cluster-endpoint:9200"]
    # 【必填】目标集群的认证凭据。
    user     => "your_target_user"
    password => "your_target_password"

    # --- 安全配置 ---
    ssl => true
    # cacert => "/path/to/polarsearch_ca.crt"

    # 从元数据中动态读取索引名，保持与源端一致。
    # 注意：元数据路径因 input 插件而异，需通过调试输出确认。
    index       => "%{[@metadata][input][elasticsearch][_index]}"
    # 从元数据中读取文档 ID，实现幂等写入：同一文档重复同步时执行更新而非重复插入。
    document_id => "%{[@metadata][input][elasticsearch][_id]}"
  }

  # --- 调试输出（可选） ---
  # stdout {
  #   codec => rubydebug {
  #     metadata => true
  #   }
  # }
}