如何快速使用阿里雲Logstash執行個體 - Elasticsearch

背景資訊

在開始本文操作前，請先瞭解以下背景資訊：

前提條件

註冊阿里雲帳號。

具體操作，請參見帳號註冊。
建立專用網路和虛擬交換器。

具體操作，請參見搭建IPv4專用網路。

使用限制

源Elasticsearch、Logstash和目標Elasticsearch執行個體在同一專用網路。如果不在同一專用網路，需要通過配置NAT Gateway實現與公網的連通，詳細資料請參見配置NAT公網資料轉送。
源Elasticsearch、Logstash和目標Elasticsearch執行個體版本需滿足相容性要求，詳細資料請參見產品相容性。

操作流程

步驟一：環境準備

建立源和目標Elasticsearch執行個體、開啟目標Elasticsearch執行個體的自動建立索引功能、準備測試資料。
步驟二：建立阿里雲Logstash執行個體

建立阿里雲Logstash執行個體，等待執行個體狀態變為正常後，才可以建立並運行管道任務。
步驟三：建立並運行管道任務

建立並配置阿里雲Logstash管道任務，運行任務完成資料同步。
步驟四：查看資料同步結果

通過目標Elasticsearch執行個體的Kibana控制台，查看資料同步結果。

步驟一：環境準備

建立Elasticsearch執行個體。
1. 登入Elasticsearch控制台
2. 在左側導覽列，單擊Elasticsearch執行個體。
3. 在Elasticsearch執行個體頁面，建立2個Elasticsearch執行個體。
  建立的2個Elasticsearch執行個體，分別作為Logstash的input和output，具體操作請參見建立Elasticsearch執行個體。本文建立的執行個體版本為通用商業版6.7，使用的資料移轉方案為：Elasticsearch 6.7.0 > 阿里雲Logstash 6.7.0 > Elasticsearch 6.7.0，提供的指令碼僅適用於該資料移轉方案，其他方案不保證相容。建立的Elasticsearch執行個體的具體配置如下。
  說明
  
  如果您使用的是其他方案，可參見產品相容性判斷是否存在相容性問題。如果存在，可升級執行個體版本或新購執行個體。
  
  訪問Elasticsearch執行個體的帳號預設為elastic（本文以此為例），如果需要使用自建使用者，要給予自建使用者相應的角色和許可權，詳細資料請參見通過Elasticsearch X-Pack角色管理實現使用者權限管控。
開啟目標Elasticsearch執行個體的自動建立索引功能。

具體操作，請參見配置YML參數。

說明
Elasticsearch為了保證使用者操作資料的安全性，預設將自動建立索引配置設定為不允许。阿里雲Logstash在傳輸資料的時候，使用提交資料的方式建立索引，而不是Create index API的方式。所以在使用阿里雲Logstash上傳資料之前，需要先把叢集的自動建立索引設定為允许，或提前建立好索引和Mapping。

準備測試資料。

進入源Elasticsearch執行個體的Kibana控制台，在Dev Tools頁面的Console頁簽下，執行如下命令建立待同步的索引和文檔。

重要

進入Kibana控制台的具體步驟，請參見登入Kibana控制台。
以下指令碼以Elasticsearch 6.7版本為例，僅供測試。7.0及以上版本的樣本指令碼，請參見Elasticsearch快速入門。

建立名稱為my_index，類型為my_type的索引。

PUT /my_index
{
    "settings" : {
      "index" : {
        "number_of_shards" : "5",
        "number_of_replicas" : "1"
      }
    },
    "mappings" : {
        "my_type" : {
            "properties" : {
              "post_date": {          
                   "type": "date"       
               },
              "tags": {
                   "type": "keyword"
               },
                "title" : {
                    "type" : "text"
                }
            }
        }
    }
}

在my_index索引中插入一個名稱為1的文檔。

PUT /my_index/my_type/1?pretty
{
  "title": "One", 
  "tags": ["ruby"],
  "post_date":"2009-11-15T13:00:00"
}

在my_index索引中插入一個名稱為2的文檔。

PUT /my_index/my_type/2?pretty
{
  "title": "Two", 
  "tags": ["ruby"],
  "post_date":"2009-11-15T14:00:00"
}

步驟二：建立阿里雲Logstash執行個體

進入Logstash執行個體頁面。
1. 進入Elasticsearch控制台的Logstash頁面。
2. 在頂部功能表列，選擇與目標Elasticsearch執行個體相同的地區。
3. 在左側導覽列，單擊LogStash執行個體。
在LogStash執行個體頁面，單擊创建。
在購買頁面，完成執行個體啟動配置。
本文選擇執行個體的付費模式為隨用隨付，版本為6.7，其餘配置均保持預設。更多配置資訊，請參見建立阿里雲Logstash執行個體。
說明
- 在前期程式研發或功能測試期間，建議購買隨用隨付執行個體測試。
- 購買訂用帳戶執行個體，可以享受優惠條件。
選中服務合約，單擊DataWorks獨享資源（訂用帳戶）。
提示開通成功後，單擊DataWorks獨享資源（訂用帳戶）。
在頂部功能表列，選擇執行個體所在地區。在左側導覽列，單擊LogStash執行個體，進入LogStash執行個體頁面，查看建立成功的執行個體。

步驟三：建立並運行管道任務

等到建立的Logstash執行個體狀態變為正常後，您可以建立並運行管道任務同步資料。

在LogStash執行個體頁面，單擊目標執行個體右側操作列下的管道管理。
在管道列表地區，單擊建立管道。

輸入管道ID和Config Settings。

本文使用的Config配置如下。

input {
    elasticsearch {
        hosts => ["http://es-cn-0pp1f1y5g000h****.elasticsearch.aliyuncs.com:9200"]
        user => "elastic"
        password => "your_password"
        index => "*,-.monitoring*,-.security*,-.kibana*"
        docinfo => true
    }
}
filter {}
output {
    elasticsearch {
        hosts => ["http://es-cn-mp91cbxsm000c****.elasticsearch.aliyuncs.com:9200"]
        user => "elastic"
        password => "your_password"
        index => "%{[@metadata][_index]}"
        document_type => "%{[@metadata][_type]}"
        document_id => "%{[@metadata][_id]}"
    }
    file_extend {
        path => "/ssd/1/ls-cn-v0h1kzca****/logstash/logs/debug/test"
    }
}

參數	說明
hosts	Elasticsearch服務的訪問地址。input中為`http://<源執行個體ID>.elasticsearch.aliyuncs.com:9200`；output中為`http://<目標執行個體ID>.elasticsearch.aliyuncs.com:9200`。
user	訪問Elasticsearch服務的使用者名稱，預設為elastic。
password	對應使用者的密碼。elastic使用者的密碼在建立執行個體時設定，如果忘記可進行重設，重設密碼的注意事項和操作步驟請參見重設執行個體訪問密碼。
index	指定同步索引名。設定為`,-.monitoring,-.security,-.kibana`，表示同步除了`.`開頭的系統索引外的所有索引。`%{[@metadata][_index]}`，表示匹配中繼資料中的index，即同步後索引的名稱和源索引名稱相同。說明系統索引一般用來儲存Elasticsearch叢集的監控日誌，無需同步。
docinfo	設定為true，將會提取Elasticsearch文檔的元資訊，例如index、type和id。
document_type	指定同步後索引的類型。設定為`%{[@metadata][_type]}`，表示匹配中繼資料中的type，即同步後索引的類型和源索引類型相同。
document_id	指定同步後文檔的ID。設定為`%{[@metadata][_id]}`，表示匹配中繼資料中的id，即同步後文檔的ID和來源文件ID相同。
file_extend	可選，用來開啟調試日誌功能，並通過path參數配置調試日誌的輸出路徑。建議您配置該參數，配置後，可直接在控制台上查看輸出結果。如果未配置，需要去目標端確認輸出結果，再返回控制台修改，這樣會耗費大量的時間和人力。詳細資料，請參見使用Logstash管道配置調試功能。重要使用file_extend參數前，需要先安裝logstash-output-file_extend外掛程式。具體操作，請參見安裝或卸載外掛程式。其中的path參數預設為系統指定路徑，請勿修改。您也可以單擊開啟配置調試擷取path路徑。

Config配置的結構及支援的資料類型的詳細資料（不同版本支援的資料類型可能不同），請參見Structure of a Config File。

單擊下一步，配置管道參數。

在配置的管道參數中，Pipeline Workers配置為執行個體的CPU核心數，其他參數均為預設值。詳細參數說明，請參見通過設定檔管理管道。
單擊保存或者儲存並部署。
- 保存：將管道資訊儲存在Logstash裡並觸發執行個體變更，配置不會生效。儲存後，系統會返回管道管理頁面。可在管道列表地區，單擊操作列下的立即部署，觸發執行個體重啟，使配置生效。
- 儲存並部署：儲存並且部署後，會觸發執行個體重啟，使配置生效。
在建立成功提示框中，單擊確認。

確認後，可在管道列表中查看建立成功的管道。等待執行個體變更完成，並且管道的狀態顯示為運行中時，表示阿里雲Logstash開始執行同步任務。

步驟四：查看資料同步結果

資料同步任務配置完成並開始運行後，您可以通過目標Elasticsearch的Kibana控制台，查看資料同步結果。

登入目標Elasticsearch執行個體的Kibana控制台，根據頁面提示進入Kibana首頁。
登入Kibana控制台的具體操作，請參見登入Kibana控制台。
說明
本文以Elasticsearch 6.7.0版本為例，其他版本操作可能略有差別，請以實際介面為準。
在左側導覽列，單擊Dev Tools。

在Console中，執行如下命令查看資料同步結果。

GET /my_index/_search
{
  "query": {
    "match_all": {}
  }
}

預期結果如下。

{
  "took" : 2,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 4,
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "my_index",
        "_type" : "doc",
        "_id" : "aKG3bHABnsveUjNW2iXy",
        "_score" : 1.0,
        "_source" : {
          "post_date" : "2009-11-15T14:00:00",
          "@version" : "1",
          "@timestamp" : "2020-02-22T11:45:52.992Z",
          "tags" : [
            "ruby"
          ],
          "title" : "Two"
        }
      },
      ...
    ]
  }
}

如果源端和目標端資料一致，表示資料同步成功。您也可以通過GET _cat/indices?v命令，查看源端和目標端相同索引的大小是否一致，來判斷資料是否同步成功。

Elasticsearch：快速入門