互聯(lián)網(wǎng)的飛速發(fā)展促進(jìn)了很多新媒體的發(fā)展，不論是知名的大V，明星還是圍觀(guān)群眾都可以通過(guò)手機在微博，朋友圈或者點(diǎn)評網(wǎng)站上發(fā)表狀態(tài)，分享自己的所見(jiàn)所想，使得“人人都有了麥克風(fēng)”。不論是熱點(diǎn)新聞還是娛樂(lè )八卦，傳播速度遠超我們的想象。可以在短短數分鐘內，有數萬(wàn)計轉發(fā)，數百萬(wàn)的閱讀。如此海量的信息可以得到爆炸式的傳播，如何能夠實(shí)時(shí)的把握民情并作出對應的處理對很多企業(yè)來(lái)說(shuō)都是至關(guān)重要的。大數據時(shí)代，除了媒體信息以外，商品在各類(lèi)電商平臺的訂單量，用戶(hù)的購買(mǎi)評論也都對后續的消費者產(chǎn)生很大的影響。商家的產(chǎn)品設計者需要匯總統計和分析各類(lèi)平臺的數據做為依據，決定后續的產(chǎn)品發(fā)展，公司的公關(guān)和市場(chǎng)部門(mén)也需要根據輿情作出相應的及時(shí)處理，而這一切也意味著(zhù)傳統的輿情系統升級成為大數據輿情采集和分析系統。
分析完輿情場(chǎng)景后，我們再來(lái)具體細化看下大數據輿情系統，對我們的數據存儲和計算系統提出哪些需求：
- 海量原始數據的實(shí)時(shí)入庫：為了實(shí)現一整套輿情系統，需要有上游原始輸出的采集，也就是爬蟲(chóng)系統。爬蟲(chóng)需要采集各類(lèi)門(mén)戶(hù)，自媒體的網(wǎng)頁(yè)內容。在抓取前需要去重，抓取后還需要分析提取，例如進(jìn)行子網(wǎng)頁(yè)的抓取。
- 始網(wǎng)頁(yè)數據的處理：不論是主流門(mén)戶(hù)還是自媒體的網(wǎng)頁(yè)信息，抓取后我們需要做一定的數據提取，把原始的網(wǎng)頁(yè)內容轉化為結構化數據，例如文章的標題，摘要等，如果是商品點(diǎn)評類(lèi)消息也需要提取有效的點(diǎn)評。
- 結構化數據的輿情分析：當各類(lèi)原始輸出變成結構化的數據后，我們需要有一個(gè)實(shí)時(shí)的計算產(chǎn)品把各類(lèi)輸出做合理的分類(lèi)，進(jìn)一步對分類(lèi)后的內容進(jìn)行情感打標。根據業(yè)務(wù)的需求這里可能會(huì )產(chǎn)生不同的輸出，例如品牌當下是否有熱點(diǎn)話(huà)題，輿情影響力分析，轉播路徑分析，參與用戶(hù)統計和畫(huà)像，輿論情感分析或者是否有重大預警。
- 輿情分析系統中間和結果數據的存儲，交互分析查詢(xún)：從網(wǎng)頁(yè)原始數據清洗到最終的輿情報表這中間會(huì )產(chǎn)生很多類(lèi)型的數據。這些數據有的會(huì )提供給數據分析同學(xué)進(jìn)行輿情分析系統的調優(yōu)，有的數據會(huì )提供給業(yè)務(wù)部門(mén)根據輿情結果進(jìn)行決策。這些查詢(xún)可能會(huì )很靈活，需要我們的存儲系統具備全文檢索，多字段組合靈活的交互分析能力。
- 重大輿情事件的實(shí)時(shí)預警：對于輿情的結果除了正常的搜索和展示需求以外，當有重大事件出現我們需要能做到實(shí)時(shí)的預警。
我們計劃分兩篇介紹完整的輿情新架構，第一篇主要是提供架構設計，會(huì )先介紹時(shí)下主流的大數據計算架構，并分析一些優(yōu)缺點(diǎn)，然后引入輿情大數據架構。第二篇會(huì )有完整的數據庫表設計和部分示例代碼。大家敬請期待。
系統設計
需求分析
結合文章開(kāi)頭對輿情系統的描述，海量大數據輿情分析系統流程圖大體如下：
圖1 輿情系統業(yè)務(wù)流程
原始網(wǎng)頁(yè)存儲庫，這個(gè)庫需要能支持海量數據，低成本，低延時(shí)寫(xiě)入。網(wǎng)頁(yè)數據寫(xiě)入后，要做實(shí)時(shí)結構化提取，提取出來(lái)的數據再進(jìn)行降噪，分詞，圖片ocr處理等。對分詞文本，圖片進(jìn)行情感識別產(chǎn)生輿情數據結果集。傳統的離線(xiàn)全量計算很難滿(mǎn)足輿情系統的時(shí)效性需求。
計算引擎在做數據處理時(shí)，可能還需要從存儲庫中獲取一些元數據，例如用戶(hù)信息，情感詞元數據信息等。
除了實(shí)時(shí)的計算鏈路，對存量數據定期要做一些聚類(lèi)，優(yōu)化我們的情感詞識別庫，或者上游根據業(yè)務(wù)需要觸發(fā)情感處理規則更新，根據新的情感打標庫對存量數據做一次輿情計算。
輿情的結果數據集有不同類(lèi)的使用需求。對于重大輿情，需要做實(shí)時(shí)的預警。完整的輿情結果數據展示層需要支持全文檢索，靈活的屬性字段組合查詢(xún)。業(yè)務(wù)上可能根據屬性字段中的置信度，輿情時(shí)間，或者關(guān)鍵詞組合進(jìn)行分析。
根據前面的介紹，輿情大數據分析系統需要兩類(lèi)計算，一類(lèi)是實(shí)時(shí)計算包括海量網(wǎng)頁(yè)內容實(shí)時(shí)抽取，情感詞分析并進(jìn)行網(wǎng)頁(yè)輿情結果存儲。另一類(lèi)是離線(xiàn)計算，系統需要對歷史數據進(jìn)行回溯，結合人工標注等方式優(yōu)化情感詞庫，對一些實(shí)時(shí)計算的結果進(jìn)行矯正等。所以在系統設計上，需要選擇一套既可以做實(shí)時(shí)計算又能做批量離線(xiàn)計算的系統。在開(kāi)源大數據解決方案中，Lambda架構恰好可以滿(mǎn)足這些需求，下面我們來(lái)介紹下Lambda的架構。
Lambda架構 （wiki）
圖2 Lambda架構圖
Lambda架構可以說(shuō)是Hadoop，Spark體系下最火的大數據架構。這套架構的最大優(yōu)勢就是在支持海量數據批量計算處理（也就是離線(xiàn)處理）同時(shí)也支持流式的實(shí)時(shí)處理（即熱數據處理）。
具體是如何實(shí)現的呢，首先上游一般是一個(gè)隊列服務(wù)例如kafka，實(shí)時(shí)存儲數據的寫(xiě)入。kafka隊列會(huì )有兩個(gè)訂閱者，一個(gè)是全量數據即圖片中上半部分，全量數據會(huì )被存儲在類(lèi)似HDFS這樣的存儲介質(zhì)上。當有離線(xiàn)計算任務(wù)到來(lái)，計算資源（例如Hadoop）會(huì )訪(fǎng)問(wèn)存儲系統上的全量數據，進(jìn)行全量批計算的處理邏輯。經(jīng)過(guò)map/reduce環(huán)節后全量的結果會(huì )被寫(xiě)入一個(gè)結構化的存儲引擎例如Hbase中，提供給業(yè)務(wù)方查詢(xún)。隊列的另一個(gè)消費訂閱方是流計算引擎，流計算引擎往往會(huì )實(shí)時(shí)的消費隊列中的數據進(jìn)行計算處理，例如Spark Streaming實(shí)時(shí)訂閱Kafka的數據，流計算結果也會(huì )寫(xiě)入一個(gè)結構化數據引擎。批量計算和流計算的結果寫(xiě)入的結構化存儲引擎即上圖標注3的"Serving Layer"，這一層主要提供結果數據的展示和查詢(xún)。
在這套架構中，批量計算的特點(diǎn)是需要支持處理海量的數據，并根據業(yè)務(wù)的需求，關(guān)聯(lián)一些其他業(yè)務(wù)指標進(jìn)行計算。批量計算的好處是計算邏輯可以根據業(yè)務(wù)需求靈活調整，同時(shí)計算結果可以反復重算，同樣的計算邏輯多次計算結果不會(huì )改變。批量計算的缺點(diǎn)是計算周期相對較長(cháng)，很難滿(mǎn)足實(shí)時(shí)出結果的需求，所以隨著(zhù)大數據計算的演進(jìn)，提出了實(shí)時(shí)計算的需求。實(shí)時(shí)計算在Lambda架構中是通過(guò)實(shí)時(shí)數據流來(lái)實(shí)現，相比批處理，數據增量流的處理方式?jīng)Q定了數據往往是最近新產(chǎn)生的數據，也就是熱數據。正因為熱數據這一特點(diǎn)，流計算可以滿(mǎn)足業(yè)務(wù)對計算的低延時(shí)需求，例如在輿情分析系統中，我們往往希望輿情信息可以在網(wǎng)頁(yè)抓取下來(lái)后，分鐘級別拿到計算結果，給業(yè)務(wù)方充足的時(shí)間進(jìn)行輿情反饋。下面我們就來(lái)具體看一下，基于Lambda架構的思想如何實(shí)現一套完整的輿情大數據架構。
開(kāi)源輿情大數據方案
通過(guò)這個(gè)流程圖，讓我們了解了整個(gè)輿情系統的建設過(guò)程中，需要經(jīng)過(guò)不同的存儲和計算系統。對數據的組織和查詢(xún)有不同的需求。在業(yè)界基于開(kāi)源的大數據系統并結合Lambda架構，整套系統可以設計如下：
圖3 開(kāi)源輿情架構圖
系統的最上游是分布式的爬蟲(chóng)引擎，根據抓取任務(wù)抓取訂閱的網(wǎng)頁(yè)原文內容。爬蟲(chóng)會(huì )把抓取到的網(wǎng)頁(yè)內容實(shí)時(shí)寫(xiě)入Kafka隊列，進(jìn)入Kafka隊列的數據根據前面描述的計算需求，會(huì )實(shí)時(shí)流入流計算引擎（例如Spark或者Flink），也會(huì )持久化存儲在Hbase，進(jìn)行全量數據的存儲。全量網(wǎng)頁(yè)的存儲可以滿(mǎn)足網(wǎng)頁(yè)爬取去重，批量離線(xiàn)計算的需求。
- 流計算會(huì )對原始網(wǎng)頁(yè)進(jìn)行結構化提取，將非結構化網(wǎng)頁(yè)內容轉化為結構數據并進(jìn)行分詞，例如提取出網(wǎng)頁(yè)的標題，作者，摘要等，對正文和摘要內容進(jìn)行分詞。提取和分詞結果會(huì )寫(xiě)回Hbase。結構化提取和分詞后，流計算引擎會(huì )結合情感詞庫進(jìn)行網(wǎng)頁(yè)情感分析，判斷是否有輿情產(chǎn)生。
- 流計算引擎分析的輿情結果存儲Mysql或者Hbase數據庫中，為了方便結果集的搜索查看，需要把數據同步到一個(gè)搜索引擎例如Elasticsearch，方便進(jìn)行屬性字段的組合查詢(xún)。如果是重大的輿情時(shí)間，需要寫(xiě)入Kafka隊列觸發(fā)輿情報警。
- 全量的結構化數據會(huì )定期通過(guò)Spark系統進(jìn)行離線(xiàn)計算，更新情感詞庫或者接受新的計算策略重新計算歷史數據修正實(shí)時(shí)計算的結果。
開(kāi)源架構分析
上面的輿情大數據架構，通過(guò)Kafka對接流計算，Hbase對接批計算來(lái)實(shí)現Lambda架構中的“batch view”和“real-time view”，整套架構還是比較清晰的，可以很好的滿(mǎn)足在線(xiàn)和離線(xiàn)兩類(lèi)計算需求。但是把這一套系統應用在生產(chǎn)并不是一件容易的事情，主要有下面一些原因。
整套架構涉及到非常多的存儲和計算系統包括：Kafka，Hbase，Spark，Flink，Elasticsearch。數據會(huì )在不同的存儲和計算系統中流動(dòng)，運維好整套架構中的每一個(gè)開(kāi)源產(chǎn)品都是一個(gè)很大的挑戰。任何一個(gè)產(chǎn)品或者是產(chǎn)品間的通道出現故障，對整個(gè)輿情分析結果的時(shí)效性都會(huì )產(chǎn)生影響。
為了實(shí)現批計算和流計算，原始的網(wǎng)頁(yè)需要分別存儲在Kafka和Hbase中，離線(xiàn)計算是消費hbase中的數據，流計算消費Kafka的數據，這樣會(huì )帶來(lái)存儲資源的冗余，同時(shí)也導致需要維護兩套計算邏輯，計算代碼開(kāi)發(fā)和維護成本也會(huì )上升。
輿情的計算結果存儲在Mysql或者Hbase，為了豐富組合查詢(xún)語(yǔ)句，需要把數據同步構建到Elasticsearch中。查詢(xún)的時(shí)候可能需要組合Mysql和Elasticsearch的查詢(xún)結果。這里沒(méi)有跳過(guò)數據庫，直接把結果數據寫(xiě)入Elasticsearch這類(lèi)搜索系統，是因為搜索系統的數據實(shí)時(shí)寫(xiě)入能力和數據可靠性不如數據庫，業(yè)界通常是把數據庫和搜索系統整合，整合下的系統兼備了數據庫和搜索系統的優(yōu)勢，但是兩個(gè)引擎之間數據的同步和跨系統查詢(xún)對運維和開(kāi)發(fā)帶來(lái)很多額外的成本。
新的大數據架構Lambda plus
通過(guò)前面的分析，相信大家都會(huì )有一個(gè)疑問(wèn)，有沒(méi)有簡(jiǎn)化的的大數據架構，在可以滿(mǎn)足Lambda對計算需求的假設，又能減少存儲計算以及模塊的個(gè)數呢。Linkedin的Jay Kreps提出了Kappa架構，關(guān)于Lambda和Kappa的對比可以參考"云上大數據方案"這篇，這里不展開(kāi)詳細對比，簡(jiǎn)單說(shuō)下，Kappa為了簡(jiǎn)化兩份存儲，取消了全量的數據存儲庫，通過(guò)在Kafka保留更長(cháng)日志，當有回溯重新計算需求到來(lái)時(shí)，重新從隊列的頭部開(kāi)始訂閱數據，再一次用流的方式處理Kafka隊列中保存的所有數據。這樣設計的好處是解決了需要維護兩份存儲和兩套計算邏輯的痛點(diǎn)，美中不足的地方是隊列可以保留的歷史數據畢竟有限，難以做到無(wú)時(shí)間限制的回溯。分析到這里，我們沿著(zhù)Kappa針對Lambda的改進(jìn)思路，向前多思考一些：假如有一個(gè)存儲引擎，既滿(mǎn)足數據庫可以高效的寫(xiě)入和隨機查詢(xún)，又能像隊列服務(wù)，滿(mǎn)足先進(jìn)先出，是不是就可以把Lambda和Kappa架構揉合在一起，打造一個(gè)Lambda plus架構呢？
新架構在Lambda的基礎上可以提升以下幾點(diǎn)：
- 在支持流計算和批計算的同時(shí)，讓計算邏輯可以復用，實(shí)現“一套代碼兩類(lèi)需求”。
- 統一歷史數據全量和在線(xiàn)實(shí)時(shí)增量數據的存儲，實(shí)現“一份存儲兩類(lèi)計算”。
- 為了方便輿情結果查詢(xún)需求，“batch view”和“real-time view”存儲在既可以支持高吞吐的實(shí)時(shí)寫(xiě)入，也可以支持多字段組合搜索和全文檢索。
總結起來(lái)就是整套新架構的核心是解決存儲的問(wèn)題，以及如何靈活的對接計算。我們希望整套方案是類(lèi)似下面的架構：
圖4 Lambda Plus架構
- 數據流實(shí)時(shí)寫(xiě)入一個(gè)分布式的數據庫，借助于數據庫查詢(xún)能力，全量數據可以輕松的對接批量計算系統進(jìn)行離線(xiàn)處理。
- 數據庫通過(guò)數據庫日志接口，支持增量讀取，實(shí)現對接流計算引擎進(jìn)行實(shí)時(shí)計算。
- 批計算和流計算的結果寫(xiě)回分布式數據庫，分布式數據庫提供豐富的查詢(xún)語(yǔ)意，實(shí)現計算結果的交互式查詢(xún)。
整套架構中，存儲層面通過(guò)結合數據庫主表數據和數據庫日志來(lái)取代大數據架構中的隊列服務(wù)，計算系統選取天然支持批和流的計算引擎例如Flink或者Spark。這樣一來(lái)，我們既可以像Lambda進(jìn)行無(wú)限制的歷史數據回溯，又可以像Kappa架構一樣一套邏輯，存儲處理兩類(lèi)計算任務(wù)。這樣的一套架構我們取名為“Lambda plus”，下面就詳細展開(kāi)如何在阿里云上打造這樣的一套大數據架構。
云上輿情系統架構
在阿里云眾多存儲和計算產(chǎn)品中，貼合上述大數據架構的需求，我們選用兩款產(chǎn)品來(lái)實(shí)現整套輿情大數據系統。存儲層面使用阿里云自研的分布式多模型數據庫Tablestore，計算層選用Blink來(lái)實(shí)現流批一體計算。
圖5 云上輿情大數據架構
這套架構在存儲層面，全部基于Tablestore，一個(gè)數據庫解決不同存儲需求，根據之前輿情系統的介紹，網(wǎng)頁(yè)爬蟲(chóng)數據在系統流動(dòng)中會(huì )有四個(gè)階段分別是原始網(wǎng)頁(yè)內容，網(wǎng)頁(yè)結構化數據，分析規則元數據和輿情結果，輿情結果索引。我們利用Tablestore寬行和schema free的特性，合并原始網(wǎng)頁(yè)和網(wǎng)頁(yè)結構化數據成一張網(wǎng)頁(yè)數據。網(wǎng)頁(yè)數據表和計算系統通過(guò)Tablestore新功能通道服務(wù)進(jìn)行對接。通道服務(wù)基于數據庫日志，數據的組織結構按照數據的寫(xiě)入順序進(jìn)行存儲，正是這一特性，賦能數據庫具備了隊列流式消費能力。使得存儲引擎既可以具備數據庫的隨機訪(fǎng)問(wèn)，也可以具備隊列的按照寫(xiě)入順序訪(fǎng)問(wèn)，這也就滿(mǎn)足我們上面提到整合Lambda和kappa架構的需求。分析規則元數據表由分析規則，情感詞庫組層，對應實(shí)時(shí)計算中的維表。
計算系統這里選用阿里云實(shí)時(shí)流計算產(chǎn)品Blink，Blink是一款支持流計算和批計算一體的實(shí)時(shí)計算產(chǎn)品。并且類(lèi)似Tablestore可以很容易的做到分布式水平擴展，讓計算資源隨著(zhù)業(yè)務(wù)數據增長(cháng)彈性擴容。使用Tablestore + Blink的優(yōu)勢有以下幾點(diǎn)：
- Tablestore已經(jīng)深度和Blink進(jìn)行整合，支持源表，維表和目的表，業(yè)務(wù)無(wú)需為數據流動(dòng)開(kāi)發(fā)代碼。
- 整套架構大幅降低組建個(gè)數，從開(kāi)源產(chǎn)品的6～7個(gè)組建減少到2個(gè)，Tablestore和Blink都是全托管0運維的產(chǎn)品，并且都能做到很好的水平彈性，業(yè)務(wù)峰值擴展無(wú)壓力，使得大數據架構的運維成本大幅降低。
- 業(yè)務(wù)方只需要關(guān)注數據的處理部分邏輯，和Tablestore的交互邏輯都已經(jīng)集成在Blink中。
- 開(kāi)源方案中，如果數據庫源希望對接實(shí)時(shí)計算，還需要雙寫(xiě)一個(gè)隊列，讓流計算引擎消費隊列中的數據。我們的架構中數據庫既作為數據表，又是隊列通道可以實(shí)時(shí)增量數據消費。大大簡(jiǎn)化了架構的開(kāi)發(fā)和使用成本。
- 流批一體，在輿情系統中實(shí)時(shí)性是至關(guān)重要的，所以我們需要一個(gè)實(shí)時(shí)計算引擎，而B(niǎo)link除了實(shí)時(shí)計算以外，也支持批處理Tablestore的數據， 在業(yè)務(wù)低峰期，往往也需要批量處理一些數據并作為反饋結果寫(xiě)回Tablestore，例如情感分析反饋等。那么一套架構既可以支持流處理又可以支持批處理是再好不過(guò)。這里我們可以參考之前的一篇文章《實(shí)時(shí)計算最佳實(shí)踐：基于表格存儲和Blink的大數據實(shí)時(shí)計算》。一套架構帶來(lái)的優(yōu)勢是，一套分析代碼既可以做實(shí)時(shí)流計算又可以離線(xiàn)批處理。
https://yq.aliyun.com/articles/692526
整個(gè)計算流程會(huì )產(chǎn)生實(shí)時(shí)的輿情計算結果。重大輿情事件的預警，通過(guò)Tablestore和函數計算觸發(fā)器對接來(lái)實(shí)現。Tablestore和函數計算做了增量數據的無(wú)縫對接，通過(guò)結果表寫(xiě)入事件，可以輕松的通過(guò)函數計算觸發(fā)短信或者郵件通知。完整的輿情分析結果和展示搜索利用了Tablestore的新功能多元索引，徹底解決了開(kāi)源Hbase+Solr多引擎的痛點(diǎn)：
運維復雜，需要有運維hbase和solr兩套系統的能力，同時(shí)還需要維護數據同步的鏈路。
Solr數據一致性不如Hbase，在Hbase和Solr數據語(yǔ)意并不是完全一致，加上Solr/Elasticsearch在數據一致性很難做到像數據庫那么嚴格。在一些極端情況下會(huì )出現數據不一致的問(wèn)題，開(kāi)源方案也很難做到跨系統的一致性比對。
查詢(xún)接口需要維護兩套API，需要同時(shí)使用Hbase client和Solr client，索引中沒(méi)有的字段需要主動(dòng)反查Hbase，易用性較差。
參考文獻
1、Lambda大數據架構
https://mapr.com/tech-briefs/stream-processing-mapr/
2、Kappa大數據架構
https://www.oreilly.com/ideas/questioning-the-lambda-architecture
3、Lambda和Kappa架構對比
https://www.ericsson.com/en/blog/2015/11/data-processing-architectures--lambda-and-kappa
總結
本文基于《百億級全網(wǎng)輿情分析系統存儲設計》并結合Tablestore的新功能做了現代大數據輿情系統的架構升級，實(shí)現了海量信息下的實(shí)時(shí)輿情分析存儲系統。也介紹了開(kāi)源方案，并和我們的方案做了詳細的對比。
