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

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


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

该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.   
  14. hive> select count(1),count(distinct seq) from lxw_all_ids2; 
  15. 253402337       253402337 
  16.   

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

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


本文作者:佚名

来源:51CTO

相关文章
|
1月前
|
分布式计算 资源调度 Hadoop
大数据-80 Spark 简要概述 系统架构 部署模式 与Hadoop MapReduce对比
大数据-80 Spark 简要概述 系统架构 部署模式 与Hadoop MapReduce对比
62 2
|
3月前
|
算法 Go
[go 面试] 雪花算法与分布式ID生成
[go 面试] 雪花算法与分布式ID生成
|
11天前
|
NoSQL 算法 关系型数据库
分布式 ID 详解 ( 5大分布式 ID 生成方案 )
本文详解分布式全局唯一ID及其5种实现方案,关注【mikechen的互联网架构】,10年+BAT架构经验倾囊相授。
分布式 ID 详解 ( 5大分布式 ID 生成方案 )
|
9天前
|
分布式计算 Java 开发工具
阿里云MaxCompute-XGBoost on Spark 极限梯度提升算法的分布式训练与模型持久化oss的实现与代码浅析
本文介绍了XGBoost在MaxCompute+OSS架构下模型持久化遇到的问题及其解决方案。首先简要介绍了XGBoost的特点和应用场景,随后详细描述了客户在将XGBoost on Spark任务从HDFS迁移到OSS时遇到的异常情况。通过分析异常堆栈和源代码,发现使用的`nativeBooster.saveModel`方法不支持OSS路径,而使用`write.overwrite().save`方法则能成功保存模型。最后提供了完整的Scala代码示例、Maven配置和提交命令,帮助用户顺利迁移模型存储路径。
|
1月前
|
SQL 分布式计算 Java
大数据-96 Spark 集群 SparkSQL Scala编写SQL操作SparkSQL的数据源:JSON、CSV、JDBC、Hive
大数据-96 Spark 集群 SparkSQL Scala编写SQL操作SparkSQL的数据源:JSON、CSV、JDBC、Hive
33 0
|
1月前
|
SQL 分布式计算 关系型数据库
Hadoop-24 Sqoop迁移 MySQL到Hive 与 Hive到MySQL SQL生成数据 HDFS集群 Sqoop import jdbc ETL MapReduce
Hadoop-24 Sqoop迁移 MySQL到Hive 与 Hive到MySQL SQL生成数据 HDFS集群 Sqoop import jdbc ETL MapReduce
79 0
|
2月前
|
分布式计算 资源调度 Hadoop
在YARN集群上运行部署MapReduce分布式计算框架
主要介绍了如何在YARN集群上配置和运行MapReduce分布式计算框架,包括准备数据、运行MapReduce任务、查看任务日志,并启动HistoryServer服务以便于日志查看。
57 0
|
4月前
|
SQL 分布式计算 Java
E-MapReduce Serverless Spark体验评测
从了解到部署实践,全方位带你体验大数据平台EMR Serverless Spark的魅力。
297 7
E-MapReduce Serverless Spark体验评测
|
3月前
|
缓存 分布式计算 Java
详细解读MapReduce框架中的分布式缓存
【8月更文挑战第31天】
31 0
|
4月前
|
分布式计算 监控 Serverless
E-MapReduce Serverless Spark 版测评
E-MapReduce Serverless Spark 版测评
11599 10