把已抓取好的網路輿情資訊（以txt形式存放），儲存到HBase中，再進行資訊分析。

要求：

以檔名作Key，整個檔案內容作Value

思路：

txt檔案先上傳到HDFS中，再使用HBase MapReduce將檔案寫入HBase中。（很簡單的思路）

問題分析：

首先必須分析到的問題是，如何讀取解析txt檔案，TextInputFormat是預設的檔案解析類，但此處顯然無法滿足要求。因此，必須得自定義檔案解析類（當然繼承自FileInputFormat）。

本文給出 讀取解析整個檔案的自定義FileInputFormat——WholeFileInputFormat，最終可實現以檔名作Key，整個檔案內容作Value，儲存至HBase。

package file;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class FileInputFormatMake {
	
	
	public static void main(String arge[]) throws IOException, InterruptedException, ClassNotFoundException{
		Configuration conf =new Configuration();
		//設定zookeeper
		conf.set("hbase.zookeeper.quorum", "hadoop00");
		//設定hbase表名稱
		conf.set(TableOutputFormat.OUTPUT_TABLE, "tool");
		//將該值改大,防止hbase超時退出
		conf.set("dfs.socket.timeout", "180000");
		Job job = new Job(conf,"HBaseBatchImport");
		job.setMapperClass(BatchImportMapper.class);
		job.setReducerClass(BatchImportReducer.class);
		job.setNumReduceTasks(1);
		//設定map的輸出,不設定reduce的輸出型別
		job.setMapOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.setInputFormatClass(WholeFileInputFormat.class);
		//不設定輸出路徑,設定輸出格式型別
		job.setOutputFormatClass(TableOutputFormat.class);
		FileInputFormat.setInputPaths(job, "hdfs://hadoop00:9000/tool");//tool資料夾下有多個檔案
		job.waitForCompletion(true);
		
		
	}
	
	
	static class BatchImportMapper extends Mapper<Text,Text,Text,Text>{
		@Override
		protected void map(Text key, Text value, Context context) {
					try {
						context.write(key, value);
					} catch (IOException e) {
						e.printStackTrace();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
		}
		
	}
	
	
	static class BatchImportReducer extends TableReducer<Text,Text,NullWritable>{
		@Override
		protected void reduce(Text k2, Iterable<Text> v2,Context context)
				throws IOException, InterruptedException {
					for(Text text:v2){
						Put put = new Put(Bytes.toBytes(k2.toString()));
						put.add(Bytes.toBytes("context"), Bytes.toBytes("value"), Bytes.toBytes(text.toString()));
						context.write(NullWritable.get(), put);
					}
		
		}
		
	}
	
	
	
	static class WholeFileInputFormat extends FileInputFormat{
		@Override
		protected boolean isSplitable(JobContext context, Path filename) {
			return false;
		}
		@Override
		public RecordReader createRecordReader(InputSplit split,
				TaskAttemptContext context) throws IOException,
				InterruptedException {
			
			return new WholeFileRecordReader();
		}
	}
	static  class WholeFileRecordReader extends RecordReader{
		private FileSplit fileSplit;
		private FSDataInputStream fis;
		private Text key = null;
		private Text value = null;
		
		private boolean processed = false;
		@Override
		public void initialize(InputSplit inputSplit, TaskAttemptContext context)
				throws IOException, InterruptedException {
			fileSplit = (FileSplit)inputSplit;
			Configuration job = context.getConfiguration();
		    Path file = fileSplit.getPath();
		    FileSystem fs = file.getFileSystem(job);
		    fis = fs.open(file);
			
		}

		@Override
		public boolean nextKeyValue() throws IOException, InterruptedException {
			
			if(key == null){
				key = new Text();
			}
			if(value == null){
				value = new Text();
			}
			if(!processed){
				byte[] content = new byte[(int)fileSplit.getLength()];
				Path file = fileSplit.getPath();
				System.out.println(file.getName());
				key.set(file.getName());
				
				org.apache.hadoop.io.IOUtils.readFully(fis, content, 0, content.length);
				String sendString=new String(  content , "ISO-8859-1" );
				System.out.println(sendString);
				value.set(new Text(sendString));
			processed = true;
			return true;
			}
			return false;
		}

		@Override
		public Text getCurrentKey() throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			return this.key;
		}

		@Override
		public Text getCurrentValue() throws IOException,
				InterruptedException {
			
			return this.value;
		}

		@Override
		public float getProgress() throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			return processed ? fileSplit.getLength() : 0;
		}

		@Override
		public void close() throws IOException {
			
			
		}
		
	}
	
	

}