如何跨平台在本地开发环境提交MapReduce作业到CDH集群

温馨提示:要看高清无码套图,请使用手机打开并单击图片放大查看。

1.文档编写目的


在开发Hadoop的MapReduce作业时需要重复的打包并手动传输到集群运行往往比较麻烦,有时我们也需要在本地能够直接调试代码如在Intellij能直接连接到集群提交作业,或者我们需要跨平台的提交MapReduce作业到集群。那么如何实现呢?本篇文章主要讲述如何跨平台在本地开发环境下提交作业到Hadoop集群,这里我们还是分为Kerberos环境和非Kerberos环境。

  • 内容概述

1.环境准备

2.非Kerberos及Kerberos环境连接示例

  • 测试环境

1.Kerberos集群CDH5.11.2,OS为Redhat7.2

2.非Kerberos集群CDH5.13,OS为CentOS6.5

3.Windows + Intellij

  • 前置条件

1.CDH集群运行正常

2.本地开发环境与集群网络互通且端口放通

2.环境准备


1.Maven依赖

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>2.6.0-cdh5.11.2</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>2.6.0-cdh5.11.2</version>
</dependency>

Maven依赖需要注意Fayson在《如何使用Java代码访问CDH的Solr服务》提到使用cdh的Maven库

2.创建访问集群的Keytab文件(非Kerberos集群可跳过此步)

[ec2-user@ip-172-31-22-86 keytab]$ sudo kadmin.local
Authenticating as principal mapred/admin@CLOUDERA.COM with password.
kadmin.local:  listprincs fayson*
fayson@CLOUDERA.COM
kadmin.local:  xst -norandkey -k fayson.keytab fayson@CLOUDERA.COM
...
kadmin.local:  exit
[ec2-user@ip-172-31-22-86 keytab]$ ll
total 4
-rw------- 1 root root 514 Nov 28 10:54 fayson.keytab
[ec2-user@ip-172-31-22-86 keytab]$ 

3.获取集群krb5.conf文件,内容如下(非Kerberos集群可跳过此步)

includedir /etc/krb5.conf.d/

[logging]
 default = FILE:/var/log/krb5libs.log
 kdc = FILE:/var/log/krb5kdc.log
 admin_server = FILE:/var/log/kadmind.log

[libdefaults]
 dns_lookup_realm = false
 ticket_lifetime = 24h
 renew_lifetime = 7d
 forwardable = true
 rdns = false
 default_realm = CLOUDERA.COM
 #default_ccache_name = KEYRING:persistent:%{uid}

[realms]
 CLOUDERA.COM = {
  kdc = ip-172-31-22-86.ap-southeast-1.compute.internal
  admin_server = ip-172-31-22-86.ap-southeast-1.compute.internal
 }

4.配置hosts文件

172.31.22.86 ip-172-31-22-86.ap-southeast-1.compute.internal
172.31.26.102 ip-172-31-26-102.ap-southeast-1.compute.internal
172.31.21.45 ip-172-31-21-45.ap-southeast-1.compute.internal
172.31.26.80 ip-172-31-26-80.ap-southeast-1.compute.internal

5.通过Cloudera Manager下载Yarn客户端配置

6.工程目录结构

以下用WordCount例子来说明。

3.Kerberos和非Kerberos的公共类


WordCountMapper类

public class WordCountMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
    @Override
    protected void map(LongWritable key, Text value,Context context)
            throws IOException, InterruptedException {
        //获取到一行文件的内容
        String line = value.toString();
        //切分这一行的内容为一个单词数组
        String[] words = StringUtils.split(line, " ");
        //遍历  输出  <word,1>
        for(String word:words){
            context.write(new Text(word), new LongWritable(1));
        }
    }
}

WordCountReducer类

public class WordCountReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
    @Override
    protected void reduce(Text key, Iterable<LongWritable> values,Context context)
            throws IOException, InterruptedException {
        long count = 0;
        for(LongWritable value:values){
            //调用value的get()方法将long值取出来
            count += value.get();
        }
        //输出<单词:count>键值对
        context.write(key, new LongWritable(count));
    }
}

InitMapReduceJob类

public class InitMapReduceJob {

    public static Job initWordCountJob(Configuration conf) {
        Job wcjob = null;
        try {
            conf.setBoolean("mapreduce.app-submission.cross-platform", true);  //设置跨平台提交作业
            //设置job所使用的jar包,使用Configuration对象调用set()方法,设置mapreduce.job.jar wcount.jar
            conf.set("mapred.jar", "C:\\Users\\Administrator\\IdeaProjects\\hbasedevelop\\target\\hbase-develop-1.0-SNAPSHOT.jar");
            //创建job对象需要conf对象,conf对象包含的信息是:所用的jar包
            wcjob = Job.getInstance(conf);
            wcjob.setMapperClass(WordCountMapper.class);
            wcjob.setReducerClass(WordCountReducer.class);

            //wcjob的mapper类输出的kv数据类型
            wcjob.setMapOutputKeyClass(Text.class);
            wcjob.setMapOutputValueClass(LongWritable.class);

            //wcjob的reducer类输出的kv数据类型
            //job对象调用setOutputKey
            wcjob.setOutputKeyClass(Text.class);
            wcjob.setOutputValueClass(LongWritable.class);
            FileInputFormat.setInputPaths(wcjob, "/fayson");
            FileOutputFormat.setOutputPath(wcjob, new Path("/wc/output"));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return wcjob;
    }
}

注意:代码中黄底标识部分,如果未设置会导致作业会运行失败。

ConfigurationUtil类

public class ConfigurationUtil {
    /**
     * 获取Hadoop配置信息
     * @param confPath
     * @return
     */
    public static Configuration getConfiguration(String confPath) {
        Configuration configuration = new YarnConfiguration();
        configuration.addResource(new Path(confPath + File.separator + "core-site.xml"));
        configuration.addResource(new Path(confPath + File.separator + "hdfs-site.xml"));
        configuration.addResource(new Path(confPath + File.separator + "mapred-site.xml"));
        configuration.addResource(new Path(confPath + File.separator + "yarn-site.xml"));
        configuration.setBoolean("dfs.support.append", true);
        configuration.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        configuration.setBoolean("fs.hdfs.impl.disable.cache", true);
        return configuration;
    }
}

4.非Kerberos环境


1.Intellij运行示例代码

public class NodeKBMRTest {

    private static String confPath = System.getProperty("user.dir") + File.separator + "nonekb-conf";

    public static void main(String[] args) {
        try {
            Configuration conf = ConfigurationUtil.getConfiguration(confPath);
            Job wcjob = InitMapReduceJob.initWordCountJob(conf);
            wcjob.setJarByClass(NodeKBMRTest.class);
            wcjob.setJobName("NodeKBMRTest");

            //调用job对象的waitForCompletion()方法,提交作业。
            boolean res = wcjob.waitForCompletion(true);
            System.exit(res ? 0 : 1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

2.直接在Intellij运行提交MR作业到Hadoop集群

运行成功

3.查看HDFS输出结果

5.Kerberos环境


1.Intellij运行示例代码

public class KBMRTest {

    private static String confPath = System.getProperty("user.dir") + File.separator + "conf";

    public static void main(String[] args) {
        try {
            System.setProperty("java.security.krb5.conf", "/Volumes/Transcend/keytab/krb5.conf");
            System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");
            System.setProperty("sun.security.krb5.debug", "true"); //Kerberos Debug模式
            Configuration conf = ConfigurationUtil.getConfiguration(confPath);
            //登录Kerberos账号
            UserGroupInformation.setConfiguration(conf);
            UserGroupInformation.loginUserFromKeytab("fayson@CLOUDERA.COM", "/Volumes/Transcend/keytab/fayson.keytab");
            UserGroupInformation userGroupInformation = UserGroupInformation.getCurrentUser();

            Job wcjob = InitMapReduceJob.initWordCountJob(conf);
            wcjob.setJarByClass(KBMRTest.class);
            wcjob.setJobName("KBMRTest");

            //调用job对象的waitForCompletion()方法,提交作业。
            boolean res = wcjob.waitForCompletion(true);
            System.exit(res ? 0 : 1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

2.直接在Intellij运行代码,代码自动推送jar到集群执行

Yarn作业界面

3.查看HDFS创建的目录及文件

注意:在提交作业时,如果代码修改需要重新编译打包,并将jar放到黄底标注的目录。

GitHub源码地址:

https://github.com/javaxsky/cdhproject

为天地立心,为生民立命,为往圣继绝学,为万世开太平。

温馨提示:要看高清无码套图,请使用手机打开并单击图片放大查看

原文发布于微信公众号 - Hadoop实操(gh_c4c535955d0f)

原文发表时间:2017-12-06

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

编辑于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Fish

从零爬着学spark

为啥叫爬着学 在此之前,我连spark是什么都不清楚,在看完《Spark快速大数据分析》(Learning Spark:Lighting-fast Data A...

2057
来自专栏Java编程技术

MyBatis中使用流式查询避免数据量过大导致OOM

其中fetchSize="-2147483648",Integer.MIN_VALUE=-2147483648

6151
来自专栏Ryan Miao

Java Web基础入门

前言 语言都是相通的,只要搞清楚概念后就可以编写代码了。而概念是需要学习成本的。 Java基础 不用看《编程思想》,基础语法看 http://www.runo...

4057
来自专栏闵开慧

大数据面试题整理

JAVA相关 1-1)List 与set 的区别? 老掉牙的问题了,还在这里老生常谈:List特点:元素有放入顺序,元素可重复 ,Set特点:元素无放入顺序,...

4.6K14
来自专栏王小雷

Hive基础(1)

Hive基础(1) Hive的HQL(2) 1. Hive并不是分布式的,它独立于机器之外,类似于Hadoop的客户端。 2. 元数据和数据的区别,前者如表名、...

2557
来自专栏岑玉海

Spark源码系列(七)Spark on yarn具体实现

本来不打算写的了,但是真的是闲来无事,整天看美剧也没啥意思。这一章打算讲一下Spark on yarn的实现,1.0.0里面已经是一个stable的版本了,可是...

5755
来自专栏抠抠空间

Flask之基本使用与配置

Flask是一个基于Python开发并且依赖jinja2模板和Werkzeug WSGI服务的一个微型框架,对于Werkzeug本质是Socket服务端,其用于...

982
来自专栏技术小讲堂

Entity Framework4.3 Code-First基于代码的数据迁移讲解1.建立一个最初的模型和数据库   2.启动Migration(数据迁移)3.第一个数据迁移4.订制的数据迁移4.动态

前段时间一直在研究Entity Framework4,但是苦于没有找到我特别中意的教程,要么就是千篇一律的文章,而且写的特别简单,可以说,糟践了微软这么牛埃克斯...

3478
来自专栏Albert陈凯

用MapReduce分析Hbase将结果插入mysql中

从HBASE读取清洗过的数据,写入到mysql的表中 NewInstallUserRunner.java 计算新增用户入口类 ? NewInstallUser...

4208
来自专栏Python中文社区

Python多进程并行编程实践-mpi4py的使用

專 欄 ❈PytLab,Python 中文社区专栏作者。主要从事科学计算与高性能计算领域的应用,主要语言为Python,C,C++。熟悉数值算法(最优化方法,...

8227

扫码关注云+社区

领取腾讯云代金券