Elasticsearch實戰 | 如何從數千萬手機號中識別出情侶號?

  • 2020 年 3 月 31 日
  • 筆記

1、問題描述

您好,請教個問題。我現在有2千多萬的手機號碼資訊保存在es里。5個分片,3個節點。 現在的需求是將後八位相同的號碼匹配到一起,重新放到一個index里。組成情侶號。方便後續查詢情侶號列表。 我目前的做法是用scroll查詢出一萬條,多執行緒循環一萬條中的每條,去全庫掃描—但是這種做法一分鐘才能處理一萬條。您有什麼新的思路沒。 死磕Elasticsearch知識星球 https://t.zsxq.com/Iie66qV

問題補充:索引存儲了手機號,同時存儲了插入時間。

2、問題分析

2.1 情侶號的定義

後八位相同的號碼即為情侶號。

舉例:

13011112222  13511112222  13711112222  

2.2 如何對後8位建立索引,以方便後續的識別?

方案一 不單獨建索引,用script來實現

缺點:script效率低一些

方案二:寫入數據的時候,同時基於後八位創建新的欄位。

2.3 8位相同的號碼匹配到一起,重新放到一個index里怎麼實現?

Elasticsearch自帶reindex功能就是實現索引遷移的,當然自定義讀寫也可以實現。

方案一:遍歷方式+寫入。

  • 步驟 1:基於時間遞增循環遍歷,以起始的手機號為種子數據,滿足後八位相同的加上標記flag=1。
  • 步驟 2:循環步驟1,滿足flag=1直接跳過,直到所有手機號遍歷一遍。
  • 步驟 3:將包含flag=1的欄位,reindex到情侶號索引。

方案二:聚合出情侶號組,將聚合結果reindex到情侶號索引。

考慮到數據量級千萬級別,全量聚合不現實。

可以,基於時間切片,取出最小時間戳、最大時間戳,根據數據總量和時間範圍劃分出時間間隔。

舉例:以30分鐘為單位切割千萬級數據。

  • 步驟 1:terms聚合後8位手機號。

terms聚合只返回對應:key,value值,默認value值由高到低排序。

key:代表手機號後8位,value:代表相同後8位的數據量。

  • 步驟 2:top_hits子聚合取出手機號詳情。
  • 步驟 3:json解析識別出步驟2的所有手機號或_id。
  • 步驟 4:reindex步驟3的_id數據到情侶號索引。
  • 步驟 5:時間切片周期遞增,直到所有數據遍歷完畢。

2.4 擴展自問:手機號怎麼存,才能查出來後8位?

舉例:查詢「11112222」,返回2.1列表的三個手機號。

  • 方案1:wildcard模糊匹配。

優點:無需額外欄位存儲。

缺點:效率低。

  • 方案2:ngram分詞+match_phrase處理。

優點:效率高。

缺點:需要獨立存儲的後8位欄位。

3、實戰一把

3.1 數據建模

3.1.1 欄位設計

只包含非業務的有效必要欄位。

(1)插入時間戳欄位 insert_time, date類型。

由:ingest默認生成,不手動添加,提高效率。

(2)手機號欄位 phone_number, text和keyword類型。

  • text類型基於ngram分詞,主要方便phone_number全文檢索。
  • keyword類型方便:排序和聚合使用。

(3)後8位手機號欄位 last_eight_number, keyword類型。

只聚合和排序用,不檢索。

3.1.2 ingest處理初始化數據先行

ingest pipeline的核心功能可以理解為寫入前數據的ETL。

而:insert_time可以自動生成、last_eight_number可以基於phone_number提取。

定義如下:

# 0.create ingest_pipeline of insert_time and last_eight_number  PUT _ingest/pipeline/initialize  {    "description": "Adds insert_time timestamp to documents",    "processors": [      {        "set": {          "field": "_source.insert_time",          "value": "{{_ingest.timestamp}}"        }      },      {        "script": {          "lang": "painless",          "source": "ctx.last_eight_number = (ctx.phone_number.substring(3,11))"        }      }    ]  }  

3.1.3 模板定義

兩個索引:

  • 索引1:phone_index,存儲全部手機號(數千萬)
  • 索引2:phone_couple_index,存儲情侶號

由於兩索引Mapping結構一樣,使用模板管理會更為方便。

定義如下:

# 1.create template of phone_index and phone_couple_index  PUT _template/phone_template  {    "index_patterns": "phone_*",    "settings": {      "number_of_replicas": 0,      "index.default_pipeline": "initialize",      "index": {        "max_ngram_diff": "13",        "analysis": {          "analyzer": {            "ngram_analyzer": {              "tokenizer": "ngram_tokenizer"            }          },          "tokenizer": {            "ngram_tokenizer": {              "token_chars": [                "letter",                "digit"              ],              "min_gram": "1",              "type": "ngram",              "max_gram": "11"            }          }        }      }    },    "mappings": {      "properties": {        "insert_time":{          "type":"date"        },        "last_eight_number":{          "type":"keyword"        },        "phone_number": {          "type": "text",          "fields": {            "keyword": {              "type": "keyword"            }          },          "analyzer": "ngram_analyzer"        }      }    }  }  

3.1.4 索引定義

PUT phone_index  PUT phone_couple_index  

3.2 數據寫入

採用模擬數據,實際業務會有所區別。

POST phone_index/_bulk  {"index":{"_id":1}}  {"phone_number" : "13511112222"}  {"index":{"_id":2}}  {"phone_number" : "13611112222"}  {"index":{"_id":3}}  {"phone_number" : "13711112222"}  {"index":{"_id":4}}  {"phone_number" : "13811112222"}  {"index":{"_id":5}}  {"phone_number" : "13844248474"}  {"index":{"_id":6}}  {"phone_number" : "13866113333"}  {"index":{"_id":7}}  {"phone_number" : "15766113333"}    

模擬數據顯示,有兩組情侶號。

  • 第一組情侶號尾數:「11112222」
  • 第二組情侶號尾數:「66113333」

3.2 數據聚合

如前所述,聚合的目的是:提取出情侶號(>=2)的手機號或對應id。

GET phone_index/_search  {    "size": 0,   "query": {     "range": {       "insert_time": {         "gte": 1584871200000,         "lte": 1584892800000       }     }   },    "aggs": {      "last_aggs": {        "terms": {          "field": "last_eight_number",          "min_doc_count": 2,          "size": 10,          "shard_size": 30        },        "aggs": {          "sub_top_hits_aggs": {            "top_hits": {              "size": 100,              "_source": {                "includes": "phone_number"              },              "sort": [                {                  "phone_number.keyword": {                    "order": "asc"                  }                }              ]            }          }        }      }    }  }  

注意:

  • 查詢的目的:按時間間隔取數據。原因:「聚合全量性能太差」
  • 外層聚合last_aggs統計:情侶號分組及數量。
  • 內層子聚合sub_top_hits_aggs統計:下鑽的手機號或_id等資訊。
  • min_doc_count作用:聚合後的分組記錄最小條數,情侶號必須>=2,則設置為2。

3.4 數據遷移

基於3.3 取出的滿足條件的id進行跨索引遷移。

POST _reindex  {    "source": {      "index": "phone_index",      "query": {        "terms": {          "_id": [            1,            2,            3,            4,            6,            7          ]        }      }    },    "dest": {      "index": "phone_couple_index"    }  }  

注意:實際業務需要考慮數據規模,劃定輪詢時間間隔區間。

建議:按照2.3章節的流程圖執行。

4、方案進一步探究

第3節的實戰一把實際是基於基礎數據都寫入ES了再做的處理。

核心的操作都是基於Elasticsearch完成的。

試想一下,這個環節如果提前是不是更合理呢?

數據圖如下所示:

  • 電話數據資訊寫入消息隊列(如:kafka、rocketmq、rabbitmq等)。
  • 消息隊列可以直接同步到ES的phone_index索引。如:紅線所示。
  • 情侶號的處理藉助第三方redis服務實現,逐條過濾,滿足條件的數據同步到ES的情侶號索引phone_couple_index。如:綠線所示。

這樣,Elasticsearch只干它最擅長的事情,剩下的工作前置交給消息隊列完成。

5、小結

本文就提出問題做了詳細的闡述和實踐,用到Elasticsearch 模板、Ingest、reindex等核心知識點和操作,給線上業務提供了理論參考。

大家對本文有異議或者有更好的方案,歡迎留言交流。