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中直接使用： packagecom.lxw1234.hive.udf; importorg.apache.hadoop.hive.ql.exec.MapredContext; importorg.apache.hadoop.hive.ql.exec.UDFArgumentException; importorg.apache.hadoop.hive.ql.metadata.HiveException; importorg.apache.hadoop.hive.ql.udf.UDFType; importorg.apache.hadoop.hive.ql.udf.generic.GenericUDF; importorg.apache.hadoop.hive.serde2.objectinspector.ObjectInspector; importorg.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory; importorg.apache.hadoop.io.LongWritable; @UDFType(deterministic=false,stateful=true) publicclassRowSeq2extendsGenericUDF{ privatestaticLongWritableresult=newLongWritable(); privatestaticfinalcharSEPARATOR='_'; privatestaticfinalStringATTEMPT="attempt"; privatelonginitID=0l; privateintincrement=0; @Override publicvoidconfigure(MapredContextcontext){ increment=context.getJobConf().getNumMapTasks(); if(increment==0){ thrownewIllegalArgumentException("mapred.map.tasksiszero"); } initID=getInitId(context.getJobConf().get("mapred.task.id"),increment); if(initID==0l){ thrownewIllegalArgumentException("mapred.task.id"); } System.out.println("initID:"+initID+"increment:"+increment); } @Override publicObjectInspectorinitialize(ObjectInspector[]arguments) throwsUDFArgumentException{ returnPrimitiveObjectInspectorFactory.writableLongObjectInspector; } @Override publicObjectevaluate(DeferredObject[]arguments)throwsHiveException{ result.set(getValue()); increment(increment); returnresult; } @Override publicStringgetDisplayString(String[]children){ return"RowSeq-func()"; } privatesynchronizedvoidincrement(intincr){ initID+=incr; } privatesynchronizedlonggetValue(){ returninitID; } //attempt_1478926768563_0537_m_000004_0//return0+1 privatelonggetInitId(StringtaskAttemptIDstr,intnumTasks) throwsIllegalArgumentException{ try{ String[]parts=taskAttemptIDstr.split(Character.toString(SEPARATOR)); if(parts.length==6){ if(parts[0].equals(ATTEMPT)){ if(!parts[3].equals("m")&&!parts[3].equals("r")){ thrownewException(); } longresult=Long.parseLong(parts[4]); if(result>=numTasks){//iftaskid>=numtasks thrownewException("TaskAttemptIdstring:"+taskAttemptIDstr +"parseID["+result+"]>=numTasks["+numTasks+"].."); } returnresult+1; } } }catch(Exceptione){} thrownewIllegalArgumentException("TaskAttemptIdstring:"+taskAttemptIDstr +"isnotproperlyformed"); } } 有一张去重后的用户id(字符串类型)表，需要位每个用户id生成一个唯一的数值型seq: ADDjarfile:///tmp/udf.jar; CREATEtemporaryfunctionseq2as'com.lxw1234.hive.udf.RowSeq2'; hive>>desclxw_all_ids; OK idstring Timetaken:0.074seconds,Fetched:1row(s) hive>select*fromlxw_all_idslimit5; OK 01779E7A06ABF5565A4982_cookie 031E2D2408C29556420255_cookie 03371ADA0B6E405806FFCD_cookie 0517C4B701BC1256BFF6EC_cookie 05F12ADE0E880455931C1A_cookie Timetaken:0.215seconds,Fetched:5row(s) hive>selectcount(1)fromlxw_all_ids; 253402337 hive>createtablelxw_all_ids2asselectid,seq2()asseqfromlxw_all_ids; … HadoopjobinformationforStage-1:numberofmappers:27;numberofreducers:0 … 该Job使用了27个Map Task，没有使用Reduce，那么将会产生27个结果文件。 再看结果表中的数据： hive>select*fromlxw_all_ids2limit10; OK 766CA2770527B257D332AA_cookie1 5A0492DB0000C557A81383_cookie28 8C06A5770F176E58301EEF_cookie55 6498F47B0BCAFE5842B83A_cookie82 6DA33CB709A23758428A44_cookie109 B766347B0D27925842AC2D_cookie136 5794357B050C99584251AC_cookie163 81D67A7B011BEA5842776C_cookie190 9D2F8EB40AEA525792347D_cookie217 BD21077B09F9E25844D2C1_cookie244 hive>selectcount(1),count(distinctseq)fromlxw_all_ids2; 253402337253402337 limit 10只从第一个结果文件，即MapTaskId为0的结果文件中拿了10条，这个Map中，start=1，increment=27，因此生成的ID如上所示。 count(1),count(distinct seq)的值相同，说明seq没有重复值，你可以试试max(seq)，结果必然大于253402337，说明seq是”非连续唯一数值型ID“. 本文作者：佚名 来源：51CTO
