大數(shù)據(jù)開發(fā)--基于ip2region實(shí)現(xiàn)在流式數(shù)據(jù)ip歸屬轉(zhuǎn)換

IT那活兒發(fā)布于2023-01-11 13:20 / 1304人閱讀

點(diǎn)擊上方“IT那活兒”，關(guān)注后了解更多內(nèi)容，不管IT什么活兒，干就完了！??！

文章前言

我們?cè)诹魇接?jì)算數(shù)據(jù)加工流程中，有個(gè)場景是實(shí)現(xiàn)一種基于流式數(shù)據(jù)里面ip轉(zhuǎn)換成對(duì)應(yīng)的歸屬地信息，這個(gè)場景首先想到是如何找一個(gè)符合條件的ip映射庫，而這個(gè)ip映射庫需滿足如下條件：

1. 性能好：因?yàn)閷?shí)時(shí)數(shù)據(jù)量大，每條數(shù)據(jù)都需要實(shí)時(shí)調(diào)用，獲取ip映射關(guān)系，所以性能要好，如果延時(shí)較大，會(huì)影響后面流程節(jié)點(diǎn)處理；

2. 開源，能擴(kuò)展：開源，能內(nèi)網(wǎng)部署，并提供擴(kuò)展功能，支持加新的ip映射關(guān)系；

3. 方便調(diào)用：支持api調(diào)用，且精準(zhǔn)率要高。

通過調(diào)研，開源ip2region.db數(shù)據(jù)庫符合我們上面實(shí)際場景需求，本文檔主要介紹在flink流式計(jì)算中如何使用了ip2region，通過測試程序了解flink 是如何加載外部文件機(jī)制，如需了解ip2region更多詳情，可自行到官網(wǎng)地址查閱https://github.com/lionsoul2014/ip2region

功能實(shí)現(xiàn)

1. 下載項(xiàng)目

git clone https://github.com/lionsoul2014/ip2region.git

2. 引用maven依賴

程序引用ip2region依賴。

<dependency>

<groupId>org.lionsoulgroupId>

<artifactId>ip2regionartifactId>

<version>1.7.2version>

dependency>

3. demo編寫

api比較簡單就三行代碼，采用的是memory查詢算法。

打包測試

程序打包成功后，將程序提交到flink on yarn 環(huán)境之前，我們要考慮ip2regiog.db存儲(chǔ)的位置及加載方式，否則在分布式環(huán)境下flink程序是無否讀到該數(shù)據(jù)庫：

方式一：ip2region.db庫放入hadoop各計(jì)算節(jié)點(diǎn):

將ip2region.db數(shù)據(jù)庫文件放入hadoop各計(jì)算節(jié)點(diǎn)上指定位置上（如：/home/gpadmin/jar_repo/config/ip2region.db），由分配到此計(jì)算節(jié)點(diǎn)flink任務(wù)到指定目錄下加載對(duì)應(yīng)的數(shù)據(jù)庫文件，實(shí)現(xiàn)ip歸屬轉(zhuǎn)換。

1）新建basic.config配置文件，加入配置項(xiàng)

ip2region.db.file.path：/home/ip2region.db //指定讀取的路徑。

2）提交測試程序到y(tǒng)arn

flink run -m yarn-cluster -p 1 -yjm 1024 -ytm 1024 -ynm 
ipregiontest  -c com.shsnc.fk.task.metricnormal.MetricsCollectTask regiondb.jar basic.config

輸出計(jì)算結(jié)果：

2021-12-17 16:27:26,470 INFO 
com.XXXXX.fk.task.dataprocess.function.DataProcessPreRichMap
Func [] - regin:中國|0|XXX|XXX|XX

3）分析加載流程

我們?cè)趂link run 啟動(dòng)命行里面，沒有直接指向ip2region.db加載方式，而是在應(yīng)用程序加載配置項(xiàng)獲取對(duì)應(yīng)的數(shù)據(jù)庫文件存儲(chǔ)位置，通過觀察任務(wù)提交到flink on yarn 環(huán)境過程中，發(fā)現(xiàn)提交任務(wù)過程，tm自動(dòng)會(huì)將該數(shù)據(jù)庫文件放入yarn任務(wù)緩存目錄下（Task ClassPath），任務(wù)運(yùn)行過程中會(huì)到緩存目錄下找到數(shù)據(jù)庫文件。

ps -ef|grep container_1636698971952_0029_01_000002（容器id）。

ll /proc/進(jìn)程號(hào)。

這種方式測試程序可以找到ip2region.db數(shù)據(jù)庫文件，并且能計(jì)算出所需要ip歸屬，缺點(diǎn)是每臺(tái)hadoop計(jì)算節(jié)點(diǎn)上都要將ip2region.db數(shù)據(jù)庫文件放入指定位置，否則會(huì)提示讀不到數(shù)據(jù)庫文件現(xiàn)象，如果集群環(huán)境節(jié)點(diǎn)較多使用起來并不方便。

方式二：一次性加載方式

通過檢查flink run命令參數(shù)，發(fā)現(xiàn)-yt 參數(shù)方式很好解決我們當(dāng)前面臨的問題，先看一下參數(shù)的使用說明：

參數(shù)方式方式是 -yt 目錄名，該參數(shù)意思是在指定目錄中傳輸文件,處理流程如下：

-yt 目錄名--hdfs(存儲(chǔ))--tm（下載）-tm(classpath)。

1）提交任務(wù)

flink run -m yarn-cluster -p 1 -yjm 1024 -ytm 1024 -yt 
/home/gpadmin/jar_repo/config -ynm ipregiontest -c com.XXXXX.fk.task.metricnormal.MetricsCollectTask regiondb.jar basic.config

注意這里指定是目錄，而不是指定的是具體加載的文件。

2）hdfs查找存儲(chǔ)位置

hdfs dfs -find hdfs://master:port/ -iname “ip2region.db”

可以看出flink 任務(wù)在提交過程中，會(huì)將ip2region.db自動(dòng)存儲(chǔ)在hdfs上flink對(duì)應(yīng)的任務(wù)目錄下，同時(shí)把-yt 提向的目錄（config ）也存儲(chǔ)了。

3）自動(dòng)下載

檢查每個(gè)haoop計(jì)算節(jié)點(diǎn)，tm會(huì)自動(dòng)到hdfs上下載ip2region.db存放到緩存目錄下（Task ClassPath），所以任務(wù)運(yùn)行過程就要讀到ip2region.db數(shù)據(jù)庫文件，通過api很方便獲取ip歸屬。