首页 ›互联网技术资讯›内容详情

Hadoop用户怎么自定义

蜗牛互联网技术资讯2021-12-093110

这篇文章主要讲解了“Hadoop用户怎么自定义”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习“Hadoop用户怎么自定义”吧！

一：Hadoop内置的数据类型。

    Hadoop提供如下内置的数据类型，这些数据类型都实现了WritableComparable接口，以便用这些类型定义的数据可以被序列化进行网络传输和文件存储，以及进行大小比较。

BooleanWritable标准布尔型数值
ByteWritable单字节数值
DoubleWritable双字节数
FloatWritable浮点数
IntWritable整型数
LongWritable长整型数
Text使用UTF-8格式存储的文本
NullWritable当<key,value>中的key或value为空时使用
//简单知道这些类型
IntWritable iw = new IntWritable(1);
System.out.println(  iw.get() );  // 1 
	
BooleanWritable bw = new BooleanWritable(true);
System.out.println(  bw.get() );  // true

二：Hadoop-用户自定义的数据类型。

    自定义数据类型时，需满足两个基本要求，即

        1.实现Writable接口，以便该数据能被序列化后完成网络传输或文件输入/输出。

        2.如果该数据需要作为主键key使用，或需要比较数值大小时，则需要实现WritableComparable接口。

//Hadoop2.6.4版 - Writable源码：
public interface Writable {
 
  void write(DataOutput out) throws IOException;

  void readFields(DataInput in) throws IOException;

}
public interface WritableComparable<T> extends Writable, Comparable<T> {}

三：Hadoop内置的数据输入格式和RecordReader。

    数据输入格式（InputFormat）用于描述MapReduce作业的数据输入规范。MapReduce框架依靠数据输入格式完成输入规范检查、对数据文件进行输入分块（InputSplit），以及提供从输入分块中将数据记录逐一读出、并转换为Map过程的输入键值对等功能。

    Hadoop提供了丰富的内置数据输入格式，最常用的数据输入格式包括：TextInputFormat 和 KeyValueInputFormat。

    TextInputFormat是系统默认的数据输入格式，可以将文本文件分块并逐行读入以便Map节点进行处理。读入一行时，所产生的主键key就是当前行在整个文本文件中的字节偏移位置，而value就是该行的内容。

//TextInputFormat部分源码：
public class TextInputFormat extends FileInputFormat<LongWritable, Text> {

  @Override
  public RecordReader<LongWritable, Text> 
    createRecordReader(InputSplit split,
                       TaskAttemptContext context) {
    String delimiter = context.getConfiguration().get(
        "textinputformat.record.delimiter");
    byte[] recordDelimiterBytes = null;
    if (null != delimiter)
      recordDelimiterBytes = delimiter.getBytes(Charsets.UTF_8);
    return new LineRecordReader(recordDelimiterBytes);
  }

  //....
}

    KeyValueTextInputFormat是另一个常用的数据输入格式，可将一个按照<key,value>格式逐行存放的文本文件逐行读出，并自动解析生成相应的key和value。

//KeyValueTextInputFormat部分源码：
public class KeyValueTextInputFormat extends FileInputFormat<Text, Text> {

  // ...

  public RecordReader<Text, Text> createRecordReader(InputSplit genericSplit,
      TaskAttemptContext context) throws IOException {
    
    context.setStatus(genericSplit.toString());
    return new KeyValueLineRecordReader(context.getConfiguration());
  }

}

    RecordReader：对于一个数据输入格式，都需要有一个对应的RecordReader,主要用于将一个文件中的数据记录拆分成具体的键值对。TextInputFormat的默认RecordReader是LineRecordReader，而KeyValueTextInputFormat的默认RecordReader是KeyValueLineRecordReader。

四：Hadoop内置的数据输出格式与RecordWriter。

    数据输出格式（OutputFormat）用于描述MapReduce作业的数据输出规范。MapReduce框架依靠数据输出格式完成输出规范检查以及提供作业结果数据输出功能。

    同样，最常用的数据输出格式是TextOutputFormat,也是系统默认的数据输出格式，可以将计算结果以 “key + \t + vaue”的形式逐行输出到文本文件中。

    与数据输入格式类似样，数据输出格式也提供一个对应的RecordWriter,以便系统明确输出结果写入到文件中的具体格式。TextInputFormat的默认RecordWriter是LineRecordWriter,其实际操作是将结果数据以“key + \t + value”的形式输出到文本文件中。

//TextOutputFormat的部分源码：
public class TextOutputFormat<K, V> extends FileOutputFormat<K, V> {

  protected static class LineRecordWriter<K, V> extends RecordWriter<K, V> {
    // ...

    public LineRecordWriter(DataOutputStream out, String keyValueSeparator) {
      //...
    }

    public LineRecordWriter(DataOutputStream out) {
      this(out, "\t");
    }

    private void writeObject(Object o) throws IOException {
       // ...
    }

    public synchronized void write(K key, V value) throws IOException {
      //...
      out.write(newline);
    }

  }

  public RecordWriter<K, V> getRecordWriter(TaskAttemptContext job
                         ) throws IOException, InterruptedException {
     // ...
  }
}

五：通过打印输出UserInfo小例子来实现简单的用户自定义数据类型，数据输入格式，数据输出格式。 （简单的说就是模仿源码，基本上没多大变化）。

        以下附上案例源码：

1.定义自己的UserInfo，作为数据类型。

package com.hadoop.mapreduce.test4.outputformat;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableComparable;

public class UserInfo implements WritableComparable<UserInfo> {

	private int id;
	private String name;
	private int age;
	private String sex;
	private String address;
	
	public UserInfo() {
	}
	public UserInfo(int id, String name, int age, String sex, String address) {
		this.id = id;
		this.name = name;
		this.age = age;
		this.sex = sex;
		this.address = address;
	}

	// JavaBean 普通的get set方法....

	@Override
	public void readFields(DataInput in) throws IOException {
		this.id = in.readInt();
		this.name = in.readUTF();
		this.age = in.readInt();
		this.sex = in.readUTF();
		this.address = in.readUTF();
	}
	
	@Override
	public void write(DataOutput out) throws IOException {
		out.writeInt(id);
		out.writeUTF(name);
		out.writeInt(age);
		out.writeUTF(sex);
		out.writeUTF(address);
	}

	@Override
	public String toString() {
		return "Id:" + id + ", Name:" + name + ", Age:" + age + ", Sex:" + sex + ", Address:" + address ;
	}

	@Override
	public int compareTo(UserInfo userInfo) {
		return 0;
	}
}

2.定制自己的数据输入格式：UserInfoTextInputFormat。

package com.hadoop.mapreduce.test4.outputformat;
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

public class UserInfoTextInputFormat extends FileInputFormat<Text, UserInfo> {
	@Override
	public RecordReader<Text, UserInfo> createRecordReader(InputSplit split, TaskAttemptContext context)
			throws IOException, InterruptedException {
		context.setStatus(split.toString());
		UserInfoRecordReader userInforRecordReader = new UserInfoRecordReader(context.getConfiguration() );
		return userInforRecordReader;
	}
}

3.定制自己的RecordReader：UserInfoRecordReader。

package com.hadoop.mapreduce.test4.outputformat;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;

public class UserInfoRecordReader extends RecordReader<Text, UserInfo> {
	public static final String KEY_VALUE_SEPERATOR = 
			"mapreduce.input.keyvaluelinerecordreader.key.value.separator";

	private final LineRecordReader lineRecordReader;

	private byte separator = (byte) '\t';

	private Text innerValue;
	private Text key;

	private UserInfo value;


	public Class getKeyClass() { 
		return Text.class;
	}

	public UserInfoRecordReader(Configuration conf)throws IOException {
		lineRecordReader = new LineRecordReader();
		String sepStr = conf.get(KEY_VALUE_SEPERATOR,"\t");
		this.separator = (byte) sepStr.charAt(0);
	}

	public void initialize(InputSplit genericSplit,TaskAttemptContext context) throws IOException {
		lineRecordReader.initialize(genericSplit, context);
	}

	public static int findSeparator(byte[] utf, int start, int length, byte sep) {
		for (int i = start; i < (start + length); i++) {
			if (utf[i] == sep) {
				return i;
			}
		}
		return -1; //将这个截取标识符的位置给返回回去。
	}

	public static void setKeyValue(Text key, UserInfo value, byte[] line,int lineLen, int pos) {
		if (pos == -1) {
			key.set(line, 0, lineLen);
			value.setId(0);
			value.setName("");
			value.setAge(0);
			value.setSex("");
			value.setAddress("");
		} else {
			key.set(line, 0, pos); //设置键  从 第 0位置 到 截取标识符的位置
			Text text = new Text();
			text.set(line, pos + 1, lineLen - pos - 1);
			System.out.println("text的值： "+text);
			String[] str = text.toString().split(",");
			for (int i=0;i<str.length;i++) {
				//System.out.println("根据逗号分隔开来的值：  " + str[i] );
				String[] strKeyValue = str[i].split(":");
				//System.out.println("strKeyValue的Key-Value：" + key+"--->"+value);
				if("ID".equals(strKeyValue[0])){
					value.setId(Integer.parseInt( strKeyValue[1]) );
				}else if("Name".equals(strKeyValue[0])){
					value.setName( strKeyValue[1]);
				}else if("Age".equals(strKeyValue[0])){
					value.setAge(Integer.parseInt( strKeyValue[1]) );
				}else if("Sex".equals(strKeyValue[0])){
					value.setSex(strKeyValue[1] );
				}else if("Address".equals(strKeyValue[0])){
					value.setAddress(strKeyValue[1] );
				}
			}
//			System.out.println( "key--> " + key);
//			System.out.println( "value--> "+value +"\n\n");
		}
	}
	
	public synchronized boolean nextKeyValue()throws IOException {
		byte[] line = null;
		int lineLen = -1;
		if (key == null) {
			key = new Text();
		}
		if (value == null) {
			value = new UserInfo(); 
		}
		if (lineRecordReader.nextKeyValue()) {
			innerValue = lineRecordReader.getCurrentValue();
			line = innerValue.getBytes();
			lineLen = innerValue.getLength();
		} else {
			return false;
		}
		if (line == null){
			return false;
		}

		int pos = findSeparator(line, 0, lineLen, this.separator);
		setKeyValue(key, value, line, lineLen, pos);
		return true;
	}

	public Text getCurrentKey() {
		return key;
	}
	public UserInfo getCurrentValue() {
		return value;
	}

	public float getProgress() throws IOException {
		return lineRecordReader.getProgress();
	}

	public synchronized void close() throws IOException { 
		lineRecordReader.close();
	}

}

3.定制自己的输出格式：UserInfoTextOutputFormat。

package com.hadoop.mapreduce.test4.outputformat;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.ReflectionUtils;

public class UserInfoTextOutputFormat<K, V> extends FileOutputFormat<K, V> {
	public static String SEPERATOR = "mapreduce.output.textoutputformat.separator";
	protected static class LineRecordWriter<K, V> extends RecordWriter<K, V> {
		private static final String utf8 = "UTF-8";
		private static final byte[] newline;
		static {
			try {
				newline = "\n".getBytes(utf8);
				//System.out.println(  "newline --> " + newline);
			} catch (UnsupportedEncodingException uee) {
				throw new IllegalArgumentException("can't find " + utf8 + " encoding");
			}
		}

		protected DataOutputStream out;
		private final byte[] keyValueSeparator;

		public LineRecordWriter(DataOutputStream out, String keyValueSeparator) {
			this.out = out;
			try {
				this.keyValueSeparator = keyValueSeparator.getBytes(utf8);
			} catch (UnsupportedEncodingException uee) {
				throw new IllegalArgumentException("can't find " + utf8 + " encoding");
			}
		}

		public LineRecordWriter(DataOutputStream out) {
			this(out, "\t");
		}

		private void writeObject(Object o) throws IOException {
			if (o instanceof Text) {
				Text to = (Text) o;
				System.out.println(  "o instanceof Text  --> True : "+ to.toString()  );
				out.write(to.getBytes(), 0, to.getLength());
			} else {
				out.write(o.toString().getBytes(utf8));
				System.out.println( "o instanceof Text  --> false : "+ o.toString()  );
			}
		}

		public synchronized void write(K key, V value) throws IOException {
			boolean nullKey = key == null || key instanceof NullWritable;
			boolean nullValue = value == null || value instanceof NullWritable;
			System.out.println(  "nullKey--> "+nullKey +" ,  nullValue--> "+nullValue);
			if (nullKey && nullValue) {
				return;
			}
			System.out.println( " nullkey --> "+ nullKey + ", !nullkey -->"+nullKey);
			if (!nullKey) {
				writeObject(key);
			}
			System.out.println( "(nullKey || nullValue) --> " + (nullKey || nullValue) );
			if (!(nullKey || nullValue)) {
				out.write(keyValueSeparator);
			}
			if (!nullValue) {
				writeObject(value);
			}
			out.write(newline);
		}

		public synchronized void close(TaskAttemptContext context) throws IOException {
			out.close();
		}
	}

	public RecordWriter<K, V> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
		Configuration conf = job.getConfiguration();
		boolean isCompressed = getCompressOutput(job);
		String keyValueSeparator= conf.get(SEPERATOR, "---->");
		System.out.println(  "keyValueSeparator---> "+keyValueSeparator);
		CompressionCodec codec = null;
		String extension = "";
		if (isCompressed) {
			Class<? extends CompressionCodec> codecClass = 
					getOutputCompressorClass(job, GzipCodec.class);
			codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, conf);
			extension = codec.getDefaultExtension();
		}
		Path file = getDefaultWorkFile(job, extension);
		System.out.println(  "file --> Path : "+ file  );
		FileSystem fs = file.getFileSystem(conf);
		
		if (!isCompressed) {
			FSDataOutputStream fileOut = fs.create(file, false);
			System.out.println( "if---isCompressed-->: "+fileOut);
			return new LineRecordWriter<K, V>(fileOut, keyValueSeparator);
		} else {
			FSDataOutputStream fileOut = fs.create(file, false);
			System.out.println( "else---isCompressed-->: "+fileOut);
			return new LineRecordWriter<K, V>(new DataOutputStream(codec.createOutputStream(fileOut)),keyValueSeparator);
		}
	}
}

5.测试类：PrintUserInfo

package com.hadoop.mapreduce.test4.outputformat;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class PrintUserInfo {
	public static final IntWritable ONE = new IntWritable(1);
	public static class UserInfoMapper extends Mapper<Text, UserInfo, Text, UserInfo>{
		@Override
		protected void map(Text key, UserInfo value, Mapper<Text, UserInfo, Text, UserInfo>.Context context)
				throws IOException, InterruptedException {
			super.map(key, value, context);
		}
	}
	
	public static void main(String[] args) {
		try {
			Configuration conf = new Configuration();
			Job job = Job.getInstance(conf, "UserInfo");
			
			job.setJarByClass(PrintUserInfo.class);
			job.setMapperClass(UserInfoMapper.class);
			
			//定制输入格式：
			job.setInputFormatClass(UserInfoTextInputFormat.class);
			//定制输出格式：
			job.setOutputFormatClass(UserInfoTextOutputFormat.class);
			
			job.setMapOutputKeyClass(Text.class);
			//用的自己定义的数据类型
			job.setMapOutputValueClass(UserInfo.class);
			
			FileInputFormat.addInputPath(job, new Path("hdfs://192.168.226.129:9000/rootdir/mapuserinfo.txt"));
			FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.226.129:9000/rootdir/data/output7/"+System.currentTimeMillis()+"/"));
			System.exit(job.waitForCompletion(true) ? 0 : 1);//执行job
			
		} catch (IOException e) {
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}

6.输出结果：

1.数据文件：

1	ID:221,Name:xj,Age:22,Sex:man,Address:hunan,
2	ID:222,Name:cc,Age:21,Sex:Woman,Address:miluo,

2.结果文件：

1---->Id:221, Name:xj, Age:22, Sex:man, Address:hunan
2---->Id:222, Name:cc, Age:21, Sex:Woman, Address:miluo

感谢各位的阅读，以上就是“Hadoop用户怎么自定义”的内容了，经过本文的学习后，相信大家对Hadoop用户怎么自定义这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是蜗牛博客，小编将为大家推送更多相关知识点的文章，欢迎关注！

免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：niceseo99@gmail.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
#hadoop#v2ray高速节点分享
本文地址：http://www.vps90.com/vps/5041.html
版权声明：如无特殊标注，文章均为本站原创，转载时请以链接形式注明文章出处。

相关文章

评论

关于蜗牛120vps博客

蜗牛120博客

感谢大家的支持，欢迎交流讨论！
『蜗牛120博客』

  • 11080011阅读数
  • 118607评论数

最近发表

  • Python集合是什么及怎么创建
  • 日本服务器租用 站群服务器 onevps：新增（支付宝+中文网站）
  • Go反向代理ReverseProxy解析的方法是什么
  • 日本服务器租用 站群服务器 kvmla：日本/香港/新加坡，VPS直接8折优惠（+充值多送20%），独立服务器一律7.5折优惠（首月反300元）
  • NLTK安装错误怎么解决
  • SpringBoot表单提交全局日期格式转换器如何实现
  • openSUSE如何处理硬件驱动程序
  • Vue中代码传送怎么实现
  • 怎样才能挑选到适合的vps服务器
  • golang中怎么保证精度

    • 标签列表

    有免费节点资源，我们会通知你！加入纸飞机订阅群

    ×
    天气预报查看日历分享网页手机扫码留言评论电报频道链接