Flink在BMC的應(yīng)用與實(shí)踐

1、Flink在BMC的應(yīng)用與實(shí)踐Apache FlinkApache Flink是由Apache軟件基金會(huì)開(kāi)發(fā)的開(kāi)源流處理框架19年1月份，阿里巴巴以 9000 萬(wàn)歐元收購(gòu)Flink母公司 Data Artisans流數(shù)據(jù) Data StreamingSpark與Flink的對(duì)比Apache Flink - 數(shù)據(jù)流上的有狀態(tài)計(jì)算Apache Flink - 同時(shí)支持流式及批量分析應(yīng)用Apache Flink APIWhy SQL/Table API？PyFlinkBMC數(shù)據(jù)平臺(tái)架構(gòu)及技術(shù)棧SwarmNBRadical計(jì)算引擎數(shù)據(jù)倉(cāng)儲(chǔ)WebSocketFlink實(shí)踐（一）數(shù)據(jù)大屏展示雙十一期間各店

2、鋪銷(xiāo)售情況Flink實(shí)踐（一）：數(shù)據(jù)大屏Input：各店鋪后臺(tái)銷(xiāo)售數(shù)據(jù)Output：銷(xiāo)售金額總計(jì)Flink實(shí)踐（一）：數(shù)據(jù)大屏 SourceFlink JobFlink實(shí)踐（一）：數(shù)據(jù)大屏 KafkaSourceFlink JobFlink實(shí)踐（二）需求公司里有一個(gè)美妝經(jīng)銷(xiāo)經(jīng)銷(xiāo)項(xiàng)目，即存在旗艦店的同時(shí)有數(shù)以千計(jì)的經(jīng)銷(xiāo)商店鋪，業(yè)務(wù)同事希望可以通過(guò)技術(shù)手段監(jiān)控經(jīng)銷(xiāo)商店鋪的商品通過(guò)技術(shù)手段監(jiān)控經(jīng)銷(xiāo)商店鋪的商品價(jià)格不低于旗艦店價(jià)格不低于旗艦店，避免影響旗艦店銷(xiāo)售Flink實(shí)踐（二）：經(jīng)銷(xiāo)價(jià)格監(jiān)控Input1：線上店鋪商品頁(yè)面價(jià)格數(shù)據(jù)Input2：線上店鋪商品優(yōu)惠券數(shù)據(jù)Input3：商品控制價(jià)格.csv

3、Output：異常的商品價(jià)格及對(duì)應(yīng)的店鋪（經(jīng)銷(xiāo)商）名稱Flink實(shí)踐（二）：經(jīng)銷(xiāo)價(jià)格監(jiān)控 KafkaSourceCSVSourceInput1：線上店鋪?lái)?yè)面價(jià)格數(shù)據(jù)Input2：商品控制價(jià)格.csv源碼分析1：注冊(cè)環(huán)境2：注冊(cè)Source3：注冊(cè)Sink4：注冊(cè)UDF5：計(jì)算主體6：提交作業(yè)并執(zhí)行Kafka Source 數(shù)據(jù)結(jié)構(gòu)Kafka Source1：連接Kafka配置2：將Kafka中數(shù)據(jù)定義為JSON3：對(duì)數(shù)據(jù)進(jìn)行序列化4：在環(huán)境中注冊(cè)數(shù)據(jù)源表名為KafkaSourceCSV Source 數(shù)據(jù)結(jié)構(gòu)商品價(jià)格參考的 CSV 文件的數(shù)據(jù)樣例為：CSV Source1：聲明CSV文件存放路

4、徑2：將CSV文件格式化為OldCsv格式3：對(duì)數(shù)據(jù)進(jìn)行序列化4：在環(huán)境中注冊(cè)數(shù)據(jù)源表名為CsvSourceCSV Sink 數(shù)據(jù)結(jié)構(gòu)將要輸出的CSV格式CSV Sink1：輸出文件路徑2：輸出字段3：在環(huán)境中注冊(cè)CSV 表UDF輸入和輸出都有了，我們根據(jù)業(yè)務(wù)同事要求的計(jì)算和判斷邏輯，即銷(xiāo)商店鋪的商品價(jià)格不低于旗艦店這個(gè)規(guī)則去編寫(xiě)。根據(jù)編寫(xiě)業(yè)務(wù)的邏輯，Table API 中算子不能滿足所有需求，所以我們需要自定義 UDF 去處理其中幾個(gè)字段，分別為“根據(jù)商品名稱匹配實(shí)際商品”，“根據(jù)商品名稱、商品頁(yè)面價(jià)格，識(shí)別商品容量”，“按需計(jì)算優(yōu)惠價(jià)格”，“格式化優(yōu)惠券信息”等。UDF 的包依賴UDF i

5、dentifyGoods識(shí)別商品名稱，與控價(jià)表的商品名稱對(duì)應(yīng)1：將一個(gè)標(biāo)題轉(zhuǎn)換成set類(lèi)型，方便做交集運(yùn)算2：去掉標(biāo)題中所有中文字符3：先匹配商品名稱之外的字符，并刪除4：找到最為匹配的商品標(biāo)題，否則判定為未知商品5：將該方法注冊(cè)為UDFUDF identifyCapacity根據(jù)商品名稱及頁(yè)面價(jià)格，識(shí)別商品容量UDF get_min_price按需計(jì)算優(yōu)惠價(jià)格1：用裝飾器UDF注冊(cè)UDF2：識(shí)別所有優(yōu)惠券3：若存在優(yōu)惠券信息，計(jì)算最低價(jià)4：返回最低價(jià)UDF formatDiscountInfo格式化優(yōu)惠券信息，以便輸出.csv文件計(jì)算Job主體方法1.1.左輸出表：左輸出表：selectse

6、lect中調(diào)用方法中調(diào)用方法identify_goodsidentify_goods識(shí)別商品名稱識(shí)別商品名稱與基礎(chǔ)信息表對(duì)應(yīng)與基礎(chǔ)信息表對(duì)應(yīng)2.2.左輸出表：左輸出表：selectselect中調(diào)用方法中調(diào)用方法identify_capacityidentify_capacity根據(jù)商品根據(jù)商品名稱與商品標(biāo)價(jià)識(shí)別商品容量名稱與商品標(biāo)價(jià)識(shí)別商品容量3.3.左輸出表：左輸出表：selectselect中調(diào)用方法中調(diào)用方法get_min_priceget_min_price根據(jù)頁(yè)面價(jià)與根據(jù)頁(yè)面價(jià)與優(yōu)惠信息計(jì)算最低價(jià)優(yōu)惠信息計(jì)算最低價(jià)4.4.左輸出表：左輸出表：selectselect中調(diào)用方法中調(diào)用

7、方法format_discount_infoformat_discount_info格式化格式化優(yōu)惠信息，方便存放到優(yōu)惠信息，方便存放到 .csv .csv文件文件5.5.右輸出表：獲取數(shù)據(jù)右輸出表：獲取數(shù)據(jù)6.6.最終輸出表：按商品名稱、商品容量最終輸出表：按商品名稱、商品容量 join join 左右兩表左右兩表7.7.輸出輸出joinjoin結(jié)果至結(jié)果至Csv SinkCsv Sink工程上線后，由于數(shù)據(jù)收集端在不停的提供數(shù)據(jù)，借助流計(jì)算，我們已經(jīng)鑒別出 200 個(gè)涉嫌違規(guī)定價(jià)的商品鏈接，在維護(hù)品牌力和價(jià)格力的同時(shí)，避免旗艦店銷(xiāo)售損失 40 余萬(wàn)，而這只是我們眾多監(jiān)控用 Flink Job 中的一個(gè)。