diff --git a/2017-big-data.iml b/2017-big-data.iml new file mode 100644 index 0000000..e1dd04d --- /dev/null +++ b/2017-big-data.iml @@ -0,0 +1,79 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/answers.txt b/answers.txt index 00fa802..6659395 100644 --- a/answers.txt +++ b/answers.txt @@ -1,3 +1,3 @@ -2. is 126420 -3. 41.602 -4. french 5742 +2. was 18391755 +3. 37.164249490179486 +4. October 1072615 \ No newline at end of file diff --git a/pom.xml b/pom.xml index 16acaa5..fc61ac2 100644 --- a/pom.xml +++ b/pom.xml @@ -1,7 +1,7 @@ 4.0.0 - pritykovskaya + tFNiYaFF 2017-big-data jar 1.0-SNAPSHOT @@ -29,7 +29,7 @@ - pritykovskaya.WordCount + tFNiYaFF.Main @@ -38,8 +38,8 @@ org.apache.maven.plugins maven-compiler-plugin - 1.6 - 1.6 + 1.7 + 1.7 diff --git a/report.txt b/report.txt new file mode 100644 index 0000000..3509248 --- /dev/null +++ b/report.txt @@ -0,0 +1,14 @@ + 1 +2900,292s - wordcount + + 2 +36,356s - +6,076s - 7- + + 3 +78,431s - - + + 4 +78,385s - +6,072s - +1,048s - 5- (5 => 0 to 4) \ No newline at end of file diff --git a/src/main/java/pritykovskaya/WordCount.java b/src/main/java/pritykovskaya/WordCount.java deleted file mode 100644 index b047511..0000000 --- a/src/main/java/pritykovskaya/WordCount.java +++ /dev/null @@ -1,80 +0,0 @@ -package pritykovskaya; - - -import java.io.IOException; -import java.util.StringTokenizer; - -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.conf.Configured; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.io.IntWritable; -import org.apache.hadoop.io.LongWritable; -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.input.TextInputFormat; -import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; -import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; -import org.apache.hadoop.util.Tool; -import org.apache.hadoop.util.ToolRunner; - - -public class WordCount extends Configured implements Tool { - - public static class MyMapper extends Mapper { - private static final IntWritable ONE = new IntWritable(1); - private final transient Text word = new Text(); - - @Override public void map(final LongWritable key, final Text value, final Context context) - throws IOException, InterruptedException { - final String line = value.toString(); - final StringTokenizer tokenizer = new StringTokenizer(line); - while (tokenizer.hasMoreTokens()) { - word.set(tokenizer.nextToken()); - context.write(word, ONE); - } - } - } - - - public static class MyReducer extends Reducer { - - @Override - public void reduce(final Text key, final Iterable values, final Context context) - throws IOException, InterruptedException { - int sum = 0; - for (final IntWritable val : values) { - sum += val.get(); - } - context.write(key, new IntWritable(sum)); - } - } - - - @Override public int run(final String[] args) throws Exception { - final Configuration conf = this.getConf(); - final Job job = Job.getInstance(conf, "Word Count"); - job.setJarByClass(WordCount.class); - - job.setMapperClass(MyMapper.class); - job.setReducerClass(MyReducer.class); - - job.setOutputKeyClass(Text.class); - job.setOutputValueClass(IntWritable.class); - - job.setInputFormatClass(TextInputFormat.class); - job.setOutputFormatClass(TextOutputFormat.class); - - FileInputFormat.addInputPath(job, new Path(args[0])); - FileOutputFormat.setOutputPath(job, new Path(args[1])); - - return job.waitForCompletion(true) ? 0 : 1; - } - - public static void main(final String[] args) throws Exception { - final int returnCode = ToolRunner.run(new Configuration(), new WordCount(), args); - System.exit(returnCode); - } -} diff --git a/src/main/java/tFNiYaFF/Main.java b/src/main/java/tFNiYaFF/Main.java new file mode 100644 index 0000000..3a8f8fb --- /dev/null +++ b/src/main/java/tFNiYaFF/Main.java @@ -0,0 +1,105 @@ +package tFNiYaFF; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.util.Tool; +import org.apache.hadoop.util.ToolRunner; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.List; + +public class Main { + private static final String FILE_REPORT = "report.txt"; + + public static void main(String[] args) throws Exception { + String inputDirectory = "input"; + String wordcountDirectory = "output-wordcount"; + String orderedDirectory = "output-ordered"; + String task2OutputDirectory = "output-7-word"; + String task3OutputDirectory = "output-stopwordscount"; + String namesDirectory = "output-names"; + String namesOrderedDirectory = "output-ordered-names"; + String task4OutputDirectory = "output-5-name"; + + Report task1 = new Report("Задание 1") + .addAction( + "Считаем wordcount", + calculateTaskTime(new WordCount(), inputDirectory, wordcountDirectory) + ); + + Report task2 = new Report("Задание 2") + .addAction( + "Сортируем слова", + calculateTaskTime(new WordsOrder(), wordcountDirectory, orderedDirectory) + ) + .addAction( + "Вывод только 7-го слова", + calculateTaskTime(new NValue(), orderedDirectory, task2OutputDirectory, "6") + ); + + Report task3 = new Report("Задание 3") + .addAction( + "Считаем процент стоп-слов", + calculateTaskTime(new StopWordsCount(), wordcountDirectory, task3OutputDirectory, "stop_words_en.txt") + ); + Report task4 = new Report("Задание 4") + .addAction( + "Находим имена и кладем их в папку", + calculateTaskTime(new NamesPercent(), wordcountDirectory, namesDirectory) + ) + .addAction( + "Сортируем имена по убыванию частоты", + calculateTaskTime(new WordsOrder(), namesDirectory, namesOrderedDirectory) + ) + .addAction( + "Выводим только 5-е имя (5 => 0 to 4)", + calculateTaskTime(new NValue(), namesOrderedDirectory, task4OutputDirectory, "4") + ); + + + Files.write( + Paths.get(FILE_REPORT), + String.join( + "\r\n\r\n", + task1.getString(), + task2.getString(), + task3.getString(), + task4.getString() + ).getBytes() + ); + } + + private static double calculateTaskTime(Tool tool, String... parameters) throws Exception{ + long from = System.currentTimeMillis(); + + ToolRunner.run(new Configuration(), tool, parameters); + + long to = System.currentTimeMillis(); + return (to - from)/(double)1000; + } + + private static class Report { + private List stringList; + + Report(String name) { + stringList = new ArrayList<>(); + stringList.add(name); + } + + Report addAction(String description, double time){ + stringList.add(formatTimeWithText(time, description)); + + return this; + } + + private String formatTimeWithText(double time, String text){ + return String.format("%.3f", time) + "s" + " - " + text; + } + + public String getString() throws IOException { + return String.join("\r\n", stringList); + } + } +} \ No newline at end of file diff --git a/src/main/java/tFNiYaFF/NValue.java b/src/main/java/tFNiYaFF/NValue.java new file mode 100644 index 0000000..c838b22 --- /dev/null +++ b/src/main/java/tFNiYaFF/NValue.java @@ -0,0 +1,117 @@ +package tFNiYaFF; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.conf.Configured; +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.input.TextInputFormat; +import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; +import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; +import org.apache.hadoop.util.Tool; +import org.apache.hadoop.util.ToolRunner; + +import java.io.IOException; +import java.util.SortedSet; +import java.util.TreeSet; + +public class NValue extends Configured implements Tool { + private static int valueNumber; + private static IntWritable ONE = new IntWritable(1); + + static class MyMapper extends Mapper { + private int count; + + @Override + protected void setup(Context context) throws IOException, InterruptedException { + super.setup(context); + count = 0; + } + + @Override + protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { + final String line = value.toString(); + + int pos = line.indexOf(0x09); + + int inputCount = Integer.valueOf(line.substring(0, pos)); + String inputString = line.substring(pos+1); + + if (count <= valueNumber){ + context.write(ONE, new TextWithCountWriteble(inputString, inputCount)); + count++; + } + } + } + + static class MyReducer extends Reducer { + private SortedSet setOfTextWithCount; + + @Override + protected void setup(Context context) throws IOException, InterruptedException { + super.setup(context); + setOfTextWithCount = new TreeSet<>(); + } + + @Override + protected void reduce(IntWritable key, Iterable values, Context context) throws IOException, InterruptedException { + values.forEach((textWithCountWriteble -> { + setOfTextWithCount.add(textWithCountWriteble.clone()); + if (setOfTextWithCount.size() > valueNumber + 1){ + setOfTextWithCount.remove(setOfTextWithCount.last()); + } + })); + } + + @Override + protected void cleanup(Context context) throws IOException, InterruptedException { + super.cleanup(context); + + int i = 0; + for (TextWithCountWriteble textWithCountWriteble : setOfTextWithCount){ + if (i == valueNumber){ + context.write(new Text(textWithCountWriteble.getText()), new IntWritable(textWithCountWriteble.getCount())); + break; + } + i++; + } + } + } + + @Override + public int run(String[] args) throws Exception { + final Configuration conf = this.getConf(); + final Job job = new Job(conf, "NValue"); + + valueNumber = Integer.valueOf(args[2]); + + job.setJarByClass(NValue.class); + + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.class); + + job.setMapperClass(MyMapper.class); + job.setReducerClass(MyReducer.class); + + job.setMapOutputKeyClass(IntWritable.class); + job.setMapOutputValueClass(TextWithCountWriteble.class); + + job.setOutputKeyClass(Text.class); + job.setOutputValueClass(IntWritable.class); + + FileInputFormat.addInputPath(job, new Path(args[0])); + FileOutputFormat.setOutputPath(job, new Path(args[1])); + + return job.waitForCompletion(true) ? 0 : 1; + } + + public static void main(String[] args) throws Exception{ + final int returnCode = ToolRunner.run(new Configuration(), new NValue(), args); + System.exit(returnCode); + + } +} diff --git a/src/main/java/tFNiYaFF/NameCount.java b/src/main/java/tFNiYaFF/NameCount.java new file mode 100644 index 0000000..4638221 --- /dev/null +++ b/src/main/java/tFNiYaFF/NameCount.java @@ -0,0 +1,121 @@ +package tFNiYaFF; + +import java.io.*; +import java.util.*; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.conf.Configured; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.io.LongWritable; +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.input.TextInputFormat; +import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; +import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; +import org.apache.hadoop.util.Tool; +import org.apache.hadoop.util.ToolRunner; + + +public class NameCount extends Configured implements Tool { + + private static Map sortByValues(Map inputMap){ + Map sortedMap = new LinkedHashMap<>(); + ArrayList keys = new ArrayList<>(); + ArrayList values = new ArrayList<>(); + for(Map.Entry pair: inputMap.entrySet()){ + keys.add(pair.getKey()); + values.add(pair.getValue()); + } + for(int i=0; i { + private static final IntWritable ONE = new IntWritable(1); + private final transient Text word = new Text(); + + @Override public void map(final LongWritable key, final Text value, final Context context) + throws IOException, InterruptedException { + final String line = value.toString(); + final StringTokenizer tokenizer = new StringTokenizer(line); + while (tokenizer.hasMoreTokens()) { + word.set(tokenizer.nextToken()); + if(word.toString().matches("^([A-Z])([a-z])+$")){ + context.write(word, ONE); + } + } + } + } + + + public static class MyReducer extends Reducer { + + private Map countMap = new HashMap<>(); + + @Override + public void reduce(final Text key, final Iterable values, final Context context) + throws IOException, InterruptedException { + int sum = 0; + for (final IntWritable val : values) { + sum += val.get(); + } + countMap.put(new Text(key),new IntWritable(sum)); + } + + @Override + protected void cleanup(Context context) throws IOException, InterruptedException { + Map sortedMap = sortByValues(countMap); + int counter = 0; + for (Text key: sortedMap.keySet()) { + if(++counter==5) { + context.write(key, sortedMap.get(key)); + } + } + } + } + + + @Override public int run(final String[] args) throws Exception { + final Configuration conf = this.getConf(); + final Job job = Job.getInstance(conf, "Name Count"); + job.setJarByClass(WordCount.class); + + job.setMapperClass(MyMapper.class); + job.setReducerClass(MyReducer.class); + + job.setOutputKeyClass(Text.class); + job.setOutputValueClass(IntWritable.class); + + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.class); + + FileInputFormat.addInputPath(job, new Path(args[0])); + FileOutputFormat.setOutputPath(job, new Path(args[1])); + + return job.waitForCompletion(true) ? 0 : 1; + } + + + public static void main(final String[] args) throws Exception { + final int returnCode = ToolRunner.run(new Configuration(), new NameCount(), args); + System.exit(returnCode); + } +} diff --git a/src/main/java/tFNiYaFF/NamesPercent.java b/src/main/java/tFNiYaFF/NamesPercent.java new file mode 100644 index 0000000..7d2c978 --- /dev/null +++ b/src/main/java/tFNiYaFF/NamesPercent.java @@ -0,0 +1,99 @@ +package tFNiYaFF; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.conf.Configured; +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.input.TextInputFormat; +import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; +import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; +import org.apache.hadoop.util.Tool; +import org.apache.hadoop.util.ToolRunner; + +import java.io.IOException; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +public class NamesPercent extends Configured implements Tool { + private static final Pattern namePattern = Pattern.compile("^[A-Z][a-z0-9]*$"); + + static class MyMapper extends Mapper { + + @Override + protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { + final String line = value.toString(); + + int pos = line.indexOf(0x09); + + String inputString = line.substring(0, pos); + int inputCount = Integer.valueOf(line.substring(pos+1)); + + context.write(new Text(inputString.toLowerCase()), new TextWithCountWriteble(inputString, inputCount)); + } + } + + static class MyReducer extends Reducer { + + @Override + protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { + int sumAllForms = 0; + int rightFormCount = 0; + String rightFormText = null; + + for (final TextWithCountWriteble value : values){ + sumAllForms += value.getCount(); + + if (rightFormText == null){ + Matcher matcher = namePattern.matcher(value.getText()); + if (matcher.matches()){ + rightFormText = value.getText(); + rightFormCount = value.getCount(); + } + } + } + + if (rightFormText == null){ + return; + } + + if (rightFormCount / (double)sumAllForms >= 0.995){ + context.write(new Text(rightFormText), new IntWritable(rightFormCount)); + } + } + } + + @Override + public int run(String[] args) throws Exception { + final Configuration conf = this.getConf(); + final Job job = new Job(conf, "NamesPercent"); + job.setJarByClass(NamesPercent.class); + + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.class); + + job.setMapperClass(MyMapper.class); + job.setReducerClass(MyReducer.class); + + job.setMapOutputKeyClass(Text.class); + job.setMapOutputValueClass(TextWithCountWriteble.class); + + job.setOutputKeyClass(Text.class); + job.setOutputValueClass(IntWritable.class); + + FileInputFormat.addInputPath(job, new Path(args[0])); + FileOutputFormat.setOutputPath(job, new Path(args[1])); + + return job.waitForCompletion(true) ? 0 : 1; + } + + public static void main(String[] args) throws Exception{ + final int returnCode = ToolRunner.run(new Configuration(), new NamesPercent(), args); + System.exit(returnCode); + + } +} diff --git a/src/main/java/tFNiYaFF/StopWordsCount.java b/src/main/java/tFNiYaFF/StopWordsCount.java new file mode 100644 index 0000000..b87fac9 --- /dev/null +++ b/src/main/java/tFNiYaFF/StopWordsCount.java @@ -0,0 +1,139 @@ +package tFNiYaFF; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.conf.Configured; +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.input.TextInputFormat; +import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; +import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; +import org.apache.hadoop.util.Tool; +import org.apache.hadoop.util.ToolRunner; + +import java.io.File; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.HashSet; +import java.util.Set; + +public class StopWordsCount extends Configured implements Tool { + private static Set stopWords; + + private static void setStopWords(String stopWordsFile){ + stopWords = new HashSet<>(); + + try { + Files + .lines(Paths.get(stopWordsFile), StandardCharsets.UTF_8) + .forEach(stopWords::add); + } catch (IOException e) { + throw new ExceptionInInitializerError(e); + } + } + + enum MY_COUNTERS { + WORDS_COUNT, + STOP_WORDS_COUNT + } + + + static class MyMapper extends Mapper { + + @Override + protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { + final String line = value.toString(); + + int pos = line.indexOf(0x09); + + String inputString = line.substring(0, pos); + int inputCount = Integer.valueOf(line.substring(pos+1)); + + context.write(new Text(inputString.toLowerCase()), new IntWritable(inputCount)); + } + } + + static class MyCombiner extends Reducer { + + @Override + protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { + int valuesCount = 0; + + for (final IntWritable value : values) { + valuesCount += value.get(); + } + + context.write(key, new IntWritable(valuesCount)); + } + } + + static class MyReducer extends Reducer { + + @Override + protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { + boolean isStopWord = stopWords.contains(key.toString()); + + for (final IntWritable value : values) { + context.getCounter(MY_COUNTERS.WORDS_COUNT).increment(value.get()); + + if (isStopWord){ + context.getCounter(MY_COUNTERS.STOP_WORDS_COUNT).increment(value.get()); + } + } + } + } + + + @Override + public int run(String[] args) throws Exception { + final Configuration conf = this.getConf(); + final Job job = new Job(conf, "StopWordsCount"); + + // третьим аргументом приходит файл стоп-слов + setStopWords(args[2]); + + job.setJarByClass(StopWordsCount.class); + + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.class); + + job.setMapperClass(MyMapper.class); + job.setReducerClass(MyReducer.class); + job.setCombinerClass(MyCombiner.class); + + job.setMapOutputKeyClass(Text.class); + job.setMapOutputValueClass(IntWritable.class); + + job.setOutputKeyClass(Text.class); + job.setOutputValueClass(IntWritable.class); + + // читаем из выходной папки wordcount + FileInputFormat.addInputPath(job, new Path(args[0])); + + // в выходную папку положится пустой файл (в context ничего не пишем) + // и в этой же папке создастся файл "output", в котором будет результат работы программы + FileOutputFormat.setOutputPath(job, new Path(args[1])); + + boolean success = job.waitForCompletion(true); + + long stopWordsCount = job.getCounters().findCounter(MY_COUNTERS.STOP_WORDS_COUNT).getValue(); + long wordsCount = job.getCounters().findCounter(MY_COUNTERS.WORDS_COUNT).getValue(); + double percent = stopWordsCount / (double)wordsCount * 100; + + File file = new File(args[1] + "/output"); + Files.write(file.toPath(), String.valueOf(percent).getBytes()); + + return success ? 0 : 1; + } + + public static void main(String[] args) throws Exception{ + final int returnCode = ToolRunner.run(new Configuration(), new StopWordsCount(), args); + System.exit(returnCode); + } +} diff --git a/src/main/java/tFNiYaFF/TextWithCountWriteble.java b/src/main/java/tFNiYaFF/TextWithCountWriteble.java new file mode 100644 index 0000000..6c8ed0d --- /dev/null +++ b/src/main/java/tFNiYaFF/TextWithCountWriteble.java @@ -0,0 +1,76 @@ +package tFNiYaFF; + +import org.apache.hadoop.io.WritableComparable; + +import java.io.DataInput; +import java.io.DataOutput; +import java.io.IOException; + +class TextWithCountWriteble implements WritableComparable, Cloneable { + private String text; + private int count; + + @Override + public void write(DataOutput dataOutput) throws IOException { + dataOutput.writeInt(count); + dataOutput.writeUTF(text); + } + + @Override + public void readFields(DataInput dataInput) throws IOException { + count = dataInput.readInt(); + text = dataInput.readUTF(); + } + + public String getText() { + return text; + } + + public int getCount() { + return count; + } + + TextWithCountWriteble(){ + // should be + } + + TextWithCountWriteble(String text, int count) { + this.text = text; + this.count = count; + } + + @Override + public boolean equals(Object other) { + if (other == null){ + return false; + } + if (other == this){ + return true; + } + if (!(other instanceof TextWithCountWriteble)){ + return false; + } + TextWithCountWriteble otherMyClass = (TextWithCountWriteble)other; + if (otherMyClass.count != count){ + return false; + } + if (!otherMyClass.text.equals(text)){ + return false; + } + return true; + } + + @Override + protected TextWithCountWriteble clone() { + return new TextWithCountWriteble(text, count); + } + + @Override + public int compareTo(TextWithCountWriteble o) { + if (equals(o)){ + return 0; + } + int intCompare = Integer.compare(count, o.count); + return (intCompare == 0) ? Integer.compare(this.hashCode(), o.hashCode()) : -intCompare; + } +} \ No newline at end of file diff --git a/src/main/java/tFNiYaFF/WordCount.java b/src/main/java/tFNiYaFF/WordCount.java new file mode 100644 index 0000000..7f9f796 --- /dev/null +++ b/src/main/java/tFNiYaFF/WordCount.java @@ -0,0 +1,78 @@ +package tFNiYaFF; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.conf.Configured; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.io.LongWritable; +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.input.TextInputFormat; +import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; +import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; +import org.apache.hadoop.util.Tool; +import org.apache.hadoop.util.ToolRunner; + +import java.io.IOException; +import java.util.StringTokenizer; + +public class WordCount extends Configured implements Tool { + + public static class MyMapper extends Mapper { + private static final IntWritable ONE = new IntWritable(1); + private final transient Text word = new Text(); + + @Override + public void map(final LongWritable key, final Text value, final Context context) + throws IOException, InterruptedException { + final String line = value.toString(); + final StringTokenizer tokenizer = new StringTokenizer(line); + while (tokenizer.hasMoreTokens()) { + word.set(tokenizer.nextToken()); + context.write(word, ONE); + } + } + } + + static class MyReducer extends Reducer { + + @Override + protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { + int sum = 0; + for (final IntWritable val : values) { + sum += val.get(); + } + context.write(key, new IntWritable(sum)); + } + } + + @Override + public int run(String[] args) throws Exception { + final Configuration conf = this.getConf(); + final Job job = new Job(conf, "Word Count"); + job.setJarByClass(WordCount.class); + + job.setMapperClass(MyMapper.class); + job.setReducerClass(MyReducer.class); + job.setCombinerClass(MyReducer.class); + + job.setOutputKeyClass(Text.class); + job.setOutputValueClass(IntWritable.class); + + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.class); + + FileInputFormat.addInputPath(job, new Path(args[0])); + FileOutputFormat.setOutputPath(job, new Path(args[1])); + + return job.waitForCompletion(true) ? 0 : 1; + } + + public static void main(String[] args) throws Exception { + final int returnCode = ToolRunner.run(new Configuration(), new WordCount(), args); + System.exit(returnCode); + } +} diff --git a/src/main/java/tFNiYaFF/WordsOrder.java b/src/main/java/tFNiYaFF/WordsOrder.java new file mode 100644 index 0000000..eb38ce5 --- /dev/null +++ b/src/main/java/tFNiYaFF/WordsOrder.java @@ -0,0 +1,86 @@ +package tFNiYaFF; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.conf.Configured; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.io.WritableComparator; +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.input.TextInputFormat; +import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; +import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; +import org.apache.hadoop.util.Tool; +import org.apache.hadoop.util.ToolRunner; + +import java.io.IOException; + +public class WordsOrder extends Configured implements Tool { + + static class MyMapper extends Mapper { + + @Override + protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { + final String line = value.toString(); + + int pos = line.indexOf(0x09); + + context.write(new IntWritable(Integer.valueOf(line.substring(pos+1))), new Text(line.substring(0, pos))); + } + } + + static class MyReducer extends Reducer { + + @Override + protected void reduce(IntWritable key, Iterable values, Context context) throws IOException, InterruptedException { + for (final Text value : values){ + context.write(key, value); + } + } + } + + static class MyDescFreqComparator extends WritableComparator { + protected MyDescFreqComparator() { + super(IntWritable.class); + } + + @Override + public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) { + return -Integer.compare(readInt(b1, s1), readInt(b2, s2)); + } + } + + @Override + public int run(String[] args) throws Exception { + final Configuration conf = this.getConf(); + final Job job = new Job(conf, "Words Order"); + job.setJarByClass(WordsOrder.class); + + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.class); + + job.setMapperClass(MyMapper.class); + job.setReducerClass(MyReducer.class); + + job.setMapOutputKeyClass(IntWritable.class); + job.setMapOutputValueClass(Text.class); + + job.setOutputKeyClass(IntWritable.class); + job.setOutputValueClass(Text.class); + + job.setSortComparatorClass(MyDescFreqComparator.class); + + FileInputFormat.addInputPath(job, new Path(args[0])); + FileOutputFormat.setOutputPath(job, new Path(args[1])); + + return job.waitForCompletion(true) ? 0 : 1; + } + + public static void main(String[] args) throws Exception{ + final int returnCode = ToolRunner.run(new Configuration(), new WordsOrder(), args); + System.exit(returnCode); + } +}