Storm日誌分析調研及其實時架構
1.Storm第一個Demo
2.Windows下基於eclipse的Storm應用開發與調試
3.Storm實例+mysql數據庫保存
4.Storm原理介紹
5. flume+kafka+storm+mysql 實時架構
1.Storm第一個Demo
Storm運行模式：html
- 本地模式(Local Mode)： 即Topology（至關於一個任務，後續會詳細講解） 運行在本地機器的單一JVM上，這個模式主要用來開發、調試。
- 遠程模式(Remote Mode):在這個模式，咱們把咱們的Topology提交到集羣，在這個模式中，Storm的全部組件都是線程安全的，由於它們都會運行在不一樣的Jvm或物理機器上，這個模式就是正式的生產模式。
寫一個HelloWord Storm
咱們如今建立這麼一個應用，統計文本文件中的單詞個數，詳細學習過Hadoop的朋友都應該寫過。那麼咱們須要具體建立這樣一個Topology，用一個spout負責讀取文本文件，用第一個bolt來解析成單詞，用第二個bolt來對解析出的單詞計數，總體結構如圖所示：
寫一個可運行的Demo很簡單，咱們只須要三步：
- 建立一個Spout讀取數據
- 建立bolt處理數據
- 建立一個Topology提交到集羣
下面咱們就寫一下，如下代碼拷貝到eclipse（依賴的jar包到官網下載便可）便可運行。java
1.建立一個Spout做爲數據源
Spout做爲數據源，它實現了IRichSpout接口，功能是讀取一個文本文件並把它的每一行內容發送給bolt。
package storm.demo.spout;
import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.util.Map;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
public class WordReader implements IRichSpout {
private static final long serialVersionUID = 1L;
private SpoutOutputCollector collector;
private FileReader fileReader;
private boolean completed = false;
public boolean isDistributed() {
return false;
}
/**
* 這是第一個方法，裏面接收了三個參數，第一個是建立Topology時的配置，
* 第二個是全部的Topology數據，第三個是用來把Spout的數據發射給bolt
* **/
@Override
public void open(Map conf, TopologyContext context,
SpoutOutputCollector collector) {
try {
//獲取建立Topology時指定的要讀取的文件路徑
this.fileReader = new FileReader(conf.get("wordsFile").toString());
} catch (FileNotFoundException e) {
throw new RuntimeException("Error reading file ["
+ conf.get("wordFile") + "]");
}
//初始化發射器
this.collector = collector;
}
/**
* 這是Spout最主要的方法，在這裏咱們讀取文本文件，並把它的每一行發射出去（給bolt）
* 這個方法會不斷被調用，爲了下降它對CPU的消耗，當任務完成時讓它sleep一下
* **/
@Override
public void nextTuple() {
if (completed) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
// Do nothing
}
return;
}
String str;
// Open the reader
BufferedReader reader = new BufferedReader(fileReader);
try {
// Read all lines
while ((str = reader.readLine()) != null) {
/**
* 發射每一行，Values是一個ArrayList的實現
*/
this.collector.emit(new Values(str), str);
}
} catch (Exception e) {
throw new RuntimeException("Error reading tuple", e);
} finally {
completed = true;
}
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("line"));
}
@Override
public void close() {
// TODO Auto-generated method stub
}
@Override
public void activate() {
// TODO Auto-generated method stub
}
@Override
public void deactivate() {
// TODO Auto-generated method stub
}
@Override
public void ack(Object msgId) {
System.out.println("OK:" + msgId);
}
@Override
public void fail(Object msgId) {
System.out.println("FAIL:" + msgId);
}
@Override
public Map<String, Object> getComponentConfiguration() {
// TODO Auto-generated method stub
return null;
}
}
2.建立兩個bolt來處理Spout發射出的數據
Spout已經成功讀取文件並把每一行做爲一個tuple(在Storm數據以tuple的形式傳遞)發射過來，咱們這裏須要建立兩個bolt分別來負責解析每一行和對單詞計數。
Bolt中最重要的是execute方法，每當一個tuple傳過來時它便會被調用。
第一個bolt：WordNormalizer
package storm.demo.bolt;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
public class WordNormalizer implements IRichBolt {
private OutputCollector collector;
@Override
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.collector = collector;
}
/**這是bolt中最重要的方法，每當接收到一個tuple時，此方法便被調用
* 這個方法的做用就是把文本文件中的每一行切分紅一個個單詞，並把這些單詞發射出去（給下一個bolt處理）
* **/
@Override
public void execute(Tuple input) {
String sentence = input.getString(0);
String[] words = sentence.split(" ");
for (String word : words) {
word = word.trim();
if (!word.isEmpty()) {
word = word.toLowerCase();
// Emit the word
List a = new ArrayList();
a.add(input);
collector.emit(a, new Values(word));
}
}
//確認成功處理一個tuple
collector.ack(input);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
@Override
public void cleanup() {
// TODO Auto-generated method stub
}
@Override
public Map<String, Object> getComponentConfiguration() {
// TODO Auto-generated method stub
return null;
}
}
第二個bolt:WordCountermysql
package storm.demo.bolt;
import java.util.HashMap;
import java.util.Map;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;
public class WordCounter implements IRichBolt {
Integer id;
String name;
Map<String, Integer> counters;
private OutputCollector collector;
@Override
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.counters = new HashMap<String, Integer>();
this.collector = collector;
this.name = context.getThisComponentId();
this.id = context.getThisTaskId();
}
@Override
public void execute(Tuple input) {
String str = input.getString(0);
if (!counters.containsKey(str)) {
counters.put(str, 1);
} else {
Integer c = counters.get(str) + 1;
counters.put(str, c);
}
// 確認成功處理一個tuple
collector.ack(input);
}
/**
* Topology執行完畢的清理工做，好比關閉鏈接、釋放資源等操做都會寫在這裏
* 由於這只是個Demo，咱們用它來打印咱們的計數器
* */
@Override
public void cleanup() {
System.out.println("-- Word Counter [" + name + "-" + id + "] --");
for (Map.Entry<String, Integer> entry : counters.entrySet()) {
System.out.println(entry.getKey() + ": " + entry.getValue());
}
counters.clear();
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// TODO Auto-generated method stub
}
@Override
public Map<String, Object> getComponentConfiguration() {
// TODO Auto-generated method stub
return null;
}
}
3.在main函數中建立一個Topology
在這裏咱們要建立一個Topology和一個LocalCluster對象，還有一個Config對象作一些配置。
package storm.demo;
import storm.demo.bolt.WordCounter;
import storm.demo.bolt.WordNormalizer;
import storm.demo.spout.WordReader;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
public class WordCountTopologyMain {
public static void main(String[] args) throws InterruptedException {
//定義一個Topology
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("word-reader",new WordReader());
builder.setBolt("word-normalizer", new WordNormalizer())
.shuffleGrouping("word-reader");
builder.setBolt("word-counter", new WordCounter(),2)
.fieldsGrouping("word-normalizer", new Fields("word"));
//配置
Config conf = new Config();
conf.put("wordsFile", "d:/text.txt");
conf.setDebug(false);
//提交Topology
conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
//建立一個本地模式cluster
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("Getting-Started-Toplogie", conf,
builder.createTopology());
Thread.sleep(1000);
cluster.shutdown();
}
}
運行這個函數咱們便可看到後臺打印出來的單詞個數。git
3.Storm實例+mysql數據庫保存
package com.qing.storm.Spout;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.util.Map;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
@SuppressWarnings("serial")
public class ReadLogSpout extends BaseRichSpout {
private SpoutOutputCollector collector;
FileInputStream fis;
InputStreamReader isr;
BufferedReader br;
@Override
public void nextTuple() {
// TODO Auto-generated method stub
String str = "";
try {
while ((str = this.br.readLine()) != null) {
this.collector.emit(new Values(str));
Thread.sleep(100);
}
} catch (Exception e) {
e.printStackTrace();
}
}
@SuppressWarnings("rawtypes")
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
// TODO Auto-generated method stub
this.collector = collector;
String file = "/opt/apache-storm-0.9.3/bin/domain.log";
try{
this.fis = new FileInputStream(file);
this.isr = new InputStreamReader(fis);
this.br = new BufferedReader(isr);
} catch (Exception e) {
e.printStackTrace();
}
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// TODO Auto-generated method stub
declarer.declare(new Fields("str"));
}
}
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
package com.qing.storm.Bolt;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
@SuppressWarnings("serial")
public class SplitBolt extends BaseBasicBolt{
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// TODO Auto-generated method stub
declarer.declare(new Fields("word"));
}
@Override
public void execute(Tuple tuple, BasicOutputCollector collector) {
// TODO Auto-generated method stub
String sentence = tuple.getString(0);
for(String word: sentence.split(" ")){
collector.emit(new Values(word));
}
}
}
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
package com.qing.storm.Bolt;
import java.util.HashMap;
import java.util.Map;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
@SuppressWarnings("serial")
public class WordCountBolt extends BaseBasicBolt{
Map<String, Integer> counts = new HashMap<String, Integer>();
public void execute(Tuple tuple, BasicOutputCollector collector) {
// TODO Auto-generated method stub
String word = tuple.getString(0);
Integer count = counts.get(word);
if(count == null){
count = 0;
}
count++;
counts.put(word, count);
collector.emit(new Values(word,count));
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// TODO Auto-generated method stub
declarer.declare(new Fields("word", "count"));
}
}
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
package com.qing.storm.Bolt;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Map;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;
@SuppressWarnings("serial")
public class MysqlBolt extends BaseRichBolt{
private OutputCollector collector;
Connection conn = null;
String from = "word_count"; //表名
private String word;
private int num;
@Override
public void declareOutputFields(OutputFieldsDeclarer arg0) {
// TODO Auto-generated method stub
}
@Override
public void prepare(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, OutputCollector collector) {
// TODO Auto-generated method stub
this.collector = collector;
try {
LinkDB();
} catch (InstantiationException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (IllegalAccessException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
private void LinkDB() throws InstantiationException, IllegalAccessException, SQLException {
// TODO Auto-generated method stub
String host_port = "127.0.0.1:3306";
String database = "storm_test";
String username = "root";
String password = "root";
String url = "jdbc:mysql://" + host_port + "/" + database;
try {
Class.forName("com.mysql.jdbc.Driver");
conn = DriverManager.getConnection(url, username, password);
} catch (ClassNotFoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
@Override
public void execute(Tuple tuple) {
// TODO Auto-generated method stub
String word= tuple.getString(0);
int num = tuple.getInteger(1);
InsertDB(word, num);
}
private void InsertDB(String word, int num) {
// TODO Auto-generated method stub
this.word = word;
this.num = num;
String sql = "replace into " + this.from+ "(word, num) values ('" +word+"',"+num+ ")";
try {
Statement statement = conn.createStatement();
statement.executeUpdate(sql);
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
package com.qing.storm.Topology;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import com.qing.storm.Bolt.MysqlBolt;
import com.qing.storm.Bolt.SplitBolt;
import com.qing.storm.Bolt.WordCountBolt;
import com.qing.storm.Spout.ReadLogSpout;
public class DB_Topology {
public static void main(String[] args){
try {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("spout", new ReadLogSpout(), 5);
builder.setBolt("split", new SplitBolt(), 8).shuffleGrouping("spout");
builder.setBolt("count", new WordCountBolt(), 10).fieldsGrouping("split", new Fields("word"));
builder.setBolt("Mysql", new MysqlBolt(),10).fieldsGrouping("count", new Fields("word","count"));
Config conf = new Config();
conf.setDebug(true);
//if (args != null && args.length > 0) {
/*設置該topology在storm集羣中要搶佔的資源slot數，一個slot對應這supervisor節點上的以個worker進程
若是你分配的spot數超過了你的物理節點所擁有的worker數目的話，有可能提交不成功，加入你的集羣上面已經有了
一些topology而如今還剩下2個worker資源，若是你在代碼裏分配4個給你的topology的話，那麼這個topology能夠提交
可是提交之後你會發現並無運行。 而當你kill掉一些topology後釋放了一些slot後你的這個topology就會恢復正常運行。
*/
//conf.setNumWorkers(1);
if (args != null && args.length > 0) {
conf.setNumWorkers(1);
StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
}
} catch (AlreadyAliveException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InvalidTopologyException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
// conf.setMaxTaskParallelism(1);
// LocalCluster cluster = new LocalCluster();
// cluster.submitTopology("word-count", conf, builder.createTopology());
// try {
// Thread.sleep(1000000);
// } catch (InterruptedException e) {
// // TODO Auto-generated catch block
// e.printStackTrace();
// }
//cluster.shutdown();
}
4.Storm原理介紹
Storm是使用Clojure語言開發，可是能夠在Storm中使用任何語言編寫應用程序，所需的只是一個鏈接到Storm 的架構的適配器。已存在針對 Scala、JRuby、Perl 和 PHP 的適配器，可是還有支持流式傳輸到 Storm 拓撲結構中的結構化查詢語言適配器。github
Storm 的關鍵屬性
Storm 實現的一些特徵決定了它的性能和可靠性的。Storm 使用 ZeroMQ 傳送消息，這就消除了中間的排隊過程，使得消息可以直接在任務自身之間流動。在消息的背後，是一種用於序列化和反序列化 Storm 的原語類型的自動化且高效的機制。
Storm 的一個最有趣的地方是它注重容錯和管理。Storm 實現了有保障的消息處理，因此每一個元組都會經過該拓撲結構進行全面處理；若是發現一個元組還未處理會自動從噴嘴處重放。Storm 還實現了任務級的故障檢測，在一個任務發生故障時，消息會自動從新分配以快速從新開始處理。Storm 包含比 Hadoop 更智能的處理管理，流程會由監管員來進行管理，以確保資源獲得充分使用。sql
Storm 實現了一種數據流模型，其中數據持續地流經一個轉換實體網絡（參見 圖 1）。一個數據流的抽象稱爲一個流，這是一個無限的元組序列。元組就像一種使用一些附加的序列化代碼來表示標準數據類型（好比整數、浮點和字節數組）或用戶定義類型的結構。每一個流由一個唯一 ID 定義，這個 ID 可用於構建數據源和接收器 (sink) 的拓撲結構。流起源於噴嘴，噴嘴將數據從外部來源流入 Storm 拓撲結構中。數據庫
接收器（或提供轉換的實體）稱爲螺栓。螺栓實現了一個流上的單一轉換和一個 Storm 拓撲結構中的全部處理。螺栓既可實現 MapReduce 之類的傳統功能，也可實現更復雜的操做（單步功能），好比過濾、聚合或與數據庫等外部實體通訊。典型的 Storm 拓撲結構會實現多個轉換，所以須要多個具備獨立元組流的螺栓。噴嘴和螺栓都實現爲 Linux® 系統中的一個或多個任務。apache
使用 Storm 爲詞頻輕鬆地實現 MapReduce 功能。如 圖 2 中所示，噴嘴生成文本數據流，螺栓實現 Map 功能（令牌化一個流的各個單詞）。來自 「map」 螺栓的流而後流入一個實現 Reduce 功能的螺栓中（以將單詞聚合到總數中）。數組
請注意，螺栓可將數據傳輸到多個螺栓，也可接受來自多個來源的數據。Storm 擁有流分組 的概念，流分組實現了混排 (shuffling)（隨機但均等地將元組分發到螺栓）或字段分組（根據流的字段進行流分區）。還存在其餘流分組，包括生成者使用本身的內部邏輯路由元組的能力。安全
可是，Storm 架構中一個最有趣的特性是有保障的消息處理。Storm 可保證一個噴嘴發射出的每一個元組都會處理；若是它在超時時間內沒有處理，Storm 會從該噴嘴重放該元組。此功能須要一些聰明的技巧來在拓撲結構中跟蹤元素，也是 Storm 的重要的附加價值之一。
除了支持可靠的消息傳送外，Storm 還使用 ZeroMQ 最大化消息傳送性能（刪除中間排隊，實現消息在任務間的直接傳送）。ZeroMQ 合併了擁塞檢測並調整了它的通訊，以優化可用的帶寬。
5. flume+kafka+storm+mysql 實時架構
flume的架構圖：
kafka的架構圖：
storm的架構圖：
flume + kafka + storm +mysql的數據流架構圖：
下面介紹一下kafka到storm的配置：
其實這些都是經過java代碼實現的，這裏用到了 KafkaSpout類，RDBMSDumperBolt類（之後這些能夠做爲工具類打包上傳到集羣中）
storm做業中，咱們寫了一個KafkaStormRdbms類，做業具體配置以下：
首先設置鏈接mysql的參數
ArrayList<String> columnNames = new ArrayList<String>();
ArrayList<String> columnTypes = new ArrayList<String>();
String tableName = "stormTestTable_01";
// Note: if the rdbms table need not to have a primary key, set the variable 'primaryKey' to 'N/A'
// else set its value to the name of the tuple field which is to be treated as primary key
String primaryKey = "N/A";
String rdbmsUrl = "jdbc:mysql://$hostname:3306/fuqingwuDB" ;
String rdbmsUserName = "fuqingwu";
String rdbmsPassword = "password";
//add the column names and the respective types in the two arraylists
columnNames.add("word");
//add the types
columnTypes.add("varchar (100)");
配置 KafkaSpout 及 Topology：
TopologyBuilder builder = new TopologyBuilder();
List<String> hosts = new ArrayList<String>();
hosts.add("hadoop01");
SpoutConfig spoutConf = SpoutConfig.fromHostStrings(hosts, 1, "flume_kafka", "/root", "id");
spoutConf.scheme = new StringScheme();
spoutConf.forceStartOffsetTime(-2);
spoutConf.zkServers = new ArrayList<String>() {{
add("hadoop01");
}};
spoutConf.zkPort = 2181;
//set the spout for the topology
builder.setSpout("spout", new KafkaSpout(spoutConf), 1);
//dump the stream data into rdbms table
RDBMSDumperBolt dumperBolt = new RDBMSDumperBolt(primaryKey, tableName, columnNames, columnTypes, rdbmsUrl, rdbmsUserName, rdbmsPassword);
builder.setBolt("dumperBolt",dumperBolt, 1).shuffleGrouping("spout");
原文鏈接：http://blog.csdn.net/baiyangfu_love/article/details/8096088
GitHub：https://github.com/baniuyao/flume-kafka
這個框架用的組件基本都是最新穩定版本，flume-ng1.4+kafka0.8+storm0.9+mysql架構設計：
1）.數據採集
負責從各節點上實時採集數據，選用cloudera的flume來實現
2）.數據接入
因爲採集數據的速度和數據處理的速度不必定同步，所以添加一個消息中間件來做爲緩衝，選用apache的kafka
3）.流式計算
對採集到的數據進行實時分析，選用apache的storm
4）.數據輸出
對分析後的結果持久化，暫定用mysql
參考：http://blog.csdn.net/mylittlered/article/details/20810265
http://www.blogchong.com/post/storm_data_Platform.html
相關文章
- 1. 日誌實時分析架構
- 2. 日誌分析系統粗略調研
- 3. CDN日誌實時分析
- 4. storm用來作實時日誌分析的淺顯使用
- 5. 實時日誌分析系統的基本架構
- 6. ElasticSearch構建千億級日誌分析系統-實時分析
- 7. 架構之ELK日誌分析系統
- 8. Lamda架構-分析日誌流水線
- 9. 點擊流日誌分析架構
- 10. 日誌實時處理架構方案
- 更多相關文章...
- • SQLite 日期 & 時間 - SQLite教程
- • Docker 架構 - Docker教程
- • Flink 數據傳輸及反壓詳解
- • 三篇文章瞭解 TiDB 技術內幕 —— 談調度
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. 日誌實時分析架構
- 2. 日誌分析系統粗略調研
- 3. CDN日誌實時分析
- 4. storm用來作實時日誌分析的淺顯使用
- 5. 實時日誌分析系統的基本架構
- 6. ElasticSearch構建千億級日誌分析系統-實時分析
- 7. 架構之ELK日誌分析系統
- 8. Lamda架構-分析日誌流水線
- 9. 點擊流日誌分析架構
- 10. 日誌實時處理架構方案