本篇文章為大家展示了Kafka中如何將數(shù)據(jù)導(dǎo)入到Elasticsearch，內(nèi)容簡明扼要并且容易理解，絕對能使你眼前一亮，通過這篇文章的詳細(xì)介紹希望你能有所收獲。

為天峨等地區(qū)用戶提供了全套網(wǎng)頁設(shè)計制作服務(wù)，及天峨網(wǎng)站建設(shè)行業(yè)解決方案。主營業(yè)務(wù)為成都網(wǎng)站設(shè)計、做網(wǎng)站、天峨網(wǎng)站設(shè)計，以傳統(tǒng)方式定制建設(shè)網(wǎng)站，并提供域名空間備案等一條龍服務(wù)，秉承以專業(yè)、用心的態(tài)度為用戶提供真誠的服務(wù)。我們深信只要達(dá)到每一位用戶的要求，就會得到認(rèn)可，從而選擇與我們長期合作。這樣，我們也可以走得更遠(yuǎn)！

Elasticsearch作為當(dāng)前主流的 全文檢索 引擎，除了強(qiáng)大的全文檢索能力和高擴(kuò)展性之外，對多種數(shù)據(jù)源的兼容能力也是其成功的秘訣之一。而Elasticsearch強(qiáng)大的數(shù)據(jù)源兼容能力，主要來源于其核心組件之一的Logstash, Logstash通過插件的形式實現(xiàn)了對多種數(shù)據(jù)源的輸入和輸出。Kafka是一種高吞吐量的分布式發(fā)布訂閱消息系統(tǒng)，是一種常見的數(shù)據(jù)源，也是Logstash支持的眾多輸入輸出源的其中一個。將從實踐的角度，研究使用Logstash Kafka Input插件實現(xiàn)將Kafka中數(shù)據(jù)導(dǎo)入到Elasticsearch的過程。

使用Logstash Kafka插件連接Kafka和Elasticsearch

1 Logstash Kafka input插件簡介

Logstash Kafka Input插件使用Kafka API從Kafka topic中讀取數(shù)據(jù)信息，使用時需要注意Kafka的版本及對應(yīng)的插件版本是否一致。該插件支持通過SSL和Kerveros SASL方式連接Kafka。另外該插件提供了group管理，并使用默認(rèn)的offset管理策略來操作Kafka topic。

Logstash默認(rèn)情況下會使用一個單獨(dú)的group來訂閱Kafka消息，每個Logstash Kafka Consumer會使用多個線程來增加吞吐量。當(dāng)然也可以多個Logstash實例使用同一個group_id，來均衡負(fù)載。另外建議把Consumer的個數(shù)設(shè)置為Kafka分區(qū)的大小，以提供更好的性能。

2 測試環(huán)境準(zhǔn)備 2.1 創(chuàng)建Elasticsearch集群

為了簡化搭建過程，本文使用了騰訊云Elasticsearch service。騰訊云Elasticsearch service不僅可以實現(xiàn)Elasticsearch集群的快速搭建，還提供了內(nèi)置Kibana，集群監(jiān)控，專用主節(jié)點，Ik分詞插件等功能，極大的簡化了Elasticsearch集群的創(chuàng)建和管理工作。

2.2 創(chuàng)建Kafka服務(wù)

Kafka服務(wù)的搭建采用騰訊云CKafka來完成。與Elasticsearch  Service 一樣，騰訊云CKafka可以實現(xiàn)Kafka服務(wù)的快速創(chuàng)建，100%兼容開源Kafka API(0.9版本)。

2.3 服務(wù)器

除了準(zhǔn)備Elasticsearch和Kafka，另外還需要準(zhǔn)備一臺服務(wù)器，用于運(yùn)行Logstash以連接Elasticsearch和Kafka。本文采用騰訊云CVM服務(wù)器

2.4 注意事項

1) 需要將Elasticsearch、Kafka和服務(wù)器創(chuàng)建在同一個網(wǎng)絡(luò)下，以便實現(xiàn)網(wǎng)絡(luò)互通。由于本文采用的是騰訊云相關(guān)的技術(shù)服務(wù)，因此只需要將Elasticsearch service，CKafka和CVM創(chuàng)建在同一個私有網(wǎng)路（VPC）下即可。

2) 注意獲取Elasticsearch serivce，CKafka和CVM的內(nèi)網(wǎng)地址和端口，以便后續(xù)服務(wù)使用

本次測試中：

服務(wù) ip port

Elasticsearch service	192.168.0.8	9200
Ckafka	192.168.13.10	9092
CVM	192.168.0.13	-

3 使用Logstash連接Elasticsearch和Kafka 3.1 Kafka準(zhǔn)備

可以參考[CKafka 使用入門]

按照上面的教程

1) 創(chuàng)建名為kafka_es_test的topic

2) 安裝JDK

3) 安裝Kafka工具包

4) 創(chuàng)建producer和consumer驗證kafka功能

3.2 安裝Logstash

Logstash的安裝和使用可以參考[一文快速上手Logstash]

3.3 配置Logstash Kafka input插件

創(chuàng)建kafka_test_pipeline.conf文件內(nèi)容如下：

input{
        kafka{
                bootstrap_servers=>"192.168.13.10:9092"
                topics=>["kafka_es_test"]
                group_id=>"logstash_kafka_test"
        }
}
output{
        elasticsearch{
                hosts=>["192.168.0.8:9200"]
        }
}

其中定義了一個kafka的input和一個elasticsearch的output

對于Kafka input插件上述三個參數(shù)為必填參數(shù)，除此之外還有一些對插件行為進(jìn)行調(diào)整的一些參數(shù)如：

auto_commit_interval_ms 用于設(shè)置Consumer提交offset給Kafka的時間間隔

consumer_threads 用于設(shè)置Consumer的線程數(shù)，默認(rèn)為1，實際中應(yīng)設(shè)置與Kafka Topic分區(qū)數(shù)一致

fetch_max_wait_ms 用于指定Consumer等待一個fetch請求達(dá)到fetch_min_bytes的最長時間

fetch_min_bytes 用于指定Consumer fetch請求應(yīng)返回的最小數(shù)據(jù)量

topics_pattern 用于通過正則訂閱符合某一規(guī)則的一組topic

更多參數(shù)參考：[Kafka Input Configuration Options]

3.4 啟動Logstash

以下操作在Logstash根目錄中進(jìn)行

1) 驗證配置

./bin/logstash -f kafka_test_pipeline.conf --config.test_and_exit

如有錯誤，根據(jù)提示修改配置文件。若配置正確會得到如下結(jié)果

Sending Logstash's logs to /root/logstash-5.6.13/logs which is now configured via log4j2.properties
[2018-11-11T15:24:01,598][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"netflow", :directory=>"/root/logstash-5.6.13/modules/netflow/configuration"}
[2018-11-11T15:24:01,603][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"fb_apache", :directory=>"/root/logstash-5.6.13/modules/fb_apache/configuration"}
Configuration OK
[2018-11-11T15:24:01,746][INFO ][logstash.runner          ] Using config.test_and_exit mode. Config Validation Result: OK. Exiting Logstash

2) 啟動Logstash

./bin/logstash -f kafka_test_pipeline.conf --config.reload.automatic

觀察日志是否有錯誤提示，并及時處理

3.4 啟動Kafka Producer

以下操作在Kafka工具包根目錄下進(jìn)行

./bin/kafka-console-producer.sh --broker-list 192.168.13.10:9092 --topic kafka_es_test

寫入測試數(shù)據(jù)

This is a message

3.5 Kibana驗證結(jié)果

登錄Elasticsearch對應(yīng)Kibana, 在Dev Tools中進(jìn)行如下操作

1) 查看索引

GET _cat/indices

可以看到一個名為logstash-xxx.xx.xx的索引被創(chuàng)建成功

green open .kibana             QUw45tN0SHqeHbF9-QVU6A 1 1 1 0 5.5kb 2.7kb
green open logstash-2018.11.11 DejRdNJVQ1e1MwbyJjJjLw 5 1 1 0 8.7kb 4.3kb

2) 查看寫入的數(shù)據(jù)

GET logstash-2018.11.11/_search

可以看到數(shù)據(jù)已經(jīng)被成功寫入

{
  "took": 0,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 1,
    "max_score": 1,
    "hits": [
      {
        "_index": "logstash-2018.11.11",
        "_type": "logs",
        "_id": "AWcBsEegMu-Dkjm1ap3H",
        "_score": 1,
        "_source": {
          "message": "This is a message",
          "@version": "1",
          "@timestamp": "2018-11-11T07:33:09.079Z"
        }
      }
    ]
  }
}

Logstash作為Elastic Stack中數(shù)據(jù)采集和處理的核心組件，為Elasticsearch提供了強(qiáng)大的數(shù)據(jù)源兼容能力。從測試過程可以看出，使用Logstash實現(xiàn)kafka和Elaticsearch的連接過程相當(dāng)簡單方便。另外Logstash的數(shù)據(jù)處理功能，也使得采用該架構(gòu)的系統(tǒng)對數(shù)據(jù)映射和處理有天然的優(yōu)勢。

上述內(nèi)容就是Kafka中如何將數(shù)據(jù)導(dǎo)入到Elasticsearch，你們學(xué)到知識或技能了嗎？如果還想學(xué)到更多技能或者豐富自己的知識儲備，歡迎關(guān)注創(chuàng)新互聯(lián)-成都網(wǎng)站建設(shè)公司行業(yè)資訊頻道。

網(wǎng)站名稱：Kafka中如何將數(shù)據(jù)導(dǎo)入到Elasticsearch-創(chuàng)新互聯(lián)
本文路徑：http://aaarwkj.com/article6/cogcog.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供網(wǎng)站維護(hù)、電子商務(wù)、微信小程序、自適應(yīng)網(wǎng)站、關(guān)鍵詞優(yōu)化、網(wǎng)站改版

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)