使用LogStash將資料同步至PolarSearch

相關參數說明

• input源庫中的index欄位支援使用萬用字元*以同步多個索引。然而，不建議使用全通配邏輯*來複製所有索引，因為這可能會導致不必要的內部索引複製。

• input源庫中的docinfo欄位，可以擷取原始索引名和文檔ID。

• output目標庫中index名字等可以通過記錄的metadata讀取，從而保持索引名等不變或在原有名字等元資訊基礎上進行定製。

• output目標庫中可以增加stdout調試選項可輸出調試資訊。

更多資訊，請參見同步Logstash事件至OpenSearch。

調試輸出

如下圖所示，通過該輸出可以觀察到記錄的_index等metadata及其結構層次。例如，_index欄位的結構層次表明其在output等後續流程中的提取邏輯為：[@metadata][input][opensearch][_index]。

設定檔樣本

# 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
  #   }
  # }
}