主要通過一個數據采集與處理的案例來介紹Streamsets（3.13.0）的使用，主要將使用Edge數據流收集streamsets系統的日志和主機性能指標，通過收集數據流收集類數據并進行簡單處理，發送至kafka中，性能指標數據入庫數據流和日志數據入庫數據流分別從kafka中消費數據，并將兩類數據進行簡單處理加載到數據庫中。

學習目的：使用edge和streamset的數據互動，使用streamset進行分布式異步數據處理。

數據流圖：

最終數據流的效果圖：

需要配置5個數據流，兩個edge采集數據流，一個數據收集數據流，兩個數據處理與入庫數據流

前期準備：

（1）需要在數據采集的節點上部署安裝Edge（不會使用的同學可以參照前面文章）。

（2）一個現成kafka集群，并創建一個兩個topic，kafka集群主要為了讓數據流達到分布式異步處理的能力。

（3）一個現成的ES集群。

構建步驟：

1.首先構建日志數據采集器數據流

配置edge數據流發布的地址（該主機上一定要安裝部署了edge）

配一下文件采集文件和數據格式，數據格式我們直接按文本傳輸

?

使用destination 類http client配置一下遠程數據收集器的地址和APP ID

2.性能指標數據采集器數據流

配置edge數據流發布的地址（該主機上一定要安裝部署了edge）

配置一下系統指標采集插件，采集哪些數據和采集的頻率，這里我們采集host、cpu、內存、磁盤等，采集頻率為兩秒

使用destination 類http client同樣配置一下數據收集器遠程的url和APP ID

3.數據收集與處理數據流

使用origin 類http sevice組件，配置數據收集器的端口、最大并發量以及APP ID

使用Http 路由插件，將接受到的數據路由不到不同分支，這里配置日志和性能指標數據路由。

使用日志解析插件對收集到的日志數據進行解析，這里我們選擇Log4j解析器，選擇使用自定義日志格式，這里的格式按照streamset的格式：

?%d{ISO8601} [user:%X{s-user}] [pipeline:%X{s-entity}] [runner:%X{s-runner}] [thread:%t] [stage:%X{s-stage}] %-5p %c{1} - %m%n

使用kafka生成插件將兩類數據輸出到不同的topic中。

4.日志數據入庫數據流

配置kafka地址和日志數據的topic和消費組

5.性能指標數據入庫數據流

配置kafka地址和性能指標數據的topic和消費組

?

Streamsets實戰之路正在更新中，盡情期待！！！

?

此文章為博主原創，轉載請標明出處和原始鏈接，謝謝。

總結

以上是生活随笔為你收集整理的StreamSets实战之路（十五）-实战篇- 数据采集与处理的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。