189 8069 5689

Hive、MapReduce、Spark分布式生成唯一数值型ID

Hive、MapReduce、Spark分布式生成唯一数值型ID

作者：佚名 2017-04-12 09:29:02

大数据

分布式

Spark Spark中生成这样的非连续唯一数值型ID，非常简单，直接使用zipWithUniqueId()即可。

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

有一张去重后的用户id(字符串类型)表，需要位每个用户id生成一个唯一的数值型seq:

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

该Job使用了27个Map Task，没有使用Reduce，那么将会产生27个结果文件。

再看结果表中的数据：

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

limit 10只从第一个结果文件，即MapTaskId为0的结果文件中拿了10条，这个Map中，start=1，increment=27，因此生成的ID如上所示。

count(1),count(distinct seq)的值相同，说明seq没有重复值，你可以试试max(seq)，结果必然大于253402337，说明seq是”非连续唯一数值型ID“.


网站题目：Hive、MapReduce、Spark分布式生成唯一数值型ID
转载注明：http://www.cdxtjz.cn/article/dhscdig.html

其他资讯