ES 實現實時從Mysql資料庫中讀取熱詞,停用詞
阿新 • • 發佈：2020-09-14
IK分詞器雖然自帶詞庫
但是在實際開發應用中對於詞庫的靈活度的要求是遠遠不夠的,IK分詞器雖然配置檔案中能新增擴充套件詞庫,但是需要重啟ES
這章就當寫一篇擴充套件了
其實IK本身是支援熱更新詞庫的,但是需要我感覺不是很好
詞庫熱更新方案:
1:IK原生的熱更新方案,部署一個WEB伺服器,提供一個Http介面,通過Modified和tag兩個Http響應頭,來完成詞庫的熱更新
2:通過修改IK原始碼支援Mysql定時更新資料
注意:推薦使用第二種方案,也是比較常用的方式,雖然第一種是官方提供的,但是官方也不建議使用
方案一:IK原生方案
1:外掛詞庫,就是在IK配置檔案中新增擴充套件詞庫檔案多個之間使用分號分割
優點:編輯指定詞庫檔案,部署比較方便
缺點:每次編輯更新後都需要重啟ES
2:遠端詞庫,就是在IK配置檔案中配置一個Http請求,可以是.dic檔案,也可以是介面,同樣多個之間使用分號分割
優點:指定靜態檔案,或者介面設定詞庫實現熱更新詞庫,不用重啟ES,是IK原生自帶的
缺點:需要通過Modified和tag兩個Http響應頭,來提供詞庫的熱更新,有時候會不生效
具體使用就不說了,在這裡具體說第二種方案
方案二:通過定時讀取Mysql完成詞庫的熱更新
首先要下載IK分詞器的原始碼
網址:https://github.com/medcl/elasticsearch-analysis-ik
下載的時候一定要選對版本,保持和ES的版本一致,否則會啟動的時候報錯,版本不一致
接著把原始碼匯入IDEA中,並在POM.xml中新增Mysql的依賴,根據自己的Mysql版本需要新增
我的Mysql是5.6.1所以新增5的驅動包
<!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java --> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>5.1.49</version> </dependency>
然後再config目錄下建立一個新的.properties配置檔案
在裡面配置Mysql的一些配置,以及我們需要的配置
jdbc.url=jdbc:mysql://192.168.43.154:3306/es?characterEncoding=UTF-8&serverTimezone=GMT&nullCatalogMeansCurrent=true jdbc.user=root jdbc.password=root # 更新詞庫 jdbc.reload.sql=select word from hot_words # 更新停用詞詞庫 jdbc.reload.stopword.sql=select stopword as word from hot_stopwords # 重新拉取時間間隔 jdbc.reload.interval=5000
建立一個新的執行緒，用於呼叫Dictionary得reLoadMainDict()方法重新載入詞庫
package org.wltea.analyzer.dic; import org.wltea.analyzer.help.ESPluginLoggerFactory; public class HotDicReloadThread implements Runnable{ private static final org.apache.logging.log4j.Logger logger = ESPluginLoggerFactory.getLogger(Dictionary.class.getName()); @Override public void run() { while (true){ logger.info("-------重新載入mysql詞典--------"); Dictionary.getSingleton().reLoadMainDict(); } } }
修改org.wltea.analyzer.dic資料夾下的Dictionary
在Dictionary類中載入mysql驅動類
private static Properties prop = new Properties(); static { try { Class.forName("com.mysql.jdbc.Driver"); } catch (ClassNotFoundException e) { logger.error("error", e); } }
接著,建立重Mysql中載入詞典的方法
/** * 從mysql中載入熱更新詞典 */ private void loadMySqlExtDict(){ Connection connection = null; Statement statement = null; ResultSet resultSet = null; try { Path file = PathUtils.get(getDictRoot(),"jdbc-reload.properties"); prop.load(new FileInputStream(file.toFile())); logger.info("-------jdbc-reload.properties-------"); for (Object key : prop.keySet()) { logger.info("key:{}", prop.getProperty(String.valueOf(key))); } logger.info("------- 查詢詞典, sql:{}-------", prop.getProperty("jdbc.reload.sql")); // 建立mysql連線 connection = DriverManager.getConnection( prop.getProperty("jdbc.url"), prop.getProperty("jdbc.user"), prop.getProperty("jdbc.password") ); // 執行查詢 statement = connection.createStatement(); resultSet = statement.executeQuery(prop.getProperty("jdbc.reload.sql")); // 迴圈輸出查詢啊結果,新增到Main.dict中去 while (resultSet.next()) { String theWord = resultSet.getString("word"); logger.info("------熱更新詞典:{}------", theWord); // 加到mainDict裡面 _MainDict.fillSegment(theWord.trim().toCharArray()); } } catch (Exception e) { logger.error("error:{}", e); } finally { try { if (resultSet != null) { resultSet.close(); } if (statement != null) { statement.close(); } if (connection != null) { connection.close(); } } catch (SQLException e){ logger.error("error", e); } } }
接著,建立載入停用詞詞典方法
/** * 從mysql中載入停用詞 */ private void loadMySqlStopwordDict(){ Connection conn = null; Statement stmt = null; ResultSet rs = null; try { Path file = PathUtils.get(getDictRoot(), "jdbc-reload.properties"); prop.load(new FileInputStream(file.toFile())); logger.info("-------jdbc-reload.properties-------"); for(Object key : prop.keySet()) { logger.info("-------key:{}", prop.getProperty(String.valueOf(key))); } logger.info("-------查詢停用詞, sql:{}",prop.getProperty("jdbc.reload.stopword.sql")); conn = DriverManager.getConnection( prop.getProperty("jdbc.url"), prop.getProperty("jdbc.user"), prop.getProperty("jdbc.password")); stmt = conn.createStatement(); rs = stmt.executeQuery(prop.getProperty("jdbc.reload.stopword.sql")); while(rs.next()) { String theWord = rs.getString("word"); logger.info("------- 載入停用詞 : {}", theWord); _StopWords.fillSegment(theWord.trim().toCharArray()); } Thread.sleep(Integer.valueOf(String.valueOf(prop.get("jdbc.reload.interval")))); } catch (Exception e) { logger.error("error", e); } finally { try { if(rs != null) { rs.close(); } if(stmt != null) { stmt.close(); } if(conn != null) { conn.close(); } } catch (SQLException e){ logger.error("error:{}", e); } } }
接下來,分別在loadMainDict()方法和loadStopWordDict()方法結尾處呼叫
/** * 載入主詞典及擴充套件詞典 */ private void loadMainDict() { // 建立一個主詞典例項 _MainDict = new DictSegment((char) 0); // 讀取主詞典檔案 Path file = PathUtils.get(getDictRoot(), Dictionary.PATH_DIC_MAIN); loadDictFile(_MainDict, file, false, "Main Dict"); // 載入擴充套件詞典 this.loadExtDict(); // 載入遠端自定義詞庫 this.loadRemoteExtDict(); // 載入Mysql外掛詞庫 this.loadMySqlExtDict(); }
/** * 載入使用者擴充套件的停止詞詞典 */ private void loadStopWordDict() { // 建立主詞典例項 _StopWords = new DictSegment((char) 0); // 讀取主詞典檔案 Path file = PathUtils.get(getDictRoot(), Dictionary.PATH_DIC_STOP); loadDictFile(_StopWords, file, false, "Main Stopwords"); // 載入擴充套件停止詞典 List<String> extStopWordDictFiles = getExtStopWordDictionarys(); if (extStopWordDictFiles != null) { for (String extStopWordDictName : extStopWordDictFiles) { logger.info("[Dict Loading] " + extStopWordDictName); // 讀取擴充套件詞典檔案 file = PathUtils.get(extStopWordDictName); loadDictFile(_StopWords, file, false, "Extra Stopwords"); } } // 載入遠端停用詞典 List<String> remoteExtStopWordDictFiles = getRemoteExtStopWordDictionarys(); for (String location : remoteExtStopWordDictFiles) { logger.info("[Dict Loading] " + location); List<String> lists = getRemoteWords(location); // 如果找不到擴充套件的字典，則忽略 if (lists == null) { logger.error("[Dict Loading] " + location + " load failed"); continue; } for (String theWord : lists) { if (theWord != null && !"".equals(theWord.trim())) { // 載入遠端詞典資料到主記憶體中 logger.info(theWord); _StopWords.fillSegment(theWord.trim().toLowerCase().toCharArray()); } } } // 載入Mysql停用詞詞庫 this.loadMySqlStopwordDict(); }
最後在initial()方法中啟動更新執行緒
/** * 詞典初始化 由於IK Analyzer的詞典採用Dictionary類的靜態方法進行詞典初始化 * 只有當Dictionary類被實際呼叫時，才會開始載入詞典， 這將延長首次分詞操作的時間 該方法提供了一個在應用載入階段就初始化字典的手段 * * @return Dictionary */ public static synchronized void initial(Configuration cfg) { if (singleton == null) { synchronized (Dictionary.class) { if (singleton == null) { singleton = new Dictionary(cfg); singleton.loadMainDict(); singleton.loadSurnameDict(); singleton.loadQuantifierDict(); singleton.loadSuffixDict(); singleton.loadPrepDict(); singleton.loadStopWordDict(); // 執行更新mysql詞庫的執行緒 new Thread(new HotDicReloadThread()).start(); if(cfg.isEnableRemoteDict()){ // 建立監控執行緒 for (String location : singleton.getRemoteExtDictionarys()) { // 10 秒是初始延遲可以修改的 60是間隔時間 單位秒 pool.scheduleAtFixedRate(new Monitor(location), 10, 60, TimeUnit.SECONDS); } for (String location : singleton.getRemoteExtStopWordDictionarys()) { pool.scheduleAtFixedRate(new Monitor(location), 10, 60, TimeUnit.SECONDS); } } } } } }
然後,修改src/main/assemblies/plugin.xml檔案中,加入Mysql
<dependencySet>
<outputDirectory>/</outputDirectory>
<useProjectArtifact>true</useProjectArtifact>
<useTransitiveFiltering>true</useTransitiveFiltering>
<includes>
<include>mysql:mysql-connector-java</include>
</includes>
</dependencySet>
原始碼到此修改完成，在自己的資料庫中建立兩張新的表
建表SQL
CREATE TABLE hot_words ( id bigint(20) NOT NULL AUTO_INCREMENT, word varchar(50) COLLATE utf8_unicode_ci DEFAULT NULL COMMENT '詞語', PRIMARY KEY (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci; CREATE TABLE hot_stopwords ( id bigint(20) NOT NULL AUTO_INCREMENT, stopword varchar(50) COLLATE utf8_unicode_ci DEFAULT NULL COMMENT '停用詞', PRIMARY KEY (id) ) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci;
接下來對原始碼進行打包:
打包之前檢查自己的POM.xml中的elasticsearch.version的版本,記得和自己的ES的版本對應,否則到時候會報錯
檢查完畢後,點選IDEA右側的package進行專案打包,如果版本不對,修改版本並點選IDEA右側的重新整理同步,進行版本的更換,然後打包
打包完成後在左側專案中會出現target目錄,會看到一個zip,我的是因為解壓了,所以有資料夾
點選右鍵在資料夾中展示,然後使用解壓工具解壓
解壓完成後,雙擊進入
先把原來ES下的plugins下的IK資料夾中的東西刪除,可以先備份,然後把自己打包解壓后里面的東西全部拷貝到ES下的plugins下的IK資料夾中
接下來進入bin目錄下啟動就可以了
當然按照慣例,我的啟動時不會那麼簡單的,很高興,我的報錯了,所有的坑都踩了一遍,之前的版本不對就踩了兩次
第一次是原始碼下載的版本不對
第二次的ES依賴版本不對
好了說報錯:報錯只貼主要內容
第三次報錯:
Caused by: java.security.AccessControlException: access denied ("java.lang.RuntimePermission" "setContextClassLoader")
這個是JRE的類的建立設值許可權不對
在jre/lib/security資料夾中有一個java.policy檔案,在其grant{}中加入授權即可
permission java.lang.RuntimePermission "createClassLoader"; permission java.lang.RuntimePermission "getClassLoader"; permission java.lang.RuntimePermission "accessDeclaredMembers"; permission java.lang.RuntimePermission "setContextClassLoader";
第四次報錯:
Caused by: java.security.AccessControlException: access denied ("java.net.SocketPermission" "192.168.43.154:3306" "connect,resolve")
這個是通訊連結等許可權不對
也是,在jre/lib/security資料夾中有一個java.policy檔案,在其grant{}中加入授權即可
permission java.net.SocketPermission "192.168.43.154:3306","accept"; permission java.net.SocketPermission "192.168.43.154:3306","listen"; permission java.net.SocketPermission "192.168.43.154:3306","resolve"; permission java.net.SocketPermission "192.168.43.154:3306","connect";
到此之後啟動無異常
最後就是測試了,啟動我的head外掛和kibana,這兩個沒有或者不會的可以看我之前寫的,也可以百度
執行分詞
但是我想要 天青色
在Mysql中新增記錄
insert into hot_words(word) value("天青色");
重新執行
也比如我想要這就是一個詞 天青色等煙雨
在Mysql中新增記錄
insert into hot_words(word) value("天青色等煙雨");
再次執行
到此實現了ES定時從mysql中讀取熱詞,停用詞這個一般用的比較少,有興趣自己測測,在使用的時候,通過業務系統往資料庫熱詞表和停用詞表新增記錄就可以了
作者：彼岸舞
時間：2020\09\13
內容關於：ElasticSearch
本文來源於網路，只做技術分享，一概不負任何責任