總覺得階段性的總結是個好習慣，很多自己做的事情，如果不及時總結一下，過一段時間就忘記了，當要用到時，又需要花費較多的時間去重新熟悉。于是決定抽點時間總結一下以前對國際搜索離線系統做的一些優化（這里說的國際搜索，主要指AE、SC和SC店鋪，AE即Ali

總覺得階段性的總結是個好習慣，很多自己做的事情，如果不及時總結一下，過一段時間就忘記了，當要用到時，又需要花費較多的時間去重新熟悉。于是決定抽點時間總結一下以前對國際搜索離線系統做的一些優化（這里說的國際搜索，主要指AE、SC和SC店鋪，AE即AliExpress，SC即Sourcing，這些優化對這幾個應用都是通用的），不僅起到一個備忘的作用，如果能給讀者帶來一些啟發，想必也是極好的。

既然是搜索離線系統相關，我們就先看一下國際搜索全量流程的幾個主要環節，如圖1所示。

圖1. 全量流程

1）dump，將數據從數據庫讀出來，寫入hbase，只有做大全量的時候才會全量dump數據庫，一般情況下每天只需跑一次小全量，數據庫中數據的更新會以增量的方式更新hbase。

2）join，讀取hbase，做多表join，生成一條條doc，一條doc包含了一條產品的全部字段。

3）global sort，即全局排序，按產品全局分global_score對產品進行全局排序，生成的單個文件內部并不要求有序。

4）abuild，讀取全局排序后生成的文件，構建索引，生成的索引會存儲在HDFS上。

5）dispatch，將索引從HDFS上分發到對應的search機器上。

6）switch，切換索引、程序、配置和算法詞典，新索引上線，對外提供服務。

這次先總結一下全局排序優化，任何項目或需求都有相應的背景，我們的離線計算中為何要做全局排序？

說到這個，又引出了分層檢索，早些時候，國際站搜索引擎對外提供服務時，在處理每個搜索請求時，都會查詢所有的segment，但其實對于每個請求，都只需返回一定數量的結果集，因此，查詢所有的segment并非必要，只會帶來性能上的損失。于是，分層檢索就在千呼萬喚中出來了。

何謂分層檢索，顧名思義，就是只查詢一定數量的segment，當結果集夠了就不再繼續查詢，這對搜索引擎查詢性能的優化是顯而易見的。

但這里存在一個問題，就是對于賣家發布的產品，質量是良莠不齊的，我們需要把質量好的優先搜索出來，所以前面segment的產品質量要高于后面的segment，否則一些質量高的展品就沒有展示機會了。比如，我們有3個segment，seg_1, seg_2, seg_3，那么seg_1中的產品質量就要比seg_2中的產品質量高，seg_2中的產品質量要比seg_3中的產品質量高，在每個segment內部并不做要求。

判斷產品質量好壞的標準是什么呢？我們引入了一個全局分global_score，每條產品的global_score都是離線計算好的，以此作為分層檢索的依據。

如圖1所示，在搜索引擎的離線計算中，有個多表join的環節，在多表join的過程中會有一些業務邏輯的計算，global_score就是在這個階段計算出來的。有了global_score，我們就可以對產品做全局排序了。假如排序之后我們生成3個文件，part_1, part_2, part_3，就要求part_1中每條doc的global_score要高于part_2中的每條doc，part_2之于part_3亦如此，但每個part內部并不要求有序。在后面建索引的過程中，會有一個保序邏輯，以此保證多個segment之間的有序。

全局排序怎么做呢？由于數據量大，我們各個應用的離線計算任務基本上都是運行在hadoop集群上的，全局排序亦如此。要達到上述的效果，即各個partition之間是按global_score有序的，我們采用的方案是：首先對數據進行采樣，按global_score進行分區，將定義分區的鍵寫入_partitions文件，再實現自定義的TotalOrderPartitioner（這里實現自定義的TotalOrderPartitioner是為了在輸出的單個文件內部將同一家公司的產品聚合在一起，即按company_id聚合，從而大大提高輸出文件的壓縮比，顯著縮短了后面abuild構建索引的運行時間），進行全局排序。采樣的核心思想是只查看一小部分鍵，獲得鍵的近似分布，并由此構建分區。

這里有必要先提一下列的概念，由于單臺search能承載的索引量有限，所以數據量大時，需要對數據進行分列，使所有數據盡量均勻分布到不同的列上。比如SC有19列，采用的做法就是根據product_id % 19將全部數據分布到19列上。在做多表join的之后，數據的分列就已經做好了。因此全局排序是對多列的數據分別進行全局排序。

在分層檢索項目上線到SC BT集群（預發布環境）時，全局排序需要80min才能運行完成，經分析，大部分的時間耗在采樣上面。看了代碼，發現每列的全局排序都對應一個job，SC有19列數據，就跑19個job分別對每列數據進行全局排序。排序之前先采樣，采樣器是在客戶端運行的，因此，分片的下載數量以加速采樣器的運行就顯得尤為重要。在優化之前的代碼實現中，每個job都是讀取對應列的數據，自己采樣的，而且多個job是串行采樣。因此，一個可行的優化方案就是多個job并行采樣，但由于我們的產品數據是分列存儲的，每一列的數據量也足夠大。比如SC現在3.6億的數據量，單列的數據就接近2千萬，因此其實每一列產品global_score的分布是基本一致的，所以，我們是否可以只對一列數據進行采樣，然后所有job都共享這一個樣本呢？這樣就不僅能大大縮短采樣時間，而且也不會引入并行的復雜性。答案是可行的。

簡單的說，全局排序優化的基本思想，就是根據數據的分布特點，使多列數據的多個全局排序job共享同一個樣本。

下面我們來看一下優化后的代碼實現：

Vector vecRunningJob = new Vector(build_num);
Vector vecJobClient = new Vector(build_num);
for (int j = 0; j < build_num; j++) {
 job.setJobName("Doc Sort job" + String.valueOf(j));
 job.setInt("dc.sort.jobindex", j);
 Vector vecInput = fileGenerator.getInPutFiles(j, build_num);
 JobConf newjob = makeJob(job, inputPath, vecInput, outputPath + "/" + j, aggregateField); // Make a job for each column
 JobClient jc = new JobClient(newjob);
 vecJobClient.add(jc);
 vecRunningJob.add(jc.submitJob(newjob));
}

其中build_num表示列數，從上面的代碼可以看出，對每列數據都會調用makeJob方法，然后提交任務進行全局排序。注意這里調用makeJob方法和提交任務是串行的，不過任務提交后是并行跑的。

?我們再看一下makeJob方法的實現：

private static JobConf makeJob(JobConf basejob, String inputPath,
 Vector vecInPutFile, String outPutPath, String aggregateField) throws Exception {
 JobConf conf = new JobConf(basejob);
 conf.setJarByClass(DCSortMain.class);
 for (int i = 0; i < vecInPutFile.size(); i++) {
 FileInputFormat.addInputPath(conf, new Path(vecInPutFile.get(i)));
 }
 Path outputDir = new Path(outPutPath);
 FileOutputFormat.setOutputPath(conf, outputDir);
 conf.setMapOutputKeyClass(DCText.class);
 conf.setMapOutputValueClass(DCText.class);
 conf.setOutputKeyClass(DCText.class);
 conf.setOutputValueClass(DCText.class);
 conf.setMapperClass(IdentityMapper.class);
 conf.setReducerClass(IdentityReducer.class);
 conf.setInputFormat(DCTextInputFormat.class);
 conf.setOutputFormat(DCTextOutputFormat.class);
 conf.set("mapred.output.compress", "true");
 conf.set("mapred.output.compression.codec", "org.apache.hadoop.io.compress.GzipCodec");
 conf.setOutputKeyComparatorClass(DCText.AggregateFieldComparator.class); // Sort numbericly by desc
 conf.setNumReduceTasks(conf.getInt("dc.sort.reduce_num", 1));
 sample(conf, inputPath); // Sample before global sort
 return conf;
}

可見，在做好相關設置后，makeJob中會調用sample方法進行采樣，也就是說，其實針對每一列的makeJob都會調用sample方法。

再來看看sample方法的實現：

private static void sample(JobConf conf, String inputPath) throws IOException, URISyntaxException {
 int jobIndex = 0;
 Path partitionFile = new Path(inputPath, jobIndex + "_partitions");
 conf.setPartitionerClass(MyTotalOrderPartitioner.class);
 conf.set("total.order.partitioner.natural.order", "false");
 MyTotalOrderPartitioner.setPartitionFile(conf, partitionFile);
 if (!sampleDone) {
 LOG.info("sample start ...");
 MyInputSampler.Sampler sampler =
 new MyInputSampler.RandomSampler(1, 20000, 10);
 MyInputSampler.writePartitionFile(conf, sampler);
 LOG.info("sample end ...");
 sampleDone = true;
 }
 // Add to DistributedCache
 URI partitionUri = new URI(partitionFile.toString() + "#" + jobIndex + "_partitions");
 DistributedCache.addCacheFile(partitionUri, conf);
 DistributedCache.createSymlink(conf);
} 

可以看出，我們引入了一個布爾變量sampleDone對采樣進行了控制，只在第1次調用makeJob方法時才執行采樣操作，后面的創建的job都不再進行采樣，而是與第1個job共享同一個_partitions文件，載入到自己使用的分布式緩存中，供后面的全局排序使用。sampleDone定義如下：

private static boolean sampleDone = false; 

順便提一下采樣操作，hadoop內置的采樣器有3個：

1）RandomSampler，以指定的采樣率均勻地從一個數據集中選擇樣本；

2）SplitSampler，只采樣一個分片中的前n個記錄；

3）IntervalSampler，以一定的間隔定期從劃分中選擇鍵，對于已排好序的數據來說是一個更好的選擇。

RandomSampler是優秀的通用采樣器，我們最終也是選擇RandomSampler，因為雖然使用另外兩個采用器，采樣時間更短，但最終數據分布卻很不均勻，只有RandomSampler才能達到預期效果。同時，我們將采樣率設置為1，最大樣本數設置為20000，最大分區設置為10。最大樣本數和最大分區只需滿足其一，即停止采樣。可以通過調整RandomSampler的這些參數達到不同的采樣效果。

優化版本上線SC BT之后，全局排序的運行時間從80min降到了30min，縮短了50min。正式環境由于hadoop集群更加強大，全局排序的運行時間更短。

原文地址：國際搜索離線系統優化之一 —— 全局排序優化, 感謝原作者分享。

聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。TEL:177 7030 7066 E-MAIL:11247931@qq.com