使用kettle执行mapreduce

作者: 哥别打脸 | 来源:发表于2018-03-09 10:53 被阅读937次

使用kettle执行mapreduce
尚硅谷大数据技术之Oozie
kettle处理操作
hadoop streaming
Hive on Spark参数调优姿势小结
Hadoop Streaming的使用
MapReduce工作机制和序列化
hadoop 找不到或无法加载主类org.apache.had
Spark学习总结
MapReduce

使用kettle执行mapreduce#

机器：192.168.9.157 10G内存，4核CPU，centos6.5

hadoop版本：2.7.3

pdi:8.0

目的

使用pdi工具实现运行mapreduce的wordcount程序，不需要写任何java代码从而简化开发人员的工作量。

参考文档

http://www.pentaho.com/sites/default/files/webdetails-pentaho-evalpath/pdf/evaluation-experiences-pdf/EL7_PentahoMapReduce.pdf

https://wiki.pentaho.com/display/BAD/Using+Pentaho+MapReduce+to+Parse+Weblog+Data

https://wiki.pentaho.com/display/BAD/Using+Pentaho+MapReduce+to+Generate+an+Aggregate+Dataset/

准备工作

选择对应的hadoop发行版

image

hadoop cluster集群配置

image

这里jobtracker实际上应该叫resourcemanger，在hadoop2.x的yarn中已经没有jobtracker概念了。
这个值与yarn-site.xml中yarn.resourcemanager.address对应。

修改config.properties

修改kettle\pdi-ce-8.0.0.0-28\data-integration\plugins\pentaho-big-data-plugin\hadoop-configurations\hdp26\config.properties增加一行

authentication.superuser.provider=NO_AUTH

复制集群上的mapred-site.xml

复制集群上mapred-site.xml到\kettle\pdi-ce-8.0.0.0-28\data-integration\plugins\pentaho-big-data-plugin\hadoop-configurations\cdh512。如果在windows上提交任务需要增加mapreduce.app-submission.cross-platform配置。mapreduce.reduce.memory.mb，mapreduce.map.memory.mb 这两个值设置了每个container所需要的内存资源，java.opts一般设置memory.mb*0.8.

<configuration>
    <property>
         <name>mapreduce.framework.name</name>
         <value>yarn</value>
    </property>

    <property> 
        <name>mapreduce.app-submission.cross-platform</name>
        <value>true</value>
    </property>
<property> 
    <name>mapreduce.reduce.memory.mb</name>
    <value>1024</value>
</property>


<property> 
        <name>mapreduce.map.memory.mb</name>
        <value>1024</value>
</property>

<property> 
    <name>mapreduce.map.java.opts</name>
    <value>-Xmx800m</value>
 </property>
 
<property> 
    <name>mapreduce.reduce.java.opts</name>
    <value>-Xmx800m</value>
    </property>
</configuration>

复制集群上的yarn-site.xml

复制集群上yarn-site.xml到\kettle\pdi-ce-8.0.0.0-28\data-integration\plugins\pentaho-big-data-plugin\hadoop-configurations\cdh512。
修改hostname为对应主机的地址，或者在本机配置host。
一定要配置yarn.nodemanager.resource.memory-mb，尽量配置大一点，这里配置了8G，因为nodemanager启动一个container就需约1.5G内存，如果给nodemanager的内存不够大，将会导致无法启动container从而无法运行mapreduce。

<configuration>
    <property>
         <name>yarn.nodemanager.aux-services</name>
         <value>mapreduce_shuffle</value>
     </property>
     
     <property>
        <name>yarn.resourcemanager.hostname</name>
        <value>192.168.9.157</value>
    </property>
     
     <property>
        <name>yarn.nodemanager.hostname</name>
        <value>192.168.9.157</value>
    </property> 
        
        
     <property>
        <name>yarn.resourcemanager.address</name>
        <value>192.168.9.157:8032</value>
    </property>
    
     <property>
        <name>yarn.nodemanager.resource.memory-mb</name>
        <value>8192</value>
    </property>
    
    
    <property>
           <name>yarn.scheduler.minimum-allocation-mb</name>
           <value>256</value>
    </property>

     <property>
       <name>yarn.nodemanager.vmem-pmem-ratio</name>
        <value>6</value>
      </property>
    ......
</configuration>

一、创建mapper的Transformation

image

input

image

其中split field row，
delimiter 填写一个空格，用于如何分割一个单词。

image

add constants ，增加一列count统计单词出现的次数，map节点还没有统计，所以所有的单词次数都是1。

image

output

image

这一系列的过程相当于实现了wordcount的mapreduce的map方法，也就是

public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
    StringTokenizer itr = new StringTokenizer(value.toString());
    while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
    }
}

主要的目的就是将hdfs上的文件map操作成如下结果

word（key） count（value）

hello 1

world 1

hello 1

二、创建reduce的Transformation

[站外图片上传中...(image-83078a-1520563970869)]

input，这里的key，value其实就是来自于map阶段的word，count。

[站外图片上传中...(image-c4be5a-1520563970869)]

group by，
相当于执行了 select key，sum（value） as sum from table group by key；

image

out

image

这一系列的操作相当于实现了wordcount的mapreduce的reduce方法，也就是

public void reduce(Text key, Iterable<IntWritable> values, Context context)
        throws IOException, InterruptedException {
    int sum = 0;
    for (IntWritable val : values) {
        sum += val.get();
    }
    result.set(sum);
    context.write(key, result);
}

也就是将map阶段的数据reduce成如下结果。