flink實時採集ftp

㈠ 大數據方面核心技術有哪些

大數據技術的體系龐大且復雜，基礎的技術包含數據的採集、數據預處理、分布式存儲、資料庫、數據倉庫、機器學習、並行計算、可視化等。

1、數據採集與預處理：

Flume NG實時日誌收集系統，支持在日誌系統中定製各類數據發送方，用於收集數據；

Zookeeper是一個分布式的，開放源碼的分布式應用程序協調服務，提供數據同步服務。

2、數據存儲：

Hadoop作為一個開源的框架，專為離線和大規模數據分析而設計，HDFS作為其核心的存儲引擎，已被廣泛用於數據存儲。

HBase，是一個分布式的、面向列的開源資料庫，可以認為是hdfs的封裝，本質是數據存儲、Nosql資料庫。

3、數據清洗：MapRece作為Hadoop的查詢引擎，用於大規模數據集的並行計算

4、數據查詢分析：

Hive的核心工作就是把SQL語句翻譯成MR程序，可以將結構化的數據映射為一張資料庫表，並提供 HQL(Hive SQL)查詢功能。

Spark 啟用了內存分布數據集，除了能夠提供互動式查詢外，它還可以優化迭代工作負載。

5、數據可視化：對接一些BI平台，將分析得到的數據進行可視化，用於指導決策服務。

㈡ 流批一體不只有Flink，還有實時數據模型

通常來講，數據倉庫的建設，都是以離線作為主要的密報，下游的應用，不論是報表還是介面，所提供的數據也大多是T-1時效性。

但伴隨著業務的變化，當離線做到沒什麼可以繼續做的時候，實時就會被拿出來，作為新一個階段的目標進行攻克。

在流批一體建設之前，這種實時訴求通常會開發成分鍾級的任務，通過近實時的方案來解決業務的問題，但分鍾級會帶來諸如任務過多、資源擠占較大、無法支持復雜邏輯等問題。

因此專門支持實時計算的框架，比如早期的Storm，能夠嘗試從純實時的角度解決業務問題，就被拿出來作為嘗試。然而Storm的局限性也很大，因為那會的任務開發只能通過Java的方式來進行，與Hive所推崇的純SQL方案相比，上手難度大了不少，同時兩套代碼的邏輯幾乎沒有可比性，這種方案也就一直沒有什麼聲音。

盡管實時技術有各種缺陷，但作為一種能夠很容易講清楚價值的項目，同時又非常便於向上匯報的技術方案，實時技術還是被或多或少的做了起來。在大多數的公司里，實時和離線就會有不同的團隊進行維護，或者是同一個團隊，但分成不同的項目來執行。這個階段，優先高效的把業務做起來，哪怕場景再簡單，但能夠證明實時有價值和前景，這個階段的目標就算完成了。

以上的各種方案，難免會帶來三個特別難以解決的問題：

（1）數據的口徑上，實時和離線很容易不統一；
（2）數據模型的規范上，實時和離線也往往是分開建設；
（3）即便是同一種口徑和同一種規范，實時和離線也要分成兩套代碼來維護。

這三個問題短時間內會被高速發展掩蓋掉，但當業務對實時的訴求越來越多、壓力越來越大的時候，口徑和代碼的不統一，就會越來越成為阻礙敏捷開發的障礙，需要有方案進行解決。

後來Flink出現了，帶來了流批一體的全新方案，這個問題便出現了解決的曙光，這也比較接近我們對於實時計算的理想方案，因為其意義堪比Hive，也成為了各個大廠面試的標配問題。

然而，僅僅學會Flink是不夠的，因為流批一體帶來的並不僅僅是技術方案或者是框架的改變，同樣帶來了數據模型的改變，這就要求我們從數據模型上，而不是技術方案上，來制定我們的實時方案。

那麼我們如何理解「實時數據模型」這件事情呢？

通常而言，我們關心的內容，包括如下幾個方面：

（1）實時數據源與離線數據源存在差異，導致相同的欄位，取值或者類型會存在不相等的情況；
（2）實時和離線由於底層執行機制的不同，通常需要維護兩套代碼，會帶來諸如口徑不統一、質量檢測難的問題；
（3）產品邏輯變化較快時，離線模型修改相對容易，但實時模型需要考慮壓測、削峰、重啟等技術問題，維護成本非常高昂。

數據倉庫之所以能夠普及並被業務接受，正是因為其模型能夠屏蔽掉底層差異的問題，並且有相對可靠的數據質量監控方法，並且變更成本非常低。而實時數倉如果想要替代掉離線數倉，以上的問題通常是需要一些模型設計甚至是平台工具的來解決，這些問題解決的重要性，並不比Flink弱。

我們先從比較可控的模型層面說起。

在離線的概念里，數倉模型設計成了DWD/DWS/ADS三個層級，原本的概念是DWD面向事實表的構建，DWS面向公共指標的統一，ADS負責靈活的口徑變化問題。

在離線的概念里，DWD/DWS/ADS三個層級需要保留，但負責的目標會有一些變化，同時還需要增加存儲統一層，也就是以TiDB/Holo為代表的資料庫，來承擔服務分析一體化的訴求。

讓我們先看DWD層，DWD承擔了屏蔽實時離線鏈路差異的問題，最重要的作用是保證表結構的統一及欄位內容的對齊。DWD最重要的意義，是保證離線表和實時表，其表結構和欄位概念是相同的。

為什麼這么強調？試想一下，在離線場景下，我們可以在DWD上靈活的增加各種統計標簽，或者是將維度退化到事實表，都是一些left join或者是服務端直接打標可以解決的事情。但在實時場景下，這會變成多流join或者是緩存等更復雜的技術場景，導致這些信息並不能有效的記錄到DWD，因此DWD的設計就要產生一些變化，有一些內容在實時場景下無法准確記錄，這一類信息需要標識到對應的欄位描述上，下游使用時才不會出錯。

同時，實時和離線存儲數據的介質，也必然有一些區別。例如離線可以存在HDFS上，實時則可能視情況保存在資料庫、HDFS甚至是內存中，這時候對於欄位格式、讀取方式都會有差異，設計表時其約束條件也會更多。

因而，DWD更多承擔了邏輯統一的職責，依舊以事實表為基礎，但約束條款要比離線更多。

再看一下DWS層，離線上DWS是負責口徑統一的重要一環，將通用的維度和口徑計算方法抽象出現，以提供跨數據域的靈活使用。但在實時場景下，這一類的維護收益通常都比較低，不僅因為實時只看當天的數據，也是因為實時本身的維度難度就較大，多一層模型其收益會急速下降，因而大多數時候會忽略掉DWS的建設，ADS直接引用DWD進行統計。

然而，DWS畢竟存儲的內容要比DWD少很多，因此如果計算資源瓶頸非常明顯，或者是業務場景不需要分析實時明細數據的情況下，或者是DWD的下游引用過多時，DWS可以承擔削峰的重任，通過減少數據量以應對大促等場景，還是有一定意義的。

接下來就是最重要的ADS層，在這一層上，邏輯統一、口徑統一、大促削峰在前置模型上都得到了一定程度的解決，ADS則像離線一樣承擔了應對需求變化的重任。

但ADS所面臨的情況和離線還是有所不同的，因為ADS的任務啟動，不僅要啟動一個離線的跑批任務，還要同時啟動一個實時的流式任務，而ADS往往會同時統計離線+實時的結果，以應對同比、環比等場景。

這時候很多具體Case要具體分析了，因為特定場景的坑會非常多。例如最常見的「同比」，要對比今年和去年的結果變化，離線往往會統計分小時的結果，但實時會累計起始時刻到當期時刻的結果，因而當一個小時沒有結束的時候，這個同比的波動變化會非常大，給人一種「數據是錯誤的」印象，新手很容易踩這個坑，從而被業務質疑。

因此，針對累計統計指標，從代碼設計上就要考慮到這種情況，都根據時間欄位統計起始到當前時刻的結果的，在代碼邏輯上會要求一些統計技巧。

很多時候，因為業務指標變化太快，改實時代碼是來不及的，這時候一部分的工作量甚至需要報表工具的數據集來解決，改動查詢sql，要比改動任務來的快捷多了。但這部分的能力，其實是依賴於存儲工具的，個人認為可以分到存儲統一層來解決。

最後是存儲統一層，因為一些特殊的場景，比如實時分析明細數據，或者是不確定時間周期的多天統計結果，如果依賴Flink SQL來解決是有些不現實的，因而這部分的壓力需要資料庫來承擔。

簡單講，就是將明細做輕度的匯總後，直接寫到資料庫，實時更新，下游自定義條件，並直接讀庫統計結果。這種場景既要求資料庫有OLAP的計算能力，也要有OLTP的穩定特點，因而TiDB和Holo這一類HTAP的引擎就變得非常重要。

因為多了實時的部分，因此過去面向離線的開發工具，也需要有一些特定的改造，以適應實時的開發和運維訴求。

對於開發工具而言，其目標集中在四個場景上：元數據定義與獲取、數據建模、開發與測試、運維與監控。

其次講數據建模，因為建模的理論已經穩定了有些年頭了，絕大多數場景下都是按照既定的方案來執行。過去離線當道時，規范執行的弱一些不是什麼大問題，但流批一體當道的年代，規范是需要強約束的，這就對了開發工具提出了一定的要求，是否能夠從平台層面上對規范進行內置，並以此來約束開發的同學，降低不規范模型對後期維護帶來的壓力。

這種建模能力的代表有兩種，一種是規范表的命名，填寫相應的分層+主題域+數據域+統計刷新方式，從源頭上規范表的目標和作用；一種是規范指標的定義和使用，例如原子指標還是派生指標，統計周期多少，業務限定用語如何規范，統計粒度怎麼填寫。

在實際開發中，通過工具的限制，如果規范可以做的好，代碼是可以自動生成出來的。當然，以上的功能，都屬於通過犧牲開發效率，來提升數據質量的范疇，使用時需要根據團隊的情況來限定。

再次是開發和測試，這是平台提供的最重要的能力。在開發層面，就是代碼的預編譯能力+發布功能。預編譯不僅要檢查代碼的邏輯是否正確，同時對於代碼中依賴的其他數據源，獲取到的元數據信息是否准確，至少欄位的命名不會有大的問題。當代碼預編譯通過，發布上線後，還需要檢測當前是否有資源支持任務啟動，並且上游的消息隊列是否是啟動的狀態。

實時的測試一直都是比較大的問題，它不像離線可以啟動一個SQL任務看看結果，實時在每個階段的輸入和輸出，是需要通過平台支持的日誌列印功能來進行輔助的。很多時候我們會新建一個測試專用的topic來測試結果，但對於流量較大的線上任務而言，這種方式無法像離線區分Dev環境一樣，能夠對資源進行隔離，因而如果能夠支持圈定數據的輸入和列印輸出，對於測試的效率而言無疑是最佳的。

最後要提到的是運維與監控能力。運維能力是指根據輸入的RPS，或者是cu使用情況，或者是任務的整體延遲，提供相應的參數調優能力，通過參數來調整任務的執行情況。並且能夠根據以上指標的變化，自定義相應的閾值，提供相應的告警能力，通過簡訊或者是消息工具的方式觸達任務維護者。

實時與離線有一些不同的是，離線可以通過增加一個監控節點的方式，通過group by判斷數據是否重復，而實時任務則非常依賴Flink自身的一致性能力，因而發現和解決問題的成本更高。

其實做到運維這個環節，對人的要求其實是更高的。因為流批一體在運維上會帶來一個好處，即實時任務和離線任務能夠錯峰執行，實時在白天壓力大，而離線在晚上壓力大。但同樣的，這種方式對於維護者而言更加痛苦，因為不僅晚上要熬夜值班，白天同樣不能休息，在大促期間甚至需要輪班來維護任務，可以說是「匯報一時爽，痛苦長相伴」。

從遠處來看，流任務和批任務，在自身的機制上就存在非常大的差異，批程序面上的是特定時間內相對靜態的數據，而流程序處理的則是change-log，雖然有可能數據在表結構層面，通過數據模型的設計來保持一致，但是在語義層面，其根本還是不一樣的。這一點可能是最制約批流一體發展的問題，也是最難實現統一或者永遠也不可能統一的。

綜上，對於實時模型，開發工具需要將監控實時部分的能力進行補全，就像DWD層需要分別維護實時和離線兩套架構一樣，開發工具也需要分別維護兩套架構的結果，因而現階段的實時開發，還做不到降低維護和開發的成本，只能減輕其中部分環節的工作量。

以上講了很長時間的實時模型，但從實際的效果上看，業務並不會感知到多麼明顯的技術變化，相反會有一種「面子工程」的感覺在裡面。

當然，我並不否認實時的價值，在「搜廣推」這三個技術佔主導的領域內，作用還是很大的。但實時畢竟要比離線的內容，更加的難以理解，出現問題的排查成本也更高。這種復雜性使得我們在應對變化時，往往做不出有效的應對，就會變得特別被動。

因而，說一句事後的話，就是「實時的價值取決於業務方，而不是技術方」。只有業務對實時痛點強烈的場景下，我們做如此復雜的研究和應對，才能體現出自己的價值，更多的時候，是在「王婆賣瓜，自賣自誇」。有這種投入，還不如多招幾個分析師更靠譜和實在。

本人之前的文章《天下數據，唯快不破》，重點強調了一個「快」字。但「天下熙熙皆為利來，天下攘攘皆為利往」，這個快更多的是在講應對「變化」的快，而不是「技術」自己的快。

所以，為了以後的職業發展，我們要跟進實時技術的變化，但從自身的工作角度出發，如何應對業務的變化，才是自己要關心的課題。

㈢ 基於flink sql構建實時數據倉庫

根據目前大數據這一塊的發展，已經不局限於離線的分析，挖掘數據潛在的價值，數據的時效性最近幾年變得剛需，實時處理的框架有storm，spark-streaming，flink等。想要做到實時數據這個方案可行，需要考慮以下幾點：1、狀態機制 2、精確一次語義 3、高吞吐量 4、可彈性伸縮的應用 5、容錯機制，剛好這幾點，flink都完美的實現了，並且支持flink sql高級API，減少了開發成本，可用實現快速迭代，易維護等優點。

離線數倉的架構圖：

實時數倉架構圖:

目前是將實時維度表和DM層數據存於hbase當中，實時公共層都存於kafka當中，並且以寫滾動日誌的方式寫入HDFS（主要是用於校驗數據）。其實在這里可以做的工作還有很多，kafka集群，flink集群，hbase集群相互獨立，這對整個實時數據倉庫的穩定性帶來一定的挑戰。

一個數據倉庫想要成體系，成資產，離不開數據域的劃分。所以參考著離線的數據倉庫，想著在實時數倉做出這方面的探索，理論上來講，離線可以實現的，實時也是可以實現的。 並且目前已經取得了成效，目前劃分的數據域跟離線大致相同，有流量域，交易域，營銷域等等。當然這裡面涉及到維表，多事務事實表，累計快照表，周期性快照表的設計，開發，到落地這里就不詳述了。

維度表也是整個實時數據倉庫不可或缺的部分。從目前整個實時數倉的建設來看，維度表有著數據量大，但是變更少的特點，我們試想過構建全平台的實時商品維度表或者是實時會員維度表，但是這類維度表太過於復雜，所以針對這類維度表下面介紹。還有另外一種就是較為簡單的維度表，這類維度可能對應著業務系統單個mysql表，或者只需要幾個表進行簡單ETL就可以產出的表，這類維表是可以做成實時的。以下有幾個實施的關鍵點：

如下是離線數據同步架構圖：

實時數據的接入其實在底層架構是一樣的，就是從kafka那邊開始不一樣，實時用flink的UDTF進行解析，而離線是定時（目前是小時級）用camus拉到HDFS，然後定時load HDFS的數據到hive表裡面去，這樣來實現離線數據的接入。實時數據的接入是用flink解析kafka的數據，然後在次寫入kafka當中去。

由於目前離線數據已經穩定運行了很久，所以實時接入數據的校驗可以對比離線數據，但是離線數據是小時級的hive數據，實時數據存於kafka當中，直接比較不了，所以做了相關處理，將kafka的數據使用flink寫HDFS滾動日誌的形式寫入HDFS，然後建立hive表小時級定時去load HDFS中的文件，以此來獲取實時數據。

完成以上兩點，剩餘還需要考慮一點，都是小時級的任務，這個時間卡點使用什麼欄位呢?首先要確定一點就是離線和實時任務卡點的時間欄位必須是一致的，不然肯定會出問題。目前離線使用camus從kafka將數據拉到HDFS上，小時級任務，使用nginx_ts這個時間欄位來卡點，這個欄位是上報到nginx伺服器上記錄的時間點。而實時的數據接入是使用flink消費kafka的數據，在以滾動日誌的形式寫入HDFS的，然後在建立hive表load HDFS文件獲取數據，雖然這個hive也是天/小時二級分區，但是離線的表是根據nginx_ts來卡點分區，但是實時的hive表是根據任務啟動去load文件的時間點去區分的分區，這是有區別的，直接篩選分區和離線的數據進行對比，會存在部分差異，應當的做法是篩選范圍分區，然後在篩選nginx_ts的區間，這樣在跟離線做對比才是合理的。

目前實時數據接入層的主要時延是在UDTF函數解析上，實時的UDTF函數是根據上報的日誌格式進行開發的，可以完成日誌的解析功能。

解析流程圖如下：

解析速率圖如下：

該圖還不是在峰值數據量的時候截的，目前以800記錄/second為准，大概一個記錄的解析速率為1.25ms。
目前該任務的flink資源配置核心數為1，假設解析速率為1.25ms一條記錄，那麼峰值只能處理800條/second，如果數據接入速率超過該值就需要增加核心數，保證解析速率。

介紹一下目前離線維度表的情況，就拿商品維度表來說，全線記錄數將近一個億，計算邏輯來自40-50個ods層的數據表，計算邏輯相當復雜，如果實時維度表也參考離線維度表來完成的話，那麼開發成本和維護成本非常大，對於技術來講也是很大的一個挑戰，並且目前也沒有需求要求維度屬性百分百准確。所以目前（偽實時維度表）准備在當天24點產出，當天的維度表給第二天實時公共層使用，即T-1的模式。偽實時維度表的計算邏輯參考離線維度表，但是為了保障在24點之前產出，需要簡化一下離線計算邏輯，並且去除一些不常用的欄位，保障偽實時維度表可以較快產出。

實時維度表的計算流程圖：

目前使用flink作為公司主流的實時計算引擎，使用內存作為狀態後端，並且固定30s的間隔做checkpoint，使用HDFS作為checkpoint的存儲組件。並且checkpoint也是作為任務restart以後恢復狀態的重要依據。熟悉flink的人應該曉得，使用內存作為狀態後端，這個內存是JVM的堆內存，畢竟是有限的東西，使用不得當，OOM是常有的事情，下面就介紹一下針對有限的內存，如果完成常規的計算。

㈣ Flink 原理詳解

Flink 是一個流處理框架，支持流處理和批處理，特點是流處理有限，可容錯，可擴展，高吞吐，低延遲。

流處理是處理一條，立馬下一個節點會從緩存中取出，在下一個節點進行計算

批處理是只有處理一批完成後，才會經過網路傳輸到下一個節點

流處理的優點是低延遲 批處理的優點是高吞吐

flink同時支持兩種，flink的網路傳輸是設計固定的緩存塊為單位，用戶可以設置緩存塊的超時值來決定換存塊什麼時候進行傳輸。 數據大於0 進行處理就是流式處理。
如果設置為無限大就是批處理模型。

Flink 集群包括 JobManager 和 TaskManager .

JobManager 主要負責調度 Job 並協調 Task 做 checkpoint，職責上很像 Storm 的 Nimbus。從 Client 處接收到 Job 和 JAR 包 等資源後，會生成優化後的執行計劃，並以 Task 的單元調度到各個 TaskManager 去執行。

TaskManager 在啟動的時候就設置好了槽位數（Slot），每個 slot 能啟動一個 Task，Task 為線程。從 JobManager 處接收需要 部署的 Task，部署啟動後，與自己的上游建立 Netty 連接，接收數據並處理。

flink on yarn 是由client 提交 app到 RM 上， 然後RM 分配一個 AppMaster負責運行 Flink JobManager 和 Yarn AppMaster, 然後 AppMaster 分配 容器去運行 Flink TaskManger

SparkStreaming 是將流處理分成微批處理的作業， 最後的處理引擎是spark job

Spark Streaming把實時輸入數據流以時間片Δt （如1秒）為單位切分成塊，Spark Streaming會把每塊數據作為一個RDD，並使用RDD操作處理每一小塊數據。每個塊都會生成一個Spark Job處理，然後分批次提交job到集群中去運行，運行每個 job的過程和真正的spark 任務沒有任何區別。

JobScheler, 負責 Job的調度通過定時器每隔一段時間根據Dstream的依賴關系生一個一個DAG圖

ReceiverTracker負責數據的接收，管理和分配
ReceiverTracker在啟動Receiver的時候他有ReceiverSupervisor,其實現是ReceiverSupervisorImpl, ReceiverSupervisor本身啟 動的時候會啟動Receiver，Receiver不斷的接收數據，通過BlockGenerator將數據轉換成Block。定時器會不斷的把Block數據通會不斷的把Block數據通過BlockManager或者WAL進行存儲，數據存儲之後ReceiverSupervisorlmpl會把存儲後的數據的元數據Metadate匯報給ReceiverTracker，其實是匯報給ReceiverTracker中的RPC實體ReceiverTrackerEndpoin

spark on yarn 的cluster模式， Spark client 向RM提交job請求, RM會分配一個 AppMaster, driver 和 運行在AppMAster節點里， AM然後把Receiver作為一個Task提交給Spark Executor 節點， Receive啟動接受數據，生成數據塊，並通知Spark Appmaster, AM會根據數據塊生成相應的Job, 並把Job 提交給空閑的 Executor 去執行。

1：需要關注流數據是否需要進行狀態管理
2：At-least-once或者Exectly-once消息投遞模式是否有特殊要求
3：對於小型獨立的項目，並且需要低延遲的場景，建議使用storm
4：如果你的項目已經使用了spark，並且秒級別的實時處理可以滿足需求的話，建議使用sparkStreaming
5：要求消息投遞語義為 Exactly Once 的場景；數據量較大，要求高吞吐低延遲的場景；需要進行狀態管理或窗口統計的場景，建議使用flink

Flink 提供的Api右 DataStream 和 DataSet ，他們都是不可變的數據集合，不可以增加刪除中的元素， 通過 Source 創建 DataStream 和 DataSet

在創建運行時有：

Flink的每一個Operator稱為一個任務， Operator 的每一個實例稱為子任務，每一個任務在JVM線程中執行。可以將多個子任務鏈接成一個任務，減少上下文切換的開銷，降低延遲。

source 和 運算元map 如果是 one by one 的關系，他們的數據交換可以通過緩存而不是網路通信

TaskManager 為控制執行任務的數量，將計算資源劃分多個slot,每個slot獨享計算資源，這種靜態分配利於任務資源隔離。

同一個任務可以共享一個slot, 不同作業不可以。

這里因為 Source 和 Map 並行度都是4 採用直連方式，他們的數據通信採用緩存形式

所以一共需要兩個TaskManager source,Map 一個，rece一個， 每個TaskManager 要3個slot

JobManager 將 JobGraph 部署 ExecutionGraph

設置的並行度，可以讓一個ExecJobVertex 對應 多個並行的ExecVertex 實例。

Flink通過狀態機管理 ExecGraph的作業執行進度。

Flink 將對象序列化為固定數量的預先分配的內存段，而不是直接把對象放在堆內存上。
Flink TaskManager 是由幾個內部組件組成的：actor 系統（負責與 Flink master 協調）、IOManager（負責將數據溢出到磁碟並將其讀取回來）、MemoryManager（負責協調內存使用。

數據源：

Sink:

時間：

處理時間：取自Operator的機器系統時間

事件時間： 由數據源產生

進入時間： 被Source節點觀察時的系統時間

如果數據源沒有自己正確創建水印，程序必須自己生成水印來確保基於事件的時間窗口可以正常工作。。

DataStream 提供了 周期性水印，間歇式水印，和遞增式水印