设为首页 加入收藏

TOP

使用mapreduce读取es中数据到hdfs
2018-12-04 08:16:06 】 浏览:105
Tags:使用 mapreduce 读取 数据 hdfs
版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/hejiale926/article/details/54022305

前段时间的一个例子使用hive建表做数据仓库,源数据位于elasticsearch中,需要定时拉取数据到hdfs
本打算直接使用hive-elasticsearch建表,但是这样相当于同步表,elasticsearch中数据删除hive中也会被删除,不妥。
最终决定使用暴力方式,mapreduce直接读取到hdfs中,然后数据处理后load加载到hive
具体代码如下:

/**   
* @Title: EsToHdfs.java 
* @Package com.midea.main 
* @Description: TODO(用一句话描述该文件做什么) 
* @author 乐   
* @date 2016年12月8日 下午1:43:50 
* @version V1.0   
*/
package com.lele.main;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.elasticsearch.hadoop.cfg.ConfigurationOptions;
import org.elasticsearch.hadoop.mr.EsInputFormat;

/**
 * @ClassName: EsToHdfs
 * @Description: TODO(将es的数据传输到hdfs)
 * @author 乐
 * @date 2016年12月8日 下午1:43:50
 */
public class EsToHdfs {

    /**
     * @Title: main
     * @Description: TODO(mapreduce主方法入口)
     * @author 乐
     * @param @param args
     * @return void
     * @throws Exception
     * @throws @date 2016年12月8日 下午1:43:50
     */

    public static void main(String[] args) throws Exception {

        //生成时间,此项目中一天拉取一回,各位略过此段
        Date date = new Date();
        SimpleDateFormat df = new SimpleDateFormat("yyyy.MM.dd");
        String de = df.format(date.getTime() - 60 * 60 * Integer.parseInt(args[0]) * 1000);

        //mapreduce配置,相信大家都不陌生吧
        Configuration conf = new Configuration();
        conf.set(ConfigurationOptions.ES_NODES, "10.47.*.*");//es地址
        conf.set(ConfigurationOptions.ES_PORT, "9200");//es端口
        //此三项为作者在测试过程中遇到es连接失败情况下配置的,经查实为机器网络原因,和代码无关
        // conf.set(ConfigurationOptions.ES_NODES_CLIENT_ONLY, "true");
        // conf.set(ConfigurationOptions.ES_NODES_DATA_ONLY, "false");
        // conf.set(ConfigurationOptions.ES_INDEX_AUTO_CREATE, "yes");
        //设置输出格式为json格式,如不设置此项,则默认为一个看不懂,解析不了的类似json的格式
        conf.set(ConfigurationOptions.ES_OUTPUT_JSON, "true");
        //设置es索引位置和type位置
        conf.set(ConfigurationOptions.ES_RESOURCE, "logstash-proserver-" + de + "/" + args[1]);

        Job job = Job.getInstance(conf, EsToHdfs.class.getSimpleName());
        job.setJobName(de + "_" + args[1]);

        job.setJarByClass(EsToHdfs.class);
        job.setSpeculativeExecution(false);

        job.setInputFormatClass(EsInputFormat.class);
        job.setMapOutputKeyClass(NullWritable.class);
        job.setMapOutputValueClass(Text.class);

        job.setMapperClass(MapperES.class);

        FileInputFormat.addInputPath(job, new Path("/es_input"));//固定为/es_input
        FileOutputFormat.setOutputPath(job, new Path("/es_output/" + de + "/" + args[1]));

        job.setNumReduceTasks(0);
        job.waitForCompletion(true);

    }

    public static class MapperES extends Mapper<Object, Object, NullWritable, Text> {
        @Override
        public void map(Object key, Object value, Context context) throws IOException, InterruptedException {
            Text doc = (Text) value;
            context.write(NullWritable.get(), doc);
        }
      }
    }

网上很多文档在map阶段输出使用的是MapWritable.class
但是作者测试不成功,最后简单粗暴直接改为Text.class测试通过,稳定运行。作者没有查找原因,有知道原因的博友可以分享一下。
elasticsearch-mapreduce官方文档地址:
https://www.elastic.co/guide/en/elasticsearch/hadoop/master/mapreduce.html
很详细,上述代码可以保证正常稳定运行,如需要更多配置,可以参阅官方文档

】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇hdfs的9000和8020端口 下一篇HDFS再学习:HA和Federation机制

最新文章

热门文章

Hot 文章

Python

C 语言

C++基础

大数据基础

linux编程基础

C/C++面试题目