Java版本
原文件：
package com.jim; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapreduce.Job; /* * MapReduce jobs are typically implemented by using a driver class. * The purpose of a driver class is to set up the configuration for the * MapReduce job and to run the job. * Typical requirements for a driver class include configuring the input * and output data formats, configuring the map and reduce classes, * and specifying intermediate data formats. * * The following is the code for the driver class: */ /** * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ import java.io.IOException; import java.util.StringTokenizer; 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.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; public class WordCount { public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{ private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(Object key, Text value, Context context ) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } } public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context ) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); System.out.println(otherArgs.length); if (otherArgs.length != 2) { System.err.println("Usage: wordcount <in> <out>"); System.out.println("input output"); System.exit(2); } Job job = new Job(conf, "word count"); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }
执行程序：
hadoop jar WordCount.jar /input /output/blog
查看结果：
hdfs dfs -cat /output/blog/part-r-00000
最终结果如图：
整体shell脚本：
#!/bin/bash hdfs dfs -rm -r /output/blog hdfs dfs -rm -r /input/* hdfs dfs -put blog.csv /input hadoop jar WordCount.jar /input /output/blog hdfs dfs -ls /output/blog/part-r-00000 hdfs dfs -get /output/blog/part-r-00000 blog.result cat blog.result
Python版本
Mapper.py
#!/usr/bin/env python #coding=utf-8 import sys import os #标准输入 def readfile(): #f = open(filename) lines = sys.stdin #print type(lines) for line in lines: #line = line.strip() #print line words = line.split(",") #print type(words) for word in words: print word,",",1 readfile()
Reduce.py
#!/usr/bin/env python #coding=utf-8 import sys import os from operator import itemgetter def reduce1(): current_word = None current_count = 0 word = None lines = sys.stdin print current_word for line in lines: line = line.strip() #print line try: word,count = line.split(',',1) except: continue # print "word:%s count:%s"%(word,count) try: count = int(count) except: continue if current_word == word: current_count += count # print "current_word:%s" % (current_word) else: if current_word: #print "current_word不为空" #pass print '%s %s' % (current_word, current_count) current_count = count current_word = word if current_word == word: print '%s %s' % (current_word, current_count) reduce1()
执行程序：
#!/bin/bash alias dt='date +%Y%m%d" "%H:%M:%S' shopt -s expand_aliases function init() { hdfs dfs -rmr /input/* hdfs dfs -put ./blogdata/blog_20170414.csv /input/ hdfs dfs -ls /input/ hdfs dfs -rmr /output/emp } function mapreduce() { hadoop jar /usr/lib/hadoop-mapreduce/hadoop-streaming-2.0.0-cdh4.1.1.jar -mapper "python mapper.py" -reducer "python reducer.py" -input /input/* -output /output/emp -file "/etl/etldata/script/python/mapper.py" -file "/etl/etldata/script/python/reducer.py" -jobconf mapred.map.tasks=10 -jobconf mapred.reduce.tasks=10 } echo "`dt`:1.start init..." > ./wordcount.log init echo "`dt`:2.end init.." >> ./wordcount.log echo "`dt`:3.start MapReduce......" >> ./wordcount.log mapreduce echo "`dt`:4.end MapReduce....." >> ./wordcount.log
Java操作HDFS
创建目录：
package com.jim; import java.io.IOException; import java.net.URI; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; public class Hdfs1 { /** * @param args * @throws IOException */ public static void main(String[] args) { // TODO Auto-generated method stub System.out.println("Start....."); //String uri = args[0]; String uri = "/input/jjm"; Configuration conf = new Configuration(); FileSystem fs = null; try { fs = FileSystem.get(URI.create(uri),conf); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } Path dfs = new Path(uri); try { fs.mkdirs(dfs); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } System.out.println("OK"); } }
上传文件、创建和删除目录
package com.jim; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import java.io.BufferedInputStream; import java.io.IOException; import java.io.InputStream; import java.net.URI; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import java.io.BufferedInputStream; import java.io.FileInputStream; import java.io.FileNotFoundException; import java.io.InputStream; import java.io.OutputStream; import org.apache.hadoop.io.IOUtils; public class Hdfs { /** * @param args */ public static void main(String[] args) { // TODO Auto-generated method stub System.out.println("Start...."); dfs.mkdir("/input/jim"); dfs.deletedir("/input/jim/jack"); System.out.println("OK"); String srcfile="/etl/etldata/script/python/emp.csv"; String trgfile="/input/jim/jim.csv"; try { dfs.putfile(srcfile, trgfile); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); } } public static class dfs { public static Configuration conf = new Configuration(); //create directory public static void mkdir(String string) { String uri=string; try { FileSystem fs =FileSystem.get(URI.create(uri),conf); Path dfs = new Path(uri); fs.mkdirs(dfs); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace();} } //delete dir public static void deletedir(String string){ String uri=string; try { FileSystem fs =FileSystem.get(URI.create(uri),conf); Path dfs = new Path(uri); fs.delete(dfs, true); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } } // put file to hdfs public static void putfile(String srcfile ,String trgfile) throws Exception{ InputStream in = new BufferedInputStream(new FileInputStream(srcfile)); Configuration conf = new Configuration(); FileSystem fs = FileSystem.get(URI.create(trgfile),conf); OutputStream out = fs.create(new Path(trgfile)); IOUtils.copyBytes(in, out, 4096, true); } } }
Python提取文章关键字
#coding:utf-8 import sys sys.path.append('/home/shutong/crawl/tools') from tools import * import jieba import jieba.posseg #需要另外加载一个词性标注模块 from jieba import analyse # 引入TextRank关键词抽取接口 #设置utf-8模式 reload(sys) sys.setdefaultencoding( "utf-8" ) print getTime() textrank = analyse.textrank def get_key_words(line): words = '' keywords = textrank(line) # 输出抽取出的关键词 for keyword in keywords: #print keyword words = words + keyword + ',' return words # 原始文本 text = "经常有初学者在博客和QQ问我，自己想往大数据方向发展，该学哪些技术，学习路线是什么样的，觉得大数据很火，就业很好，薪资很高。如果自己很迷茫，为了这些原因想往大数据方向发展，也可以，那么我就想问一下，你的专业是什么，对于计算机/软件，你的兴趣是什么？是计算机专业，对操作系统、硬件、网络、服务器感兴趣？是软件专业，对软件开发、编程、写代码感兴趣？还是数学、统计学专业，对数据和数字特别感兴趣" #print get_key_words(text) #inputfilename = '/home/shutong/mapreduce/context.txt' #outputfile = '/home/shutong/mapreduce/mapcontext.txt' inputfilename = sys.argv[1] outputfile = sys.argv[2] keywords = '' for line in getLines(inputfilename): line = line.strip() keywords = keywords + get_key_words(line) #print keywords deleteFile(outputfile) saveFile(outputfile,',',keywords) #print " keywords by textrank:" # 基于TextRank算法进行关键词抽取 #keywords = textrank(text) # 输出抽取出的关键词 #for keyword in keywords: # print keyword