初探hadoop框架

迷南。 2022-06-06 05:54 218阅读 0赞

首先做好前期的准备：  
安装hadoop集群，我是采用的伪分布式(练习使用)，四台虚拟机，  
一台为主(master)，三台为从(slave)。

因为是练习，所以就在主节点master上练习作业，没有在客户端访问集群，

语言选择的是python

任务：将一个txt文件中的单词进行计数，然后输出到集群上。

四个文件：  
一个源文件xxx.txt  
一个map文件:map.py  
一个reduce稳健：red.py  
一个run.sh(采用shell脚本编写)

**map.py**

#!/usr/bin/env python
    
    import sys
    
    for line in sys.stdin:
        ss = line.strip().split(" ")
        for s in ss:
            print '\t'.join([s.strip(),'1'])

**red.py**

#!/usr/bin/env python
    
    import sys
    
    sum=0
    count_pool=[]
    current_word=None
    
    for line in sys.stdin:
        word,val = line.strip().split('\t')
        if current_word == None:
            current_word = word
        if current_word != word:
            for count in count_pool:
                sum += count
                print "%s\t%s" % (current_word,sum)
                current_word = word
                count_pool = []
                sum = 0
        count_pool.append(int(val))
    for count in count_pool:
        sum += count
    print "%s\t%s" % (current_word, str(sum))

**run.sh**

$HADOOP_CMD="/usr/local/src/hadoop-1.2.1/bin/hadoop"
    //hadoop命令路径
    
    $STREAM_JAR_PATH="/usr/local/src/hadoop-1.2.1/contrib/streaming/hadoop-streaming-1.2.1.jar"
    //streamming路径
    INPUT_FILE_PATH_1="1.data"
    OUTPUT_PATH="/output"
    
    $HADOOP_CMD fs -rmr -skipTrash $OUTPUT_PATH
    
    #Step 1.
    $HADOOP_CMD jar $STREAM_JAR_PATH \
        -input $INPUT_FILE_PATH_1 \
        //指定作业的输入文件hdfs路径
        -output $OUTPUT_PATH \
        //指定作业输出文件hdfs路径
        -mapper "python map.py" \
        //自己的mapper程序
        -reducer "python red.py" \
        //自己的reducer程序
        -file ./map.py \
        -file ./red.py
        /*
            file
            打包文件到提交作业中
            (1)map和reduce的执行文件
            (2)map和reduce要用于输入的文件，如配置
            类似配置
            -cacheFile
            向计算节点分发hdfs文件
            -cacheArchive
            分发hdfs压缩文件
        */

常见的一些配置  
\-jobconf  
提交作业的一些配置属性  
常见配置

1.  mapred.map.tasks:map task数目
2.  mapred.reduce.tasks：reduce task数目
3.  stream.num.map.output.key.fields:指定map task输出记录中key所占域数目
4.  num.key.fields.for.partition指定对key分出来的前几部分做partiition而不是整个key
5.  mapred.job.name 作业名  
    mapred.job.priority 作业优先级  
    mapred.job.map.capacity 最多同时运行map任务数  
    mapred.job.reduce.capacity 最多同时运行reduce任务数  
    mapred.task.timeout 任务没有响应（输入输出）的最大时间  
    mapred.compress.map.output map的输出是否压缩  
    mapred.map.output.compression.codec map的输出压缩方式  
    mapred.output.compress reduce的输出是否压缩  
    mapred.output.compression.codec reduce的输出压缩方式  
    stream.map.output.field.separator map输出分隔符