1.hive.groupby.skewindata及数据倾斜优化
2.yarn源码分析（四）AppMaster启动
3.å¦ä½åå¸å¼è¿è¡mapreduceç¨åº
4.10本大数据框架Hadoop学习书籍推荐
5.Mapreduce 在通过reduce计算value之后怎么统计计算次数？
6.å¦ä½ä½¿ç¨Python为Hadoopç¼åä¸ä¸ªç®åçMapReduceç¨åº
hive.groupby.skewindata及数据倾斜优化
一、码阅hive.groupby.skewindata
当数据倾斜出现时，码阅hive.groupby.skewindata功能介入均衡负载。码阅此设置为true时，码阅生成的码阅查询计划包含两个MapReduce任务。第一个任务中，码阅源码项目有哪些Map端输出结果集合随机分发给Reduce，码阅每个Reduce执行部分聚合操作并输出结果。码阅这样处理后，码阅相同的码阅GroupBy Key可能被分发到不同的Reduce中，实现负载均衡。码阅紧接着的码阅第二个任务，根据预处理数据结果按照GroupBy Key分布到Reduce，码阅最终完成聚合操作。码阅
二、码阅数据倾斜原因分析
数据倾斜常见于Join和Group by操作。Join操作原理：将两个表通过某个共同字段进行连接，形成结果集。模拟 键盘 驱动 源码Group by操作原理：根据指定字段对数据进行分组，并对每个分组执行聚合计算。
三、数据倾斜解决方案
解决方法一：Join数据倾斜。通过使用MapJoin原理，提高Join操作效率，减轻数据倾斜影响。具体实现可参考相关博文，浅谈Hive中Map Join原理及场景。
解决方法二：Group by数据倾斜。在Map阶段执行部分聚集操作，提高效率，但需注意内存使用。开启Map后，使用combiner功能，对于同质数据效果显著。一般推荐设置两个关键参数。大灰狼8.23源码另外，设置hive.groupby.skewindata=true也是有效手段。注意，有博文中提到与set hive.map.aggr=true同时使用时可能对distinct结果有影响，但实际测试并未发现此问题。特定情况如有疑问，请随时交流。
参考资源：
1. hive数据倾斜原理与解决方案
2. hive原理与源码分析-算子Operators及查询优化器Optimizers（四）_wzq的专栏-CSDN博客
欢迎学习交流，有任何问题请随时评论指出。
yarn源码分析（四）AppMaster启动
在容器分配完成之后，启动容器的代码主要在ContainerImpl.java中进行。通过状态机转换，container从NEW状态向其他状态转移时，会调用RequestResourceTransition对象。RequestResourceTransition负责将所需的资源进行本地化，或者避免资源本地化。若需本地化，vc操作excel源码还需过渡到LOCALIZING状态。为简化理解，此处仅关注是否进行资源本地化的情况。
为了将LAUNCH_CONTAINER事件加入事件处理队列，调用了sendLaunchEvent方法。该事件由ContainersLauncher负责处理。ContainersLauncher的handle方法中，使用一个ExecutorService（线程池）容器Launcher。ContainerLaunch实现了Callable接口，其call方法生成并执行launch_container脚本。以MapReduce框架为例，该脚本在hadoop.tmp.dir/application name/container name目录下生成，其主要作用是启动MRAppMaster进程，即MapReduce的ApplicationMaster。
å¦ä½åå¸å¼è¿è¡mapreduceç¨åº
ä¸ã é¦å è¦ç¥éæ¤åæ 转载
ããè¥å¨windowsçEclipseå·¥ç¨ä¸ç´æ¥å¯å¨mapreducç¨åºï¼éè¦å æhadoopé群çé ç½®ç®å½ä¸çxmlé½æ·è´å°srcç®å½ä¸ï¼è®©ç¨åºèªå¨è¯»åé群çå°ååå»è¿è¡åå¸å¼è¿è¡(æ¨ä¹å¯ä»¥èªå·±åjava代ç å»è®¾ç½®jobçconfigurationå±æ§)ã
ããè¥ä¸æ·è´ï¼å·¥ç¨ä¸binç®å½æ²¡æå®æ´çxmlé ç½®æ件ï¼åwindowsæ§è¡çmapreduceç¨åºå ¨é¨éè¿æ¬æºçjvmæ§è¡ï¼ä½ä¸åä¹æ¯å¸¦æâlocal"åç¼çä½ä¸ï¼å¦ job_local_ã è¿ä¸æ¯çæ£çåå¸å¼è¿è¡mapreduceç¨åºã
ãã估计å¾ç 究org.apache.hadoop.conf.Configurationçæºç ï¼åæ£xmlé ç½®æ件ä¼å½±åæ§è¡mapreduce使ç¨çæ件系ç»æ¯æ¬æºçwindowsæ件系ç»è¿æ¯è¿ç¨çhdfsç³»ç»; è¿æå½±åæ§è¡mapreduceçmapperåreducerçæ¯æ¬æºçjvmè¿æ¯é群éé¢æºå¨çjvm
ããäºã æ¬æçç»è®º
ãã第ä¸ç¹å°±æ¯ï¼ windowsä¸æ§è¡mapreduceï¼å¿ é¡»æjarå å°ææslaveèç¹æè½æ£ç¡®åå¸å¼è¿è¡mapreduceç¨åºãï¼æ个éæ±æ¯è¦windowsä¸è§¦åä¸ä¸ªmapreduceåå¸å¼è¿è¡ï¼
ãã第äºç¹å°±æ¯ï¼ Linuxä¸ï¼åªéæ·è´jaræ件å°é群masterä¸,æ§è¡å½ä»¤hadoop jarPackage.jar MainClassNameå³å¯åå¸å¼è¿è¡mapreduceç¨åºã
ãã第ä¸ç¹å°±æ¯ï¼ æ¨è使ç¨éä¸ï¼å®ç°äºèªå¨æjarå 并ä¸ä¼ ï¼åå¸å¼æ§è¡çmapreduceç¨åºã
ããéä¸ã æ¨è使ç¨æ¤æ¹æ³ï¼å®ç°äºèªå¨æjarå 并ä¸ä¼ ï¼åå¸å¼æ§è¡çmapreduceç¨åºï¼
ãã请å åèåæäºç¯ï¼
ããHadoopä½ä¸æ交åæï¼ä¸ï¼~~ï¼äºï¼
ããå¼ç¨åæçé件ä¸EJob.javaå°å·¥ç¨ä¸ï¼ç¶åmainä¸æ·»å å¦ä¸æ¹æ³å代ç ã
ããpublic static File createPack() throws IOException {
ããFile jarFile = EJob.createTempJar("bin");
ããClassLoader classLoader = EJob.getClassLoader();
ããThread.currentThread().setContextClassLoader(classLoader);
ããreturn jarFile;
ãã}
ããå¨ä½ä¸å¯å¨ä»£ç ä¸ä½¿ç¨æå ï¼
ããJob job = Job.getInstance(conf, "testAnaAction");
ããæ·»å ï¼
ããString jarPath = createPack().getPath();
ããjob.setJar(jarPath);
ããå³å¯å®ç°ç´æ¥run as java application å¨windowsè·åå¸å¼çmapreduceç¨åºï¼ä¸ç¨æå·¥ä¸ä¼ jaræ件ã
ããéäºãå¾åºç»è®ºçæµè¯è¿ç¨
ããï¼æªæ空ç书ï¼åªè½éè¿æ笨çæµè¯æ¹æ³å¾åºç»è®ºäºï¼
ããä¸. ç´æ¥éè¿windowsä¸Eclipseå³å»mainç¨åºçjavaæ件ï¼ç¶å"run as application"æéæ©hadoopæ件"run on hadoop"æ¥è§¦åæ§è¡MapReduceç¨åºçæµè¯ã
ãã1ï¼å¦æä¸æjarå å°è¿é群任ælinuxæºå¨ä¸ï¼å®æ¥éå¦ä¸ï¼
ãã[work] -- ::, - org.apache.hadoop.mapreduce.Job - [main] INFO org.apache.hadoop.mapreduce.Job - map 0% reduce 0%
ãã[work] -- ::, - org.apache.hadoop.mapreduce.Job - [main] INFO org.apache.hadoop.mapreduce.Job - Task Id : attempt___m__0, Status : FAILED
ããError: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountMapper not found
ããat org.apache.hadoop.conf.Configuration.getClass(Configuration.java:)
ããat org.apache.hadoop.mapreduce.task.JobContextImpl.getMapperClass(JobContextImpl.java:)
ããat org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:)
ããat org.apache.hadoop.mapred.MapTask.run(MapTask.java:)
ããat org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:)
ããat java.security.AccessController.doPrivileged(Native Method)
ããat javax.security.auth.Subject.doAs(Subject.java:)
ããat org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:)
ããat org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:)
ããCaused by: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountMapper not found
ããat org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:)
ããat org.apache.hadoop.conf.Configuration.getClass(Configuration.java:)
ãã... 8 more
ãã# Error:åéå¤ä¸æ¬¡
ãã-- ::, - org.apache.hadoop.mapreduce.Job - [main] INFO org.apache.hadoop.mapreduce.Job - map % reduce %
ããç°è±¡å°±æ¯ï¼æ¥éï¼æ è¿åº¦ï¼æ è¿è¡ç»æã
ãã
ãã2ï¼æ·è´jarå å°âåªæ¯âé群masterç$HADOOP_HOME/share/hadoop/mapreduce/ç®å½ä¸ï¼ç´æ¥éè¿windowsçeclipse "run as application"åéè¿hadoopæ件"run on hadoop"æ¥è§¦åæ§è¡ï¼å®æ¥éåä¸ã
ããç°è±¡å°±æ¯ï¼æ¥éï¼æ è¿åº¦ï¼æ è¿è¡ç»æã
ãã3ï¼æ·è´jarå å°é群æäºslaveç$HADOOP_HOME/share/hadoop/mapreduce/ç®å½ä¸ï¼ç´æ¥éè¿windowsçeclipse "run as application"åéè¿hadoopæ件"run on hadoop"æ¥è§¦åæ§è¡
ããåæ¥éï¼
ããError: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountMapper not found
ããat org.apache.hadoop.conf.Configuration.getClass(Configuration.java:)
ããat org.apache.hadoop.mapreduce.task.JobContextImpl.getMapperClass(JobContextImpl.java:)
ããåæ¥éï¼
ããError: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class bookCount.BookCount$BookCountReducer not found
ãã
ããç°è±¡å°±æ¯ï¼ææ¥éï¼ä½ä»ç¶æè¿åº¦ï¼æè¿è¡ç»æã
本大数据框架Hadoop学习书籍推荐
Hadoop，一个用Java编写的Apache开源框架，旨在分布式处理大型数据集。装逼软件源码它简化了编程模型，让用户在无需了解分布式底层细节的情况下开发分布式程序。Hadoop成为大数据处理平台的首选，广泛应用于各种生产环境。以下是本关于Hadoop学习的推荐书籍，涵盖了从入门到深入的各个方面。 《Hadoop权威指南》 本书结合理论与实践，由浅入深地介绍了Hadoop这一高性能的海量数据处理和分析平台。读者能探索如何利用Hadoop分析大量数据集，以及如何安装和运行Hadoop集群。 《Hadoop技术内幕：深入解析MapReduce架构设计与实现原理》 本书从源代码的角度对MapReduce的架构设计与实现原理进行了详细的解析。适合Hadoop的二次开发人员、应用开发工程师和运维工程师阅读。 《Hadoop技术内幕：深入解析Hadoop Common和HDFS架构设计与实现原理》 本书同样以源代码为基础，深入剖析了Common和HDFS的架构设计与实现原理，适合相关领域专业人士阅读。 《Hadoop技术内幕：深入解析YARN架构设计与实现原理》 本书系统讲解了YARN的基本库和组件用法、应用程序设计方法、以及YARN上流行的计算框架。适合对YARN有深入研究的读者。 《深入理解Hadoop》 作者基于实践经验，深入浅出地讲解了Hadoop框架，包含大量实例和技巧，帮助开发者快速掌握分布式系统。 《Hadoop 2.X HDFS源码剖析》 本书基于Hadoop 2.6.0源码，详细剖析了HDFS 2.X中各个模块的实现细节，适合从架构设计和源码实现角度了解HDFS的读者。 《Hadoop实战》 本书深入浅出地介绍了Hadoop框架和编写数据处理程序所需的实践技能，适合需要处理大量离线数据的云计算程序员、架构师和项目经理。 《Hadoop海量数据处理：技术详解与项目实战》 本书从理论到实践，适合Hadoop初学者，也可作为高等院校相关课程的参考教材。 《Hadoop基础教程》 本书着重讲解了如何搭建Hadoop工作系统并完成任务，适合对Hadoop有初步了解的读者。 《Hadoop构建数据仓库实践》 本书适合数据库管理员、大数据技术人员、Hadoop技术人员和数据仓库技术人员，也是高等院校相关专业的教学参考。 《Hadoop应用架构》 本书提供了专业的架构指导，适用于设计Hadoop应用或集成Hadoop到现有数据基础架构的读者。 《Hadoop技术详解》 本书全面介绍Hadoop的各项操作，从设计到安装和设置，帮助读者构建稳定可靠的系统。适合希望深入理解Hadoop工作原理的开发者。Mapreduce 在通过reduce计算value之后怎么统计计算次数？
简单，不知道你看没看过Wordcount源码，其中的统计出现次数是传入一个1，通过reduce相加计算得出次数。我可以通过Map传入value时拼接一个1，在reduce中通过拆分字符串得到你要的原valeu和传入的1 ，分别去计算后再拼入输出就可以得到了
å¦ä½ä½¿ç¨Python为Hadoopç¼åä¸ä¸ªç®åçMapReduceç¨åº
MichaelG.Nollå¨ä»çBlogä¸æå°å¦ä½å¨Hadoopä¸ç¨Pythonç¼åMapReduceç¨åºï¼é©å½çgogamzaå¨å ¶Bolgä¸ä¹æå°å¦ä½ç¨Cç¼åMapReduceç¨åºï¼æç¨å¾®ä¿®æ¹äºä¸ä¸åç¨åº,å 为ä»çMap对åè¯åå使ç¨tabé®ï¼ãæå并ä»ä»¬ä¸¤äººçæç« ï¼ä¹è®©å½å çHadoopç¨æ·è½å¤ä½¿ç¨å«çè¯è¨æ¥ç¼åMapReduceç¨åºãããé¦å æ¨å¾é 好æ¨çHadoopé群ï¼è¿æ¹é¢çä»ç»ç½ä¸æ¯è¾å¤ï¼è¿å¿ç»ä¸ªé¾æ¥ï¼Hadoopå¦ä¹ ç¬è®°äºå®è£ é¨ç½²ï¼ãHadoopStreaming帮å©æ们ç¨éJavaçç¼ç¨è¯è¨ä½¿ç¨MapReduceï¼Streamingç¨STDIN(æ åè¾å ¥)åSTDOUT(æ åè¾åº)æ¥åæ们ç¼åçMapåReduceè¿è¡æ°æ®ç交æ¢æ°æ®ãä»»ä½è½å¤ä½¿ç¨STDINåSTDOUTé½å¯ä»¥ç¨æ¥ç¼åMapReduceç¨åºï¼æ¯å¦æ们ç¨Pythonçsys.stdinåsys.stdoutï¼æè æ¯Cä¸çstdinåstdoutãããæ们è¿æ¯ä½¿ç¨Hadoopçä¾åWordCountæ¥å示èå¦ä½ç¼åMapReduceï¼å¨WordCountçä¾åä¸æ们è¦è§£å³è®¡ç®å¨ä¸æ¹ææ¡£ä¸æ¯ä¸ä¸ªåè¯çåºç°é¢çãé¦å æ们å¨Mapç¨åºä¸ä¼æ¥åå°è¿æ¹ææ¡£æ¯ä¸è¡çæ°æ®ï¼ç¶åæ们ç¼åçMapç¨åºæè¿ä¸è¡æç©ºæ ¼åå¼æä¸ä¸ªæ°ç»ã并对è¿ä¸ªæ°ç»éåæ"1"ç¨æ åçè¾åºè¾åºæ¥ï¼ä»£è¡¨è¿ä¸ªåè¯åºç°äºä¸æ¬¡ãå¨Reduceä¸æ们æ¥ç»è®¡åè¯çåºç°é¢çãããããPythonCodeããMap:mapper.pyãã#!/usr/bin/envpythonimportsys#mapswordstotheircountsword2count={ }#inputcomesfromSTDIN(standardinput)forlineinsys.stdin:#removeleadingandtrailingwhitespaceline=line.strip()#splitthelineintowordswhileremovinganyemptystringswords=filter(lambdaword:word,line.split())#increasecountersforwordinwords:#writetheresultstoSTDOUT(standardoutput);#whatweoutputherewillbetheinputforthe#Reducestep,i.e.theinputforreducer.py##tab-delimited;thetrivialwordcountis1print'%s\t%s'%(word,1)ããReduce:reducer.pyãã#!/usr/bin/envpythonfromoperatorimportitemgetterimportsys#mapswordstotheircountsword2count={ }#inputcomesfromSTDINforlineinsys.stdin:#removeleadingandtrailingwhitespaceline=line.strip()#parsetheinputwegotfrommapper.pyword,count=line.split()#convertcount(currentlyastring)tointtry:count=int(count)word2count[word]=word2count.get(word,0)+countexceptValueError:#countwasnotanumber,sosilently#ignore/discardthislinepass#sortthewordslexigraphically;##thisstepisNOTrequired,wejustdoitsothatour#finaloutputwilllookmoreliketheofficialHadoop#wordcountexamplessorted_word2count=sorted(word2count.items(),key=itemgetter(0))#writetheresultstoSTDOUT(standardoutput)forword,countinsorted_word2count:print'%s\t%s'%(word,count)ããCCodeããMap:Mapper.cãã#include#include#include#include#defineBUF_SIZE#defineDELIM"\n"intmain(intargc,char*argv[]){ charbuffer[BUF_SIZE];while(fgets(buffer,BUF_SIZE-1,stdin)){ intlen=strlen(buffer);if(buffer[len-1]=='\n')buffer[len-1]=0;char*querys=index(buffer,'');char*query=NULL;if(querys==NULL)continue;querys+=1;/*nottoinclude'\t'*/query=strtok(buffer,"");while(query){ printf("%s\t1\n",query);query=strtok(NULL,"");}}return0;}h>h>h>h>ããReduce:Reducer.cãã#include#include#include#include#defineBUFFER_SIZE#defineDELIM"\t"intmain(intargc,char*argv[]){ charstrLastKey[BUFFER_SIZE];charstrLine[BUFFER_SIZE];intcount=0;*strLastKey='\0';*strLine='\0';while(fgets(strLine,BUFFER_SIZE-1,stdin)){ char*strCurrKey=NULL;char*strCurrNum=NULL;strCurrKey=strtok(strLine,DELIM);strCurrNum=strtok(NULL,DELIM);/*necessarytocheckerrorbut.*/if(strLastKey[0]=='\0'){ strcpy(strLastKey,strCurrKey);}if(strcmp(strCurrKey,strLastKey)){ printf("%s\t%d\n",strLastKey,count);count=atoi(strCurrNum);}else{ count+=atoi(strCurrNum);}strcpy(strLastKey,strCurrKey);}printf("%s\t%d\n",strLastKey,count);/*flushthecount*/return0;}h>h>h>h>ããé¦å æ们è°è¯ä¸ä¸æºç ï¼ããchmod+xmapper.pychmod+xreducer.pyecho"foofooquuxlabsfoobarquux"|./mapper.py|./reducer.pybar1foo3labs1quux2g++Mapper.c-oMapperg++Reducer.c-oReducerchmod+xMapperchmod+xReducerecho"foofooquuxlabsfoobarquux"|./Mapper|./Reducerbar1foo2labs1quux1foo1quux1ããä½ å¯è½çå°Cçè¾åºåPythonçä¸ä¸æ ·,å 为Pythonæ¯æä»æ¾å¨è¯å ¸éäº.æ们å¨Hadoopæ¶,ä¼å¯¹è¿è¿è¡æåº,ç¶åç¸åçåè¯ä¼è¿ç»å¨æ åè¾åºä¸è¾åº.ããå¨Hadoopä¸è¿è¡ç¨åºããé¦å æ们è¦ä¸è½½æ们çæµè¯ææ¡£wget页é¢ä¸æä¸çç¨phpç¼åçMapReduceç¨åº,ä¾phpç¨åºååèï¼Map:mapper.phpãã#!/usr/bin/php$word2count=array();//inputcomesfromSTDIN(standardinput)while(($line=fgets(STDIN))!==false){ //removeleadingandtrailingwhitespaceandlowercase$line=strtolower(trim($line));//splitthelineintowordswhileremovinganyemptystring$words=preg_split('/\W/',$line,0,PREG_SPLIT_NO_EMPTY);//increasecountersforeach($wordsas$word){ $word2count[$word]+=1;}}//writetheresultstoSTDOUT(standardoutput)//whatweoutputherewillbetheinputforthe//Reducestep,i.e.theinputforreducer.pyforeach($word2countas$word=>$count){ //tab-delimitedecho$word,chr(9),$count,PHP_EOL;}?>ããReduce:mapper.phpãã#!/usr/bin/php$word2count=array();//inputcomesfromSTDINwhile(($line=fgets(STDIN))!==false){ //removeleadingandtrailingwhitespace$line=trim($line);//parsetheinputwegotfrommapper.phplist($word,$count)=explode(chr(9),$line);//convertcount(currentlyastring)toint$count=intval($count);//sumcountsif($count>0)$word2count[$word]+=$count;}//sortthewordslexigraphically////thissetisNOTrequired,wejustdoitsothatour//finaloutputwilllookmoreliketheofficialHadoop//wordcountexamplesksort($word2count);//writetheresultstoSTDOUT(standardoutput)foreach($word2countas$word=>$count){ echo$word,chr(9),$count,PHP_EOL;}?>ããä½è ï¼é©¬å£«åå表äºï¼--
MapReduce源码解析之InputFormat
导读
深入探讨MapReduce框架的核心组件——InputFormat。此组件在处理多样化数据类型时，扮演着数据格式化和分片的角色。通过设置job.setInputFormatClass(TextInputFormat.class)等操作，程序能正确处理不同文件类型。InputFormat类作为抽象基础，定义了文件切分逻辑和RecordReader接口，用于读取分片数据。本节将解析InputFormat、InputSplit、RecordReader的结构与实现，以及如何在Map任务中应用此框架。
类图与源码解析
InputFormat类提供了两个关键抽象方法：getSplits()和createRecordReader()。getSplits()负责规划文件切分策略，定义逻辑上的分片，而RecordReader则从这些分片中读取数据。
InputSplit类承载了切分逻辑，表示了给定Mapper处理的逻辑数据块，包含所有K-V对的集合。
RecordReader类实现了数据读取流程，其子类如LineRecordReader，提供行数据读取功能，将输入流中的数据按行拆分，赋值为Key和Value。
具体实现与操作流程
在getSplits()方法中，FileInputFormat类负责将输入文件按照指定策略切分成多个InputSplit。
TextInputFormat类的createRecordReader()方法创建了LineRecordReader实例，用于读取文件中的每一行数据，形成K-V对。
Mapper任务执行时，通过调用RecordReader的nextKeyValue()方法，读取文件的每一行，完成数据处理。
在Map任务的run()方法中，MapContextImp类实例化了一个RecordReader，用于实现数据的迭代和处理。
总结
本文详细阐述了MapReduce框架中InputFormat的实现原理及其相关组件，包括类图、源码解析、具体实现与操作流程。后续文章将继续探讨MapReduce框架的其他关键组件源码解析，为开发者提供深入理解MapReduce的构建和优化方法。
