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等核心知識點和操作，給線上業務提供了理論參考。

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