本文已收錄github：https://github.com/BigDataScholar/TheKingOfBigData，里面有大數據高頻考點，Java一線大廠面試題資源，上百本免費電子書籍，作者親繪大數據生態圈思維導圖…持續更新，歡迎star！

前言

????????上一篇文章已經為大家介紹了 HBase 在用戶畫像的標簽數據存儲中的具體應用場景，本篇我們來談談 Elasticsearch 的使用！

原著作者：趙宏田
來源：《用戶畫像方法論與工程化解決方案》

---

Elasticsearch存儲

Elasticsearch簡介

????????Elasticsearch 是一個開源的分布式全文檢索引擎，可以近乎實時地存儲、檢索數據。而且可擴展性很好，可以擴展到上百臺服務器，處理PB級別的數據。對于用戶標簽查詢、用戶人群計算、用戶群多維透視分析這類對響應時間要求較高的場景，也可以考慮選用Elasticsearch進行存儲。

????????Elasticsearch是面向文檔型數據庫，一條數據在這里就是一個文檔，用 json 作為文檔格式。為了更清晰地理解 Elasticsearch 查詢的一些概念，將其和關系型數據庫的類型進行對照。

ElasticsearchMySQL

index	database	數據庫
type	table	表
document	row	行
mapping	column	列
GET http://…	SELECT * FROM …	查詢數據
PUT http://…	UPDATE table SET…	插入數據

????????在關系型數據庫中查詢數據時可通過選中數據庫、表、行、列來定位所查找的內容，在Elasticsearch中通過索引（index）、類型（type）、文檔（document）、字段來定位查找內容。一個Elasticsearch集群可以包括多個索引（數據庫），也就是說，其中包含了很多類型（表），這些類型中包含了很多的文檔（行），然后每個文檔中又包含了很多的字段（列）。Elasticsearch的交互可以使用Java API，也可以使用 HTTP 的RESTful API方式。

應用場景

????????在上一節的內容中，我們談到基于 HBase 的存儲方案并沒有解決數據的 高效檢索 問題。在實際應用中，經常有根據特定的幾個字段進行組合后檢索的應用場景，而 HBase 采用 rowkey 作為一級索引，不支持多條件查詢，如果要對庫里的非 rowkey 進行數據檢索和查詢，往往需要通過 MapReduce 等分布式框架進行計算，時間延遲上會比較高，難以同時滿足用戶對于復雜條件查詢和高效率響應這兩方面的需求。

????????為了既能支持對數據的高效查詢，同時也能支持通過條件篩選進行復雜查詢，需要在HBase上構建二級索引，以滿足對應的需要。在本案中我們采用Elasticsearch存儲 HBase 的索引信息，以支持復雜高效的查詢功能。

????????主要查詢過程包括：

????????1）在Elasticsearch中存放用于檢索條件的數據，并將rowkey 也存儲進去；

????????2）使用Elasticsearch的 API 根據組合標簽的條件查詢出rowkey的集合；

????????3）使用上一步得到的 rowkey 去HBase數據庫查詢對應的結果

????????HBase存儲數據的索引放在Elasticsearch中，實現了數據和索引的分離。在Elasticsearch中documentid是文檔的唯一標識，在HBase中rowkey是記錄的唯一標識。在工程實踐中，兩者可同時選用用戶在平臺上的唯一標識（如userid或deviceid）作為rowkey或documentid，進而解決 HBase 和 Elasticsearch 索引關聯的問題。

????????下面通過使用 Elasticsearch 解決用戶人群計算和分析應用場景的案例來了解這一過程。

????????對匯聚后的用戶標簽表dw.userprofile_userlabel_map_all中的數據進行清洗，過濾掉一些無效字符，達到導入Elasticsearch的條件，如圖所示：

????????然后將dw.userprofile_userlabel_map_all數據寫入Elasticsearch 中，Scala代碼如下：

object HiveDataToEs {def main(args: Array[String]): Unit = {val spark = SparkSession.builder().AppName("EsData").config("spark.serializer", "org.apache.spark.serializer.KryoSerializer").config("spark.dynamicAllocation.enabled", "false").config("es.index.auto.create", "true").config("es.nodes", "10.xx.xx.xx").config("es.batch.write.retry.count", "3") // 默認重試3次.config("es.batch.write.retry.wait", "5") // 每次重試等待時間為5秒.config("thread_pool.write.queue_size", "1000").config("thread_pool.write.size", "50").config("thread_pool.write.type", "fixed") .config("es.batch.size.bytes", "20mb") .config("es.batch.size.entries", "2000") .config("es.http.timeout","100m").enableHiveSupport().getOrCreate()val data_date = args(0).toStringimport spark.sqlval hiveDF = sql(s"""| SELECT userid, tagsmap FROM dw.userprofile_userlabel_map_all where data_date = '${data_date}'""".stripMargin) // dw.userprofile_userlabel_map_all 是聚合用戶標簽的表val rdd = hiveDF.rdd.map {row => {val userid = row.getAs[String]("userid")val userlabels = row.getAs[Map[String, Object]]("userlabels")Map("userid" -> userid, "userlabels" -> userlabels)}}EsSpark.saveToEs(rdd , "userprofile/tags", Map[String,String]("es.mApping.id"->"userid") spark.stop()} }

????????工程依賴如下：

<dependency><groupId>org.elasticsearch</groupId><artifactId>elasticsearch-hadoop</artifactId><version>6.4.2</version> </dependency>

????????將該工程打包之后提交任務，傳入日期分區參數 “20190101”執行。提交命令“spark-submit--class com.example.HiveDataToEs--master yarn--deploy-mode client--executor-memory 2g--num-executors 50--driver-memory 3g--executor-cores 2 spark-hive-to-es.jar 20190101”

????????任務執行完畢后，當日 userid 維度的用戶標簽數據全部導入Elasticsearch中。使用RESTfulAPI查詢包含某個標簽的用戶量，可實時得到返回結果。

# 查詢命令 GET userprofile/tags/_search {"size":0,"aggs": {"tagcounts": {"terms": {"field": "tags.ACTION_U_01_003"}}} }

????????從返回結果中可以看到，用戶總量（total）為100000000人，包含標簽“ACTION_U_01_003”的用戶有2500000人（doc_count）。

????????查詢人群 index 查看標簽總量：

# 查詢命令 GET userprofile/_search {"query":{"match_all": {}} }

????????查詢結果如圖所示：

????????在人群的計算和分析場景中，經過產品的迭代，前期采用 Impala 進行計算，一般耗費幾十秒到幾分鐘的時間，在使用 Elasticsearch 后，實現了對人群計算的秒級響應。

工程化案例

????????下面通過一個工程案例來講解實現畫像產品中“用戶人群”和“人群分析”功能對用戶群計算秒級響應的一種解決方案。

????????在每天的 ETL 調度中，需要將 Hive 計算的標簽數據導入Elasticsearch中。如圖所示，在標簽調度完成且通過校驗后（圖中的“標簽監控預警”任務執行完成后），將標簽數據同步到Elasticsearch中。

????????在與 Elasticsearch 數據同步完成并通過校驗后，向在 MySQL 中維護的狀態表中插入一條狀態記錄，表示當前日期的 Elasticsearch 數據可用，線上計算用戶人群的接口則讀取最近日期對應的數據。如果某天因為調度延遲等方面的原因，沒有及時將當日數據導入Elasticsearch中，接口也能讀取最近一天對應的數據，是一種可行的災備方案。

????????例如，數據同步完成后向MySQL狀態表“elasticsearch_state”中插入記錄（如圖所示），當日數據產出正常時，state字段為“0”，產出異常時為“1”。圖3-29中1月20日導入的數據出現異常，則“state”狀態字段置1，線上接口掃描該狀態記錄位后不讀取1月20日數據，而是取用最近的1月19日數據。

????????為了避免從 Hive 向 Elasticsearch 中灌入數據時發生數據缺失，在向狀態表更新狀態位前需要校驗 Elasticsearch 和 Hive 中的數據量是否一致。下面通過Python腳本來看數據校驗邏輯：

# 查詢Hive中的數據 def monitor_hive_data(data_date):hive_user = " select count(1) from dw.userprofile_userlabel_map_all where data_date='{}' ".format(data_date)user_count = os.popen("hive -S -e \"" + hive_user + "\"").read().strip()return user_count# 查詢es中的數據 def monitor_es_data(data_date):userid_search = "curl http://10.xxx.xxx.xxx:9200/_cat/count/" + data_date + "_userid/"userid_num = str(os.popen(userid_search).read()).split(' ')[-1].strip()return userid_num# 比較Hive和es中的數據，如通過校驗，更新MySQL狀態位 def update_es_data(data_date):'''data_date: 查詢數據日期'''esdata = monitor_es_data(data_date) # 查詢es中的數據hivedata = monitor_hive_data(data_date) # 查詢Hive中的數據print("esdata ======>{}".format(esdata))print("hivedata ======>{}".format(hivedata))# 更新MySQL狀態位 if (esdata[0] == hivedata[0] ):db = MySQLdb.connect(host="10.xx.xx.xx", port=3306, user="username", passwd="password",db="userprofile", charset="utf8")cursor = db.cursor()try:select_command = "INSERT INTO `elasticsearch_state` VALUES ('"+ str(data_date) +"', 'elasticsearch', '0', '2');"cursor.execute(select_command)db.commit()except Exception as e:db.rollback()exit(1)

????????上面介紹了在工程化調度流中何時將Hive中的用戶標簽數據灌入Elasticsearch中，之后業務人員在畫像產品端計算人群或透視分析人群時（如圖所示），

通過RESTful API訪問 Elasticsearch 進行計算

小結

????????結合前面幾期文章，分別為大家講解了使用 Hive、MySQL、HBase 和 Elasticsearch 存儲標簽數據的解決方案，包括：Hive存儲數據相關標簽表、人群計算表的表結構設計以及ID-Mapping的一種實現方式；MySQL存儲標簽元數據、監控數據及結果集數據；HBase存儲線上接口實時調用的數據；Elasticsearch存儲標簽用于人群計算和人群多維透視分析。存儲過程中涉及如下相關表。

• dw.userprofile_attritube_all：存儲人口屬性維度的標簽表；
• dw.userprofile_action_all：存儲行為屬性維度的標簽表；
• dw.userprofile_consume_all：存儲用戶消費維度的標簽表；
• dw.userprofile_riskmanage_all：存儲風險控制維度的標簽表；
• dw.userprofile_social_all：存儲社交屬性維度的標簽表；
• dw.userprofile_userlabel_map_all：匯聚用戶各維度標簽的表；
• dw.userprofile_usergroup_labels_all：存儲計算后人群數據的表。

????????好了，本篇文章就到這里，更多干貨文章請關注我的公眾號。你知道的越多，你不知道的也越多。我是Alice，我們下一期見！

總結

以上是生活随笔為你收集整理的用户画像 | 标签数据存储之Elasticsearch真实应用的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。