HBase中怎么把數據寫到HDFS文件中

這篇文章主要講解了“HBase中怎么把數據寫到HDFS文件中”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“HBase中怎么把數據寫到HDFS文件中”吧！

從HBase中選定某張表，比如blog，然后限定表中的某些列，比如昵稱nickname，標簽tags，將這些列的數據內容寫入到HDFS文件中去。

/**
 * 選定HBase中的某張表，限定列，然后將其內容數據寫入到HDFS文件中。
 * */
public class HBaseAndMapReduce2 {

	public static void main(String[] args) throws Exception {
		System.exit(run());
	}

	public static int run() throws Exception {
		Configuration conf = new Configuration();
		conf = HBaseConfiguration.create(conf);
		conf.set("hbase.zookeeper.quorum", "192.168.226.129");

		Job job = Job.getInstance(conf, "findFriend");
		job.setJarByClass(HBaseAndMapReduce2.class);


		Scan scan = new Scan();
		//取對業務有用的數據 tags, nickname
		scan.addColumn(Bytes.toBytes("article"), Bytes.toBytes("tags"));
		scan.addColumn(Bytes.toBytes("author"), Bytes.toBytes("nickname"));

		//ImmutableBytesWritable來自hbase數據的類型
		/**
		 * public static void initTableMapperJob(String table, Scan scan,
		 * 		Class<? extends TableMapper> mapper,
		 * 		Class<?> outputKeyClass,
		 * 		Class<?> outputValueClass, Job job)
		 * */
		//確保 blog 表存在， 且表結構與本文一樣。 
		TableMapReduceUtil.initTableMapperJob("blog", scan, FindFriendMapper.class, 
				Text.class,  Text.class, job);
		DateFormat df = new SimpleDateFormat( "yyyyMMddHHmmssS" );
		
		FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.226.129:9000/hbasemapreduce1/" +df.format( new Date() )));

		job.setReducerClass(FindFriendReducer.class);
		
		return job.waitForCompletion(true) ? 0 : 1;
	}
    
    // 輸入輸出的鍵值
	public static class FindFriendMapper extends TableMapper<Text, Text>{
		//key是hbase中的行鍵
		//value是hbase中的所行鍵的所有數據
		@Override
		protected void map(
				ImmutableBytesWritable key,
				Result value,
				Mapper<ImmutableBytesWritable, Result,Text, Text>.Context context)
						throws IOException, InterruptedException {

			Text v = null;
			String[] kStrs = null;
			List<Cell> cs = value.listCells();
			for (Cell cell : cs) {
				System.out.println( "Cell--->： "+cell );
				if ("tags".equals(Bytes.toString(CellUtil.cloneQualifier(cell)))){
					kStrs = Bytes.toString(CellUtil.cloneValue(cell)).split(",");
				}else if ("nickname".equals(Bytes.toString(CellUtil.cloneQualifier(cell)))){
					v = new Text(CellUtil.cloneValue(cell));
				}
			}

			for (String kStr : kStrs) {
				context.write(new Text(kStr.toLowerCase()), v);
			}

		}
	}

	public static class FindFriendReducer extends Reducer<Text, Text, Text, Text>{
		@Override
		protected void reduce(Text key, Iterable<Text> values,
				Reducer<Text, Text, Text, Text>.Context context)
						throws IOException, InterruptedException {
			System.out.println(  "key--->"+key);
			StringBuilder sb = new StringBuilder();
			for (Text text : values) {
				System.out.println(  "value-->"+text);
				sb.append((sb.length() > 0 ? ",":"") + text.toString());
			}
			context.write(key, new Text(sb.toString()));
		}
	}
}

輸出結構：

hadoop	Berg-OSChina,BergBerg
hbase	OSChina,BergBerg
zookeeper	OSChina,BergBerg

感謝各位的閱讀，以上就是“HBase中怎么把數據寫到HDFS文件中”的內容了，經過本文的學習后，相信大家對HBase中怎么把數據寫到HDFS文件中這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！