mapreduce,整合数据字典表

发布时间:2020-07-25 03:28:10 作者:白话
来源:网络 阅读:1181

这个坑踩了好长。结果却是map方法中的context写错位置,导致错误。


源数据内容。就是想数据表中的第二列替换成字典表中的第二列。即字典表中的红色,换成字典表的蓝色。

//数据表data.txt

//one     1       two     qqq

//two     2       two     ccc


//字典表zidian.txt

//11sex

//22sex

//3未知0sex

//4结婚1marry

//5未婚2marry

//6未知0marry


想要的结果就是


附上代码:


import java.io.BufferedReader;

import java.io.FileReader;

import java.io.IOException;

import java.net.URI;

import java.net.URISyntaxException;

import java.util.HashMap;

import java.util.Iterator;

import java.util.Map;


import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.filecache.DistributedCache;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.mapreduce.Reducer;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class Cache {


public static class Mapall extends Mapper<Object, Text, Text, Text> {


private Map<String, String> sexMap = new HashMap<String, String>();

private Path[] localFiles;


// 先做分布式缓存处理,将数据换成到内存中

public void setup(Context context) throws IOException {

Configuration conf = context.getConfiguration();

localFiles = DistributedCache.getLocalCacheFiles(conf);

for(int i = 0;i<localFiles.length;i++) {

String a ;

BufferedReader br = new BufferedReader(new FileReader(localFiles[i].toString()));

while ((a = br.readLine()) != null && a.split("\t")[3].equals("sex")) {

//以数据作为key,文字作为value

sexMap.put(a.split("\t")[2], a.split("\t")[1]);

}

br.close();

}

}


@SuppressWarnings("unlikely-arg-type")

public void map(Object key, Text value, Context context) throws IOException, InterruptedException {


// 获取sex字段,是1,2这样的数据

String sex = value.toString().split("\t")[1];

// 如果key部分有1,2这种形式,就替换成男、女这样的内容

if (sexMap.keySet().equals(sex)) {

}

context.write(new Text(sexMap.get(sex)), new Text(""));

                        //就是这里,坑我好久的时间。

}

}


public static class Reduce extends Reducer<Text, Text, Text, Text> {

public void reduce(Text key, Iterator<Text> values, Context context) throws IOException, InterruptedException {

context.write(key, new Text(""));


}

}


public static void main(String[] args)

throws URISyntaxException, IOException, ClassNotFoundException, InterruptedException {

Configuration conf = new Configuration();

DistributedCache.addCacheFile(new URI("hdfs://192.168.20.39:8020/qpf/zidian.txt"), conf);


Job job = Job.getInstance(conf, "get cache file");

job.setJarByClass(Cache.class);


job.setMapperClass(Mapall.class);

job.setReducerClass(Reduce.class);


job.setOutputKeyClass(Text.class);

job.setOutputValueClass(Text.class);


FileInputFormat.addInputPath(job, new Path("hdfs://192.168.20.39:8020/qpf/data.txt"));

FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.20.39:8020/qpf/data_out"));


System.exit(job.waitForCompletion(true) ? 0 : 1);

}

}


很简单的一个数据替换的小例子。

推荐阅读:
  1. MapReduce on Hbase
  2. 数据字典

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

数据 mapreduce 字典表 分布式缓存 distribute

上一篇:Kubernetes 之 MySQL 持久存储和故障转移(十一)

下一篇:kafka为什么那个快?

相关阅读

您好,登录后才能下订单哦!

密码登录
登录注册
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》