Hive、MapReduce、Spark分佈式生成唯一數值型ID

在實際業務場景下，經常會遇到在Hive、MapReduce、Spark中需要生成唯一的數值型ID。

一般常用的做法有：

MapReduce中使用1個Reduce來生成;

Hive中使用row_number分析函數來生成，其實也是1個Reduce;

藉助HBase或Redis或Zookeeper等其它框架的計數器來生成;

數據量不大的情況下，可以直接使用1和2方法來生成，但如果數據量巨大，1個Reduce處理起來就非常慢。

在數據量非常大的情況下，如果你僅僅需要唯一的數值型ID，注意：不是需要”連續的唯一的數值型ID”，那麼可以考慮採用本文中介紹的方法，否則，請使用第3種方法來完成。

Spark中生成這樣的非連續唯一數值型ID，非常簡單，直接使用zipWithUniqueId即可。

參考zipWithUniqueId的方法，在MapReduce和Hive中，實現如下：

在Spark中，zipWithUniqueId是通過使用分區Index作為每個分區ID的開始值，在每個分區內，ID增長的步長為該RDD的分區數，那麼在MapReduce和Hive中，也可以照此思路實現，Spark中的分區數，即為MapReduce中的Map數，Spark分區的Index，即為Map Task的ID。Map數，可以通過JobConf的getNumMapTasks，而Map Task ID，可以通過參數mapred.task.id獲取，格式如：attempt_1478926768563_0537_m_000004_0，截取m_000004_0中的4，再加1，作為該Map Task的ID起始值。注意：這兩個只均需要在Job運行時才能獲取。另外，從圖中也可以看出，每個分區/Map Task中的數據量不是絕對一致的，因此，生成的ID不是連續的。

下面的UDF可以在Hive中直接使用：

1. package com.lxw1234.hive.udf;
2. import org.apache.hadoop.hive.ql.exec.MapredContext;
3. import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
4. import org.apache.hadoop.hive.ql.metadata.HiveException;
5. import org.apache.hadoop.hive.ql.udf.UDFType;
6. import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
7. import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
8. import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
9. import org.apache.hadoop.io.LongWritable;
10. @UDFType(deterministic = false, stateful = true)
11. public class RowSeq2 extends GenericUDF {
12. private static LongWritable result = new LongWritable;
13. private static final char SEPARATOR = '_';
14. private static final String ATTEMPT = "attempt";
15. private long initID = 0l;
16. private int increment = 0;
17. @Override
18. public void configure(MapredContext context) {
19. increment = context.getJobConf.getNumMapTasks;
20. if(increment == 0) {
21. throw new IllegalArgumentException("mapred.map.tasks is zero");
22. }
23. initID = getInitId(context.getJobConf.get("mapred.task.id"),increment);
24. if(initID == 0l) {
25. throw new IllegalArgumentException("mapred.task.id");
26. }
27. System.out.println("initID : " + initID + " increment : " + increment);
28. }
29. @Override
30. public ObjectInspector initialize(ObjectInspector[] arguments)
31. throws UDFArgumentException {
32. return PrimitiveObjectInspectorFactory.writableLongObjectInspector;
33. }
34. @Override
35. public Object evaluate(DeferredObject[] arguments) throws HiveException {
36. result.set(getValue);
37. increment(increment);
38. return result;
39. }
40. @Override
41. public String getDisplayString(String[] children) {
42. return "RowSeq-func";
43. }
44. private synchronized void increment(int incr) {
45. initID += incr;
46. }
47. private synchronized long getValue {
48. return initID;
49. }
50. //attempt_1478926768563_0537_m_000004_0 // return 0+1
51. private long getInitId (String taskAttemptIDstr,int numTasks)
52. throws IllegalArgumentException {
53. try {
54. String parts = taskAttemptIDstr.split(Character.toString(SEPARATOR));
55. if(parts.length == 6) {
56. if(parts[0].equals(ATTEMPT)) {
57. if(!parts[3].equals("m") && !parts[3].equals("r")) {
58. throw new Exception;
59. }
60. long result = Long.parseLong(parts[4]);
61. if(result >= numTasks) { //if taskid >= numtasks
62. throw new Exception("TaskAttemptId string : " + taskAttemptIDstr
63. + " parse ID [" + result + "] >= numTasks[" + numTasks + "] ..");
64. }
65. return result + 1;
66. }
67. }
68. } catch (Exception e) {}
69. throw new IllegalArgumentException("TaskAttemptId string : " + taskAttemptIDstr
70. + " is not properly formed");
71. }
72. }

有一張去重後的用戶id(字符串類型)表，需要位每個用戶id生成一個唯一的數值型seq:

1. ADD jar file:///tmp/udf.jar;
2. CREATE temporary function seq2 as 'com.lxw1234.hive.udf.RowSeq2';
3. hive>> desc lxw_all_ids;
4. OK
5. id string
6. Time taken: 0.074 seconds, Fetched: 1 row(s)
7. hive> select * from lxw_all_ids limit 5;
8. OK
9. 01779E7A06ABF5565A4982_cookie
10. 031E2D2408C29556420255_cookie
11. 03371ADA0B6E405806FFCD_cookie
12. 0517C4B701BC1256BFF6EC_cookie
13. 05F12ADE0E880455931C1A_cookie
14. Time taken: 0.215 seconds, Fetched: 5 row(s)
15. hive> select count(1) from lxw_all_ids;
16. 253402337
17. hive> create table lxw_all_ids2 as select id,seq2 as seq from lxw_all_ids;
18. …
19. Hadoop job information for Stage-1: number of mappers: 27; number of reducers: 0
20. …

該Job使用了27個Map Task，沒有使用Reduce，那麼將會產生27個結果文件。

再看結果表中的數據：

1. hive> select * from lxw_all_ids2 limit 10;
2. OK
3. 766CA2770527B257D332AA_cookie 1
4. 5A0492DB0000C557A81383_cookie 28
5. 8C06A5770F176E58301EEF_cookie 55
6. 6498F47B0BCAFE5842B83A_cookie 82
7. 6DA33CB709A23758428A44_cookie 109
8. B766347B0D27925842AC2D_cookie 136
9. 5794357B050C99584251AC_cookie 163
10. 81D67A7B011BEA5842776C_cookie 190
11. 9D2F8EB40AEA525792347D_cookie 217
12. BD21077B09F9E25844D2C1_cookie 244
13. hive> select count(1),count(distinct seq) from lxw_all_ids2;
14. 253402337 253402337

limit 10只從第一個結果文件，即MapTaskId為0的結果文件中拿了10條，這個Map中，start=1，increment=27，因此生成的ID如上所示。

count(1),count(distinct seq)的值相同，說明seq沒有重複值，你可以試試max(seq)，結果必然大於253402337，說明seq是”非連續唯一數值型ID“.