文档章节

基于大数据开发套件定时调度带资源文件的MapReduce作业

_夜枫
 _夜枫
发布于 2017/03/29 22:18
字数 1201
阅读 23
收藏 0

MaxCompute里的MR作业,很少是只要跑一次就好了的。如果需要周期性调度,目前MaxCompute(原名ODPS)只提供了计算引擎,任务调度可以使用大数据开发套件来实现。这篇帖子从基础开始,介绍了3种周期性调度的方法。同时还介绍了如何使用资源文件。

代码开发

代码以文档里的WordCount 作为例子。
在这个基础上,增加资源文件的读取方法,修改Reduce类。主要的逻辑是读取资源文件,资源文件里的数据格式是字符串1,字符串2。代码逻辑是如果word count里的word如果有在字符串1里出现的话,就替换成字符串2。

    public static class SumReducer extends ReducerBase {
        private Record result = null;
        private Map<String,String> maps = null;

        @Override
        public void setup(TaskContext context) throws IOException {
            result = context.createOutputRecord();
            maps = new HashMap<String,String>();
            StringBuilder importdata = new StringBuilder();
            BufferedInputStream bufferedInput = null;
            try {
                byte[] buffer = new byte[1024];
                int bytesRead = 0;
                //读取资源文件的内容
                bufferedInput = context.readResourceFileAsStream("resource.txt");

                while ((bytesRead = bufferedInput.read(buffer)) != -1) {
                    String chunk = new String(buffer, 0, bytesRead);
                    importdata.append(chunk);
                }
                //解析资源文件的内容,把替换前,替换后的数据放到map里
                String lines[] = importdata.toString().split("\n");
                for (int i = 0; i < lines.length; i++) {
                    String[] ss = lines[i].split(",");
                    maps.put(ss[0].trim(), ss[1].trim());
                    System.out.println(ss[0]+"->"+ss[1]);
                }
            } catch (FileNotFoundException ex) {
                throw new IOException(ex);
            } catch (IOException ex) {
                throw new IOException(ex);
            } finally {
            }
        }

        @Override
        public void reduce(Record key, Iterator<Record> values,
                TaskContext context) throws IOException {
            long count = 0;
            while (values.hasNext()) {
                Record val = values.next();
                count += (Long) val.get(0);
            }
            String value = key.get(0).toString();
            if(maps.containsKey(value)){
                System.out.println(value+"->"+maps.get(value));
                value = maps.get(value);

            }
            result.set(0, value);
            result.set(1, count);
            context.write(result);
        }
    }

具体资源文件的用法可以参考文档 ,这里就不再多解释了。

客户端调用

对于测试数据,源文件的内容为

odps,MaxCompute
hello,Hello

我们先用手工调度来跑这个MR,这里跑通了后后面的所有的配置就很容易明白了。
首先需要把代码打出的jar包,和这个resource.txt文件上传到服务器上

>add jar D:\cx_word_count.jar -f;
OK: Resource 'cx_word_count.jar' have been updated.
>add file D:\resource.txt -f;
OK: Resource 'resource.txt' have been updated.

然后通过命令行来调用

jar -resources cx_word_count.jar,resource.txt -classpath D:\cx_word_count.jar com.aliyun.odps.mr.WordCount;

这里的-resources引用的是跑在服务器上的,-classpath是用来找到main方法的。理解这个对后面配置同步任务很有帮助。可以参阅文档

Crontab调用

odpscmd客户端有一个参数,是-e,可以在shell里直接调用jar命令来跑MR,当然也可以使用odpscmd -f来再调用一个脚本文件,但是这样有点麻烦了。这里就直接用-e来做。

你可以先用

/odps/cmd/bin/odpscmd  -e "jar -resources cx_word_count.jar,resource.txt -classpath /odps/cx_word_count.jar com.aliyun.odps.mr.WordCount;"

在Linux服务器上运行任务。注意安装odpscmd配置前需要先配置好java环境。然后后面的Crontab的配置就不展开了。

MR作业

配置DataIDE的MR作业的界面,很容易就让人想到MR任务的main方法。其实就是DataIDE会根据配置自己生成main方法,然后去调用MaxCompute上的任务。具体的配置可以参考这个截图:
screenshot
可以在右边看到可以配置任务的调度周期和上下游依赖,从而实现每天的定时调度,而且还能是保证上游的数据导入、预处理完成后才开始做MR操作,非常好用。

Shell任务

上述的MR任务简单方便,但是DataIDE出于安全考虑,不让用户自己写main方法。如果需要用到诸如传参数之类的功能,可以自己写Shell任务,但是调度让DataIDE来做。这样就集上面两个方法之长了。

Shell任务需要先参考文档 先配置调度的ECS信息,这里不再展开。完成后写一个Shell脚本,内容为

##@resource_reference{"cx_word_count.jar,resource.txt"}
/opt/taobao/tbdpapp/odpswrapper/odpsconsole/bin/odpscmd -u testid -p testkey --project=testproject --endpoint=http://service.odps.aliyun.com/api  -e "jar -resources cx_word_count.jar,resource.txt -classpath /odps/cx_word_count.jar com.aliyun.odps.mr.WordCount"

要把里面的Access id/key,Project 替换成你自己的,然后开始测试代码。需要特别注意的是,** Shell任务是在机器上的admin账号下运行的** ,如果发现各种奇怪的错误,比如明明存在的文件找不到一类的错误,可以先su - admin,调试下Shell命令,或者访问下对应的文件,看看是否是环境变量,文件目录权限的问题。另外也可以把错误日志重定向到某个文件里,比如/tmp文件夹下的某个临时日志文件里,方便事后调试。大家可以在admin账号下把shell调试通过后再放到数加上去调用。

另外Shell任务可以调整调度的机器,可以参考
screenshot

本文转载自:https://yq.aliyun.com/articles/66634

_夜枫
粉丝 10
博文 506
码字总数 0
作品 0
朝阳
后端工程师
私信 提问
【大数据】MaxCompute概述

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/binggetong/article/details/82769188 什么是MaxCompute 大数据计算服务(MaxCompute,原名ODPS)是一种快速、...

Elsa晓冰
2018/09/19
0
0
大数据(hadoop-生态系统概述以及版本演化)

Hadoop生态系统的特点: 源代码开源(免费) 社区活跃、参与者众多 涉及分布式存储和计算的方方面面 已得到企业界验证 Hadoop1.0和2.0: Hadoop概述: 分布式存储系统HDFS(Hadoop Distribu...

这很耳东先生
04/10
17
0
大数据开发 | MapReduce介绍

1. MapReduce 介绍 1.1MapReduce的作用 假设有一个计算文件中单词个数的需求,文件比较多也比较大,在单击运行的时候机器的内存受限,磁盘受限,运算能力受限,而一旦将单机版程序扩展到集群...

嘿你好夏天
2018/04/18
0
0
大数据(hadoop-Mapreduce原理架构)

课程目标: 1:MapReduce的应用场景 2:MapReduce编程模型 3:MapReduce的架构 4:常见MapReduce应用场景 5:总结 MapReduce的定义 源自于Google的MapReduce论文 发表于2004年12月 Hadoop M...

这很耳东先生
04/30
27
0
大数据Hadoop之 YARN认识

大数据Hadoop之 YARN认识2 Yarn是一个分布式的资源管理系统,用以提高分布式的集群环境下的资源利用率,这些资源包括内存、IO、网络、磁盘等。其产生的原因是为了解决原MapReduce框架的不足。...

飓风2000
2018/09/17
19
0

没有更多内容

加载失败,请刷新页面

加载更多

rime设置为默认简体

转载 https://github.com/ModerRAS/ModerRAS.github.io/blob/master/_posts/2018-11-07-rime%E8%AE%BE%E7%BD%AE%E4%B8%BA%E9%BB%98%E8%AE%A4%E7%AE%80%E4%BD%93.md 写在开始 我的Arch Linux上......

zhenruyan
今天
4
0
简述TCP的流量控制与拥塞控制

1. TCP流量控制 流量控制就是让发送方的发送速率不要太快,要让接收方来的及接收。 原理是通过确认报文中窗口字段来控制发送方的发送速率,发送方的发送窗口大小不能超过接收方给出窗口大小。...

鏡花水月
今天
8
0
OSChina 周日乱弹 —— 别问,问就是没空

Osc乱弹歌单(2019)请戳(这里) 【今日歌曲】 @tom_tdhzz :#今日歌曲推荐# 分享容祖儿/彭羚的单曲《心淡》: 《心淡》- 容祖儿/彭羚 手机党少年们想听歌,请使劲儿戳(这里) @wqp0010 :周...

小小编辑
今天
868
10
golang微服务框架go-micro 入门笔记2.1 micro工具之micro api

micro api micro 功能非常强大,本文将详细阐述micro api 命令行的功能 重要的事情说3次 本文全部代码https://idea.techidea8.com/open/idea.shtml?id=6 本文全部代码https://idea.techidea8....

非正式解决方案
今天
5
0
Spring Context 你真的懂了吗

今天介绍一下大家常见的一个单词 context 应该怎么去理解,正确的理解它有助于我们学习 spring 以及计算机系统中的其他知识。 1. context 是什么 我们经常在编程中见到 context 这个单词,当...

Java知其所以然
昨天
6
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部