From ab028e42ec10d28b1b1d4f16c7bace1146d52520 Mon Sep 17 00:00:00 2001 From: Mikhail Burmistrov Date: Mon, 20 Nov 2017 04:47:53 +0300 Subject: [PATCH 01/12] Fix pom xml & project structure --- pom.xml | 4 ++-- src/main/java/{pritykovskaya => mburmistrov}/WordCount.java | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) rename src/main/java/{pritykovskaya => mburmistrov}/WordCount.java (99%) diff --git a/pom.xml b/pom.xml index 16acaa5..5c3cadd 100644 --- a/pom.xml +++ b/pom.xml @@ -1,7 +1,7 @@ 4.0.0 - pritykovskaya + mburmistrov 2017-big-data jar 1.0-SNAPSHOT @@ -29,7 +29,7 @@ - pritykovskaya.WordCount + mburmistrov.WordCount diff --git a/src/main/java/pritykovskaya/WordCount.java b/src/main/java/mburmistrov/WordCount.java similarity index 99% rename from src/main/java/pritykovskaya/WordCount.java rename to src/main/java/mburmistrov/WordCount.java index b047511..0363727 100644 --- a/src/main/java/pritykovskaya/WordCount.java +++ b/src/main/java/mburmistrov/WordCount.java @@ -1,4 +1,4 @@ -package pritykovskaya; +package mburmistrov; import java.io.IOException; From 8310f523f60396aa0667120caf8b0d1792c185b5 Mon Sep 17 00:00:00 2001 From: Mikhail Burmistrov Date: Mon, 20 Nov 2017 04:49:26 +0300 Subject: [PATCH 02/12] Add task runner class --- src/main/java/mburmistrov/TaskRunner.java | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) create mode 100644 src/main/java/mburmistrov/TaskRunner.java diff --git a/src/main/java/mburmistrov/TaskRunner.java b/src/main/java/mburmistrov/TaskRunner.java new file mode 100644 index 0000000..c459e7b --- /dev/null +++ b/src/main/java/mburmistrov/TaskRunner.java @@ -0,0 +1,22 @@ +package mburmistrov; + +import org.apache.hadoop.util.ToolRunner; +import org.apache.hadoop.conf.Configuration; + +public class TaskRunner { + public static void main(String[] args) throws Exception{ + + ToolRunner.run( + new Configuration(), + new WordCount(), + new String[]{"input", "output/1"} + ); + + ToolRunner.run( + new Configuration(), + new WordSort(), + new String[]{"output/1", "output/2"} + ); + } + +} From 7c91364682edcf2671a5301bbe31c3b90667b629 Mon Sep 17 00:00:00 2001 From: Mikhail Burmistrov Date: Mon, 20 Nov 2017 04:50:15 +0300 Subject: [PATCH 03/12] Add WordSort class --- src/main/java/mburmistrov/WordSort.java | 83 +++++++++++++++++++++++++ 1 file changed, 83 insertions(+) create mode 100644 src/main/java/mburmistrov/WordSort.java diff --git a/src/main/java/mburmistrov/WordSort.java b/src/main/java/mburmistrov/WordSort.java new file mode 100644 index 0000000..5e5048e --- /dev/null +++ b/src/main/java/mburmistrov/WordSort.java @@ -0,0 +1,83 @@ +package mburmistrov; + +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.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 WordSort extends Configured implements Tool { + + public static class MyMapper extends Mapper { + + @Override + protected void map(final LongWritable key, final Text value, final Context context) throws IOException, InterruptedException { + final String line = value.toString(); + final String[] splitLine = line.split("\t"); + + context.write(new IntWritable(Integer.valueOf(splitLine[1])), new Text(splitLine[0])); + } + } + + + 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 Comparator extends WritableComparator { + protected Comparator() { + 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(final String[] args) throws Exception { + final Configuration conf = this.getConf(); + final Job job = Job.getInstance(conf, "Word Count"); + job.setJarByClass(WordSort.class); + + job.setMapperClass(WordSort.MyMapper.class); + job.setReducerClass(WordSort.MyReducer.class); + + job.setOutputKeyClass(IntWritable.class); + job.setOutputValueClass(Text.class); + + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.class); + + FileInputFormat.addInputPath(job, new Path(args[0])); + FileOutputFormat.setOutputPath(job, new Path(args[1])); + + job.setSortComparatorClass(Comparator.class); + + 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); + } +} From 73e41b584d51b41ec9d27ca405d10f9ac7696c4d Mon Sep 17 00:00:00 2001 From: Mikhail Burmistrov Date: Tue, 21 Nov 2017 12:40:03 +0300 Subject: [PATCH 04/12] Improve WordCount class --- src/main/java/mburmistrov/WordCount.java | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/src/main/java/mburmistrov/WordCount.java b/src/main/java/mburmistrov/WordCount.java index 0363727..5941efd 100644 --- a/src/main/java/mburmistrov/WordCount.java +++ b/src/main/java/mburmistrov/WordCount.java @@ -27,13 +27,18 @@ public static class MyMapper extends Mapper= 'a' && cToken.charAt(0) <= 'z') || (cToken.charAt(0) >= 'A' && cToken.charAt(0) <= 'Z')) { + word.set(cToken); + context.write(word, ONE); + } } } } From abfe50c8f2d0e825d6954fcae8dba7f9174194b2 Mon Sep 17 00:00:00 2001 From: Mikhail Burmistrov Date: Tue, 21 Nov 2017 13:04:02 +0300 Subject: [PATCH 05/12] Add WordCertainPosition & refactor --- pom.xml | 6 +- src/main/java/mburmistrov/TaskRunner.java | 11 ++ src/main/java/mburmistrov/WordCount.java | 85 ------------- .../java/mburmistrov/task1/WordCount.java | 43 +++++++ .../mburmistrov/task1/WordCountMapper.java | 30 +++++ .../mburmistrov/task1/WordCountReducer.java | 20 ++++ .../java/mburmistrov/task2/TextWCount.java | 93 ++++++++++++++ .../task2/WordCertainPosition.java | 113 ++++++++++++++++++ .../mburmistrov/{ => task2}/WordSort.java | 34 +----- .../mburmistrov/task2/WordSortMapper.java | 19 +++ .../mburmistrov/task2/WordSortReducer.java | 16 +++ 11 files changed, 353 insertions(+), 117 deletions(-) delete mode 100644 src/main/java/mburmistrov/WordCount.java create mode 100644 src/main/java/mburmistrov/task1/WordCount.java create mode 100644 src/main/java/mburmistrov/task1/WordCountMapper.java create mode 100644 src/main/java/mburmistrov/task1/WordCountReducer.java create mode 100644 src/main/java/mburmistrov/task2/TextWCount.java create mode 100644 src/main/java/mburmistrov/task2/WordCertainPosition.java rename src/main/java/mburmistrov/{ => task2}/WordSort.java (64%) create mode 100644 src/main/java/mburmistrov/task2/WordSortMapper.java create mode 100644 src/main/java/mburmistrov/task2/WordSortReducer.java diff --git a/pom.xml b/pom.xml index 5c3cadd..81f6910 100644 --- a/pom.xml +++ b/pom.xml @@ -29,7 +29,7 @@ - mburmistrov.WordCount + mburmistrov.task1.WordCount @@ -38,8 +38,8 @@ org.apache.maven.plugins maven-compiler-plugin - 1.6 - 1.6 + 1.8 + 1.8 diff --git a/src/main/java/mburmistrov/TaskRunner.java b/src/main/java/mburmistrov/TaskRunner.java index c459e7b..2fc1b43 100644 --- a/src/main/java/mburmistrov/TaskRunner.java +++ b/src/main/java/mburmistrov/TaskRunner.java @@ -1,5 +1,8 @@ package mburmistrov; +import mburmistrov.task1.WordCount; +import mburmistrov.task2.WordCertainPosition; +import mburmistrov.task2.WordSort; import org.apache.hadoop.util.ToolRunner; import org.apache.hadoop.conf.Configuration; @@ -17,6 +20,14 @@ public static void main(String[] args) throws Exception{ new WordSort(), new String[]{"output/1", "output/2"} ); + + + ToolRunner.run( + new Configuration(), + new WordCertainPosition(), + new String[]{"output/2", "output/3", "6"} + ); + } } diff --git a/src/main/java/mburmistrov/WordCount.java b/src/main/java/mburmistrov/WordCount.java deleted file mode 100644 index 5941efd..0000000 --- a/src/main/java/mburmistrov/WordCount.java +++ /dev/null @@ -1,85 +0,0 @@ -package mburmistrov; - - -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, " \t\n\r.,:;-[]()_?!'\""); - - String cToken; - - while (tokenizer.hasMoreTokens()) { - cToken = tokenizer.nextToken(); - if ((cToken.charAt(0) >= 'a' && cToken.charAt(0) <= 'z') || (cToken.charAt(0) >= 'A' && cToken.charAt(0) <= 'Z')) { - word.set(cToken); - 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/mburmistrov/task1/WordCount.java b/src/main/java/mburmistrov/task1/WordCount.java new file mode 100644 index 0000000..cd608c7 --- /dev/null +++ b/src/main/java/mburmistrov/task1/WordCount.java @@ -0,0 +1,43 @@ +package mburmistrov.task1; + +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.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 { + @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(WordCountMapper.class); + job.setReducerClass(WordCountReducer.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/mburmistrov/task1/WordCountMapper.java b/src/main/java/mburmistrov/task1/WordCountMapper.java new file mode 100644 index 0000000..90bb14e --- /dev/null +++ b/src/main/java/mburmistrov/task1/WordCountMapper.java @@ -0,0 +1,30 @@ +package mburmistrov.task1; + +import java.io.IOException; +import java.util.StringTokenizer; + +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.io.LongWritable; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapreduce.Mapper; + + +public class WordCountMapper 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, " \t\n\r.,:;-[]()_?!'\""); + + String cToken; + + while (tokenizer.hasMoreTokens()) { + cToken = tokenizer.nextToken(); + if ((cToken.charAt(0) >= 'a' && cToken.charAt(0) <= 'z') || (cToken.charAt(0) >= 'A' && cToken.charAt(0) <= 'Z')) { + word.set(cToken); + context.write(word, ONE); + } + } + } + } \ No newline at end of file diff --git a/src/main/java/mburmistrov/task1/WordCountReducer.java b/src/main/java/mburmistrov/task1/WordCountReducer.java new file mode 100644 index 0000000..89d1cb3 --- /dev/null +++ b/src/main/java/mburmistrov/task1/WordCountReducer.java @@ -0,0 +1,20 @@ +package mburmistrov.task1; + +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapreduce.Reducer; + +import java.io.IOException; + +public class WordCountReducer 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)); + } +} \ No newline at end of file diff --git a/src/main/java/mburmistrov/task2/TextWCount.java b/src/main/java/mburmistrov/task2/TextWCount.java new file mode 100644 index 0000000..0f4b1bf --- /dev/null +++ b/src/main/java/mburmistrov/task2/TextWCount.java @@ -0,0 +1,93 @@ +package mburmistrov.task2; + +import java.io.DataInput; +import java.io.DataOutput; +import java.io.IOException; + +import org.apache.hadoop.io.WritableComparable; + +class TextWCount implements WritableComparable, Cloneable { + + private int count; + private String text; + + + + @Override + public void readFields(DataInput dInput) throws IOException { + count = dInput.readInt(); + + text = dInput.readUTF(); + } + + @Override + public void write(DataOutput dOutput) throws IOException { + dOutput.writeInt(count); + + dOutput.writeUTF(text); + } + + + TextWCount(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 TextWCount)){ + return false; + } + + TextWCount otherMyClass = (TextWCount)other; + + if (!otherMyClass.text.equals(text)){ + return false; + } + + if (otherMyClass.count != count){ + return false; + } + + return true; + } + + @Override + protected TextWCount clone() { + return new TextWCount(text, count); + } + + TextWCount(){ + } + + public String getText() { + return text; + } + + public int getCount() { + return count; + } + + @Override + public int compareTo(TextWCount o) { + if (equals(o)){ + return 0; + } + int intCompare = Integer.compare(count, o.count); + if (intCompare == 0) { + return Integer.compare(this.hashCode(), o.hashCode()); + } else { + return -intCompare; + } + } +} \ No newline at end of file diff --git a/src/main/java/mburmistrov/task2/WordCertainPosition.java b/src/main/java/mburmistrov/task2/WordCertainPosition.java new file mode 100644 index 0000000..9994dfe --- /dev/null +++ b/src/main/java/mburmistrov/task2/WordCertainPosition.java @@ -0,0 +1,113 @@ +package mburmistrov.task2; + +import java.io.IOException; +import java.util.SortedSet; +import java.util.TreeSet; + +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; + +public class WordCertainPosition extends Configured implements Tool { + private static int vNum; + + 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 <= vNum){ + context.write(ONE, new TextWCount(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((textWCount -> setOfTextWithCount.add(textWCount.clone()))); + } + + @Override + protected void cleanup(Context context) throws IOException, InterruptedException { + super.cleanup(context); + + int i = 0; + + for (TextWCount textWCount : setOfTextWithCount){ + if (i == vNum){ + context.write(new Text(textWCount.getText()), new IntWritable(textWCount.getCount())); + break; + } + i++; + } + } + } + + @Override + public int run(String[] args) throws Exception { + final Configuration conf = this.getConf(); + final Job job = new Job(conf, "WordCertainPosition"); + + vNum = Integer.valueOf(args[2]); + + job.setJarByClass(WordCertainPosition.class); + + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.class); + + job.setMapperClass(MyMapper.class); + job.setReducerClass(MyReducer.class); + + job.setMapOutputKeyClass(IntWritable.class); + job.setMapOutputValueClass(TextWCount.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 WordCertainPosition(), args); + System.exit(returnCode); + } +} diff --git a/src/main/java/mburmistrov/WordSort.java b/src/main/java/mburmistrov/task2/WordSort.java similarity index 64% rename from src/main/java/mburmistrov/WordSort.java rename to src/main/java/mburmistrov/task2/WordSort.java index 5e5048e..90fbd1d 100644 --- a/src/main/java/mburmistrov/WordSort.java +++ b/src/main/java/mburmistrov/task2/WordSort.java @@ -1,14 +1,13 @@ -package mburmistrov; +package mburmistrov.task2; +import mburmistrov.task1.WordCount; 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.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; @@ -20,29 +19,6 @@ import java.io.IOException; public class WordSort extends Configured implements Tool { - - public static class MyMapper extends Mapper { - - @Override - protected void map(final LongWritable key, final Text value, final Context context) throws IOException, InterruptedException { - final String line = value.toString(); - final String[] splitLine = line.split("\t"); - - context.write(new IntWritable(Integer.valueOf(splitLine[1])), new Text(splitLine[0])); - } - } - - - 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 Comparator extends WritableComparator { protected Comparator() { super(IntWritable.class); @@ -56,11 +32,11 @@ public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) { @Override public int run(final String[] args) throws Exception { final Configuration conf = this.getConf(); - final Job job = Job.getInstance(conf, "Word Count"); + final Job job = Job.getInstance(conf, "Word Sort"); job.setJarByClass(WordSort.class); - job.setMapperClass(WordSort.MyMapper.class); - job.setReducerClass(WordSort.MyReducer.class); + job.setMapperClass(WordSortMapper.class); + job.setReducerClass(WordSortReducer.class); job.setOutputKeyClass(IntWritable.class); job.setOutputValueClass(Text.class); diff --git a/src/main/java/mburmistrov/task2/WordSortMapper.java b/src/main/java/mburmistrov/task2/WordSortMapper.java new file mode 100644 index 0000000..8ca49c7 --- /dev/null +++ b/src/main/java/mburmistrov/task2/WordSortMapper.java @@ -0,0 +1,19 @@ +package mburmistrov.task2; + +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.io.LongWritable; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapreduce.Mapper; + +import java.io.IOException; + +public class WordSortMapper extends Mapper { + + @Override + protected void map(final LongWritable key, final Text value, final Context context) throws IOException, InterruptedException { + final String line = value.toString(); + final String[] splitLine = line.split("\t"); + + context.write(new IntWritable(Integer.valueOf(splitLine[1])), new Text(splitLine[0])); + } +} \ No newline at end of file diff --git a/src/main/java/mburmistrov/task2/WordSortReducer.java b/src/main/java/mburmistrov/task2/WordSortReducer.java new file mode 100644 index 0000000..7019ff1 --- /dev/null +++ b/src/main/java/mburmistrov/task2/WordSortReducer.java @@ -0,0 +1,16 @@ +package mburmistrov.task2; + +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapreduce.Reducer; + +import java.io.IOException; + +public class WordSortReducer extends Reducer { + @Override + protected void reduce(IntWritable key, Iterable values, Context context) throws IOException, InterruptedException { + for (final Text value : values){ + context.write(key, value); + } + } +} From 309bb94abafe48b7145e72db70f93cd1151d4f06 Mon Sep 17 00:00:00 2001 From: Mikhail Burmistrov Date: Tue, 21 Nov 2017 13:34:23 +0300 Subject: [PATCH 06/12] Add stop words file --- resources/stop_words_en.txt | 319 ++++++++++++++++++++++++++++++++++++ 1 file changed, 319 insertions(+) create mode 100644 resources/stop_words_en.txt diff --git a/resources/stop_words_en.txt b/resources/stop_words_en.txt new file mode 100644 index 0000000..b7454b0 --- /dev/null +++ b/resources/stop_words_en.txt @@ -0,0 +1,319 @@ +a +about +above +across +after +afterwards +again +against +all +almost +alone +along +already +also +although +always +am +among +amongst +amoungst +amount +an +and +another +any +anyhow +anyone +anything +anyway +anywhere +are +around +as +at +back +be +became +because +become +becomes +becoming +been +before +beforehand +behind +being +below +beside +besides +between +beyond +bill +both +bottom +but +by +call +can +cannot +cant +co +computer +con +could +couldnt +cry +de +describe +detail +do +done +down +due +during +each +eg +eight +either +eleven +else +elsewhere +empty +enough +etc +even +ever +every +everyone +everything +everywhere +except +few +fifteen +fify +fill +find +fire +first +five +for +former +formerly +forty +found +four +from +front +full +further +get +give +go +had +has +hasnt +have +he +hence +her +here +hereafter +hereby +herein +hereupon +hers +herse" +him +himse" +his +how +however +hundred +i +ie +if +in +inc +indeed +interest +into +is +it +its +itse" +keep +last +latter +latterly +least +less +ltd +made +many +may +me +meanwhile +might +mill +mine +more +moreover +most +mostly +move +much +must +my +myse" +name +namely +neither +never +nevertheless +next +nine +no +nobody +none +noone +nor +not +nothing +now +nowhere +of +off +often +on +once +one +only +onto +or +other +others +otherwise +our +ours +ourselves +out +over +own +part +per +perhaps +please +put +rather +re +same +see +seem +seemed +seeming +seems +serious +several +she +should +show +side +since +sincere +six +sixty +so +some +somehow +someone +something +sometime +sometimes +somewhere +still +such +system +take +ten +than +that +the +their +them +themselves +then +thence +there +thereafter +thereby +therefore +therein +thereupon +these +they +thick +thin +third +this +those +though +three +through +throughout +thru +thus +to +together +too +top +toward +towards +twelve +twenty +two +un +under +until +up +upon +us +very +via +was +we +well +were +what +whatever +when +whence +whenever +where +whereafter +whereas +whereby +wherein +whereupon +wherever +whether +which +while +whither +who +whoever +whole +whom +whose +why +will +with +within +without +would +yet +you +your +yours +yourself +yourselves From 71ca0152f9e4bed3cad967c6a85c78294bf5627b Mon Sep 17 00:00:00 2001 From: Mikhail Burmistrov Date: Tue, 21 Nov 2017 13:34:49 +0300 Subject: [PATCH 07/12] Add StopWordProportion --- src/main/java/mburmistrov/TaskRunner.java | 8 +- .../task2/WordCertainPosition.java | 21 ++-- .../mburmistrov/task3/StopWordProportion.java | 119 ++++++++++++++++++ 3 files changed, 138 insertions(+), 10 deletions(-) create mode 100644 src/main/java/mburmistrov/task3/StopWordProportion.java diff --git a/src/main/java/mburmistrov/TaskRunner.java b/src/main/java/mburmistrov/TaskRunner.java index 2fc1b43..5ea3864 100644 --- a/src/main/java/mburmistrov/TaskRunner.java +++ b/src/main/java/mburmistrov/TaskRunner.java @@ -3,6 +3,7 @@ import mburmistrov.task1.WordCount; import mburmistrov.task2.WordCertainPosition; import mburmistrov.task2.WordSort; +import mburmistrov.task3.StopWordProportion; import org.apache.hadoop.util.ToolRunner; import org.apache.hadoop.conf.Configuration; @@ -21,13 +22,18 @@ public static void main(String[] args) throws Exception{ new String[]{"output/1", "output/2"} ); - ToolRunner.run( new Configuration(), new WordCertainPosition(), new String[]{"output/2", "output/3", "6"} ); + ToolRunner.run( + new Configuration(), + new StopWordProportion(), + new String[]{"output/1", "output/4", "resources/stop_words_en.txt"} + ); + } } diff --git a/src/main/java/mburmistrov/task2/WordCertainPosition.java b/src/main/java/mburmistrov/task2/WordCertainPosition.java index 9994dfe..68036f8 100644 --- a/src/main/java/mburmistrov/task2/WordCertainPosition.java +++ b/src/main/java/mburmistrov/task2/WordCertainPosition.java @@ -24,7 +24,7 @@ public class WordCertainPosition extends Configured implements Tool { private static IntWritable ONE = new IntWritable(1); - static class MyMapper extends Mapper{ + static class WordCertainPositionMapper extends Mapper{ private int count; @Override @@ -39,17 +39,19 @@ protected void map(Object key, Text value, Context context) throws IOException, int pos = line.indexOf(0x09); - int inputCount = Integer.valueOf(line.substring(0, pos)); - String inputString = line.substring(pos+1); + if (pos >= 0) { + int inputCount = Integer.valueOf(line.substring(0, pos)); + String inputString = line.substring(pos + 1); - if (count <= vNum){ - context.write(ONE, new TextWCount(inputString, inputCount)); - count++; + if (count <= vNum){ + context.write(ONE, new TextWCount(inputString, inputCount)); + count++; + } } } } - static class MyReducer extends Reducer{ + static class WordCertainPositionReducer extends Reducer{ private SortedSet setOfTextWithCount; @Override @@ -91,8 +93,8 @@ public int run(String[] args) throws Exception { job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); - job.setMapperClass(MyMapper.class); - job.setReducerClass(MyReducer.class); + job.setMapperClass(WordCertainPositionMapper.class); + job.setReducerClass(WordCertainPositionReducer.class); job.setMapOutputKeyClass(IntWritable.class); job.setMapOutputValueClass(TextWCount.class); @@ -108,6 +110,7 @@ public int run(String[] args) throws Exception { public static void main(String[] args) throws Exception{ final int returnCode = ToolRunner.run(new Configuration(), new WordCertainPosition(), args); + System.exit(returnCode); } } diff --git a/src/main/java/mburmistrov/task3/StopWordProportion.java b/src/main/java/mburmistrov/task3/StopWordProportion.java new file mode 100644 index 0000000..43a05d6 --- /dev/null +++ b/src/main/java/mburmistrov/task3/StopWordProportion.java @@ -0,0 +1,119 @@ +package mburmistrov.task3; + +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; + +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; + +public class StopWordProportion extends Configured implements Tool { + private static Set stopWordSet; + + private static void setStopWords(String stopWordsFile){ + stopWordSet = new HashSet<>(); + try { + Files.lines(Paths.get(stopWordsFile), StandardCharsets.UTF_8).forEach(stopWordSet::add); + } catch (IOException e) { + throw new ExceptionInInitializerError(e); + } + } + + static class StopWordProportionMapper 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); + + if (pos >= 0) { + 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 StopWordProportionReducer extends Reducer{ + @Override + protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { + boolean isStopWord = stopWordSet.contains(key.toString()); + + for (final IntWritable value : values) { + + context.getCounter(COUNTERS.WORD_COUNT).increment(value.get()); + + if (isStopWord){ + context.getCounter(COUNTERS.STOP_WORD_COUNT).increment(value.get()); + } + } + } + } + + + @Override + public int run(String[] args) throws Exception { + final Configuration conf = this.getConf(); + final Job job = new Job(conf, "StopWordProportion"); + + setStopWords(args[2]); + + job.setJarByClass(StopWordProportion.class); + + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.class); + + job.setMapOutputKeyClass(Text.class); + job.setMapOutputValueClass(IntWritable.class); + + job.setMapperClass(StopWordProportionMapper.class); + job.setReducerClass(StopWordProportionReducer.class); + + job.setOutputKeyClass(Text.class); + job.setOutputValueClass(IntWritable.class); + + FileInputFormat.addInputPath(job, new Path(args[0])); + FileOutputFormat.setOutputPath(job, new Path(args[1])); + + boolean success = job.waitForCompletion(true); + + long stopWordCount = job.getCounters().findCounter(COUNTERS.STOP_WORD_COUNT).getValue(); + long wordCount = job.getCounters().findCounter(COUNTERS.WORD_COUNT).getValue(); + + double percentProportion = stopWordCount / (double) wordCount * 100; + + String outputPath = args[1] + "/output"; + File file = new File(outputPath); + Files.write(file.toPath(), String.valueOf(percentProportion).getBytes()); + + return success ? 0 : 1; + } + + enum COUNTERS { + WORD_COUNT, + STOP_WORD_COUNT + } + + public static void main(String[] args) throws Exception{ + final int returnCode = ToolRunner.run(new Configuration(), new StopWordProportion(), args); + System.exit(returnCode); + } +} From befb3b0e600302dd3fa6553d67e38f55900193ba Mon Sep 17 00:00:00 2001 From: Mikhail Burmistrov Date: Thu, 23 Nov 2017 23:10:52 +0300 Subject: [PATCH 08/12] Add NameWordProportion --- src/main/java/mburmistrov/TaskRunner.java | 25 +++- .../mburmistrov/task1/WordCountReducer.java | 16 +-- .../java/mburmistrov/task2/TextWCount.java | 119 ++++++++-------- .../task2/WordCertainPosition.java | 130 +++++++++--------- src/main/java/mburmistrov/task2/WordSort.java | 3 +- .../mburmistrov/task2/WordSortMapper.java | 12 +- .../mburmistrov/task2/WordSortReducer.java | 10 +- .../mburmistrov/task3/StopWordProportion.java | 126 ++++++++--------- .../mburmistrov/task4/NameWordProportion.java | 48 +++++++ .../task4/NameWordProportionMapper.java | 25 ++++ .../task4/NameWordProportionReducer.java | 45 ++++++ 11 files changed, 348 insertions(+), 211 deletions(-) create mode 100644 src/main/java/mburmistrov/task4/NameWordProportion.java create mode 100644 src/main/java/mburmistrov/task4/NameWordProportionMapper.java create mode 100644 src/main/java/mburmistrov/task4/NameWordProportionReducer.java diff --git a/src/main/java/mburmistrov/TaskRunner.java b/src/main/java/mburmistrov/TaskRunner.java index 5ea3864..f54bd28 100644 --- a/src/main/java/mburmistrov/TaskRunner.java +++ b/src/main/java/mburmistrov/TaskRunner.java @@ -4,6 +4,7 @@ import mburmistrov.task2.WordCertainPosition; import mburmistrov.task2.WordSort; import mburmistrov.task3.StopWordProportion; +import mburmistrov.task4.NameWordProportion; import org.apache.hadoop.util.ToolRunner; import org.apache.hadoop.conf.Configuration; @@ -29,9 +30,27 @@ public static void main(String[] args) throws Exception{ ); ToolRunner.run( - new Configuration(), - new StopWordProportion(), - new String[]{"output/1", "output/4", "resources/stop_words_en.txt"} + new Configuration(), + new StopWordProportion(), + new String[]{"output/1", "output/4", "resources/stop_words_en.txt"} + ); + + ToolRunner.run( + new Configuration(), + new NameWordProportion(), + new String[]{"output/1", "output/5"} + ); + + ToolRunner.run( + new Configuration(), + new WordSort(), + new String[]{"output/5", "output/6"} + ); + + ToolRunner.run( + new Configuration(), + new WordCertainPosition(), + new String[]{"output/6", "output/7", "4"} ); } diff --git a/src/main/java/mburmistrov/task1/WordCountReducer.java b/src/main/java/mburmistrov/task1/WordCountReducer.java index 89d1cb3..bc73a64 100644 --- a/src/main/java/mburmistrov/task1/WordCountReducer.java +++ b/src/main/java/mburmistrov/task1/WordCountReducer.java @@ -8,13 +8,13 @@ public class WordCountReducer 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 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)); + } } \ No newline at end of file diff --git a/src/main/java/mburmistrov/task2/TextWCount.java b/src/main/java/mburmistrov/task2/TextWCount.java index 0f4b1bf..f181e60 100644 --- a/src/main/java/mburmistrov/task2/TextWCount.java +++ b/src/main/java/mburmistrov/task2/TextWCount.java @@ -6,88 +6,87 @@ import org.apache.hadoop.io.WritableComparable; -class TextWCount implements WritableComparable, Cloneable { +public class TextWCount implements WritableComparable, Cloneable { - private int count; - private String text; + private int count; + private String text; + @Override + public void readFields(DataInput dInput) throws IOException { + count = dInput.readInt(); - @Override - public void readFields(DataInput dInput) throws IOException { - count = dInput.readInt(); + text = dInput.readUTF(); + } - text = dInput.readUTF(); - } + @Override + public void write(DataOutput dOutput) throws IOException { + dOutput.writeInt(count); - @Override - public void write(DataOutput dOutput) throws IOException { - dOutput.writeInt(count); + dOutput.writeUTF(text); + } - dOutput.writeUTF(text); - } + public TextWCount(String text, int count) { + this.text = text; - TextWCount(String text, int count) { - this.text = text; + this.count = count; + } - this.count = count; - } + @Override + public boolean equals(Object other) { - @Override - public boolean equals(Object other) { + if (other == null) { + return false; + } - if (other == null){ - return false; - } + if (other == this) { + return true; + } - if (other == this){ - return true; - } + if (!(other instanceof TextWCount)) { + return false; + } - if (!(other instanceof TextWCount)){ - return false; - } + TextWCount otherMyClass = (TextWCount) other; - TextWCount otherMyClass = (TextWCount)other; + if (!otherMyClass.text.equals(text)) { + return false; + } - if (!otherMyClass.text.equals(text)){ - return false; - } + if (otherMyClass.count != count) { + return false; + } - if (otherMyClass.count != count){ - return false; - } + return true; + } - return true; - } + @Override + protected TextWCount clone() { + return new TextWCount(text, count); + } - @Override - protected TextWCount clone() { - return new TextWCount(text, count); - } + TextWCount() { + } - TextWCount(){ - } + public String getText() { + return text; + } - public String getText() { - return text; - } + public int getCount() { + return count; + } - public int getCount() { - return count; + @Override + public int compareTo(TextWCount o) { + if (equals(o)) { + return 0; } - - @Override - public int compareTo(TextWCount o) { - if (equals(o)){ - return 0; - } - int intCompare = Integer.compare(count, o.count); - if (intCompare == 0) { - return Integer.compare(this.hashCode(), o.hashCode()); - } else { - return -intCompare; - } + int intCompare = Integer.compare(count, o.count); + if (intCompare == 0) { + return Integer.compare(this.hashCode(), o.hashCode()); + } else { + return -intCompare; } + } } \ No newline at end of file diff --git a/src/main/java/mburmistrov/task2/WordCertainPosition.java b/src/main/java/mburmistrov/task2/WordCertainPosition.java index 68036f8..8c3533e 100644 --- a/src/main/java/mburmistrov/task2/WordCertainPosition.java +++ b/src/main/java/mburmistrov/task2/WordCertainPosition.java @@ -20,97 +20,97 @@ import org.apache.hadoop.util.ToolRunner; public class WordCertainPosition extends Configured implements Tool { - private static int vNum; + private static int vNum; - private static IntWritable ONE = new IntWritable(1); + private static IntWritable ONE = new IntWritable(1); - static class WordCertainPositionMapper extends Mapper{ - private int count; + static class WordCertainPositionMapper extends Mapper { + private int count; - @Override - protected void setup(Context context) throws IOException, InterruptedException { - super.setup(context); - count = 0; - } + @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(); + @Override + protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { + final String line = value.toString(); - int pos = line.indexOf(0x09); + int pos = line.indexOf(0x09); - if (pos >= 0) { - int inputCount = Integer.valueOf(line.substring(0, pos)); - String inputString = line.substring(pos + 1); + if (pos >= 0) { + int inputCount = Integer.valueOf(line.substring(0, pos)); + String inputString = line.substring(pos + 1); - if (count <= vNum){ - context.write(ONE, new TextWCount(inputString, inputCount)); - count++; - } - } + if (count <= vNum) { + context.write(ONE, new TextWCount(inputString, inputCount)); + count++; } + } } + } - static class WordCertainPositionReducer extends Reducer{ - private SortedSet setOfTextWithCount; + static class WordCertainPositionReducer extends Reducer { + private SortedSet setOfTextWithCount; - @Override - protected void setup(Context context) throws IOException, InterruptedException { - super.setup(context); - setOfTextWithCount = new TreeSet<>(); - } + @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((textWCount -> setOfTextWithCount.add(textWCount.clone()))); - } + @Override + protected void reduce(IntWritable key, Iterable values, Context context) throws IOException, InterruptedException { + values.forEach((textWCount -> setOfTextWithCount.add(textWCount.clone()))); + } - @Override - protected void cleanup(Context context) throws IOException, InterruptedException { - super.cleanup(context); + @Override + protected void cleanup(Context context) throws IOException, InterruptedException { + super.cleanup(context); - int i = 0; + int i = 0; - for (TextWCount textWCount : setOfTextWithCount){ - if (i == vNum){ - context.write(new Text(textWCount.getText()), new IntWritable(textWCount.getCount())); - break; - } - i++; - } + for (TextWCount textWCount : setOfTextWithCount) { + if (i == vNum) { + context.write(new Text(textWCount.getText()), new IntWritable(textWCount.getCount())); + break; } + i++; + } } + } - @Override - public int run(String[] args) throws Exception { - final Configuration conf = this.getConf(); - final Job job = new Job(conf, "WordCertainPosition"); + @Override + public int run(String[] args) throws Exception { + final Configuration conf = this.getConf(); + final Job job = new Job(conf, "WordCertainPosition"); - vNum = Integer.valueOf(args[2]); + vNum = Integer.valueOf(args[2]); - job.setJarByClass(WordCertainPosition.class); + job.setJarByClass(WordCertainPosition.class); - job.setInputFormatClass(TextInputFormat.class); - job.setOutputFormatClass(TextOutputFormat.class); + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.class); - job.setMapperClass(WordCertainPositionMapper.class); - job.setReducerClass(WordCertainPositionReducer.class); + job.setMapperClass(WordCertainPositionMapper.class); + job.setReducerClass(WordCertainPositionReducer.class); - job.setMapOutputKeyClass(IntWritable.class); - job.setMapOutputValueClass(TextWCount.class); + job.setMapOutputKeyClass(IntWritable.class); + job.setMapOutputValueClass(TextWCount.class); - job.setOutputKeyClass(Text.class); - job.setOutputValueClass(IntWritable.class); + job.setOutputKeyClass(Text.class); + job.setOutputValueClass(IntWritable.class); - FileInputFormat.addInputPath(job, new Path(args[0])); - FileOutputFormat.setOutputPath(job, new Path(args[1])); + FileInputFormat.addInputPath(job, new Path(args[0])); + FileOutputFormat.setOutputPath(job, new Path(args[1])); - return job.waitForCompletion(true) ? 0 : 1; - } + return job.waitForCompletion(true) ? 0 : 1; + } - public static void main(String[] args) throws Exception{ - final int returnCode = ToolRunner.run(new Configuration(), new WordCertainPosition(), args); + public static void main(String[] args) throws Exception { + final int returnCode = ToolRunner.run(new Configuration(), new WordCertainPosition(), args); - System.exit(returnCode); - } + System.exit(returnCode); + } } diff --git a/src/main/java/mburmistrov/task2/WordSort.java b/src/main/java/mburmistrov/task2/WordSort.java index 90fbd1d..3958c38 100644 --- a/src/main/java/mburmistrov/task2/WordSort.java +++ b/src/main/java/mburmistrov/task2/WordSort.java @@ -30,7 +30,8 @@ public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) { } } - @Override public int run(final String[] args) throws Exception { + @Override + public int run(final String[] args) throws Exception { final Configuration conf = this.getConf(); final Job job = Job.getInstance(conf, "Word Sort"); job.setJarByClass(WordSort.class); diff --git a/src/main/java/mburmistrov/task2/WordSortMapper.java b/src/main/java/mburmistrov/task2/WordSortMapper.java index 8ca49c7..706145f 100644 --- a/src/main/java/mburmistrov/task2/WordSortMapper.java +++ b/src/main/java/mburmistrov/task2/WordSortMapper.java @@ -9,11 +9,11 @@ public class WordSortMapper extends Mapper { - @Override - protected void map(final LongWritable key, final Text value, final Context context) throws IOException, InterruptedException { - final String line = value.toString(); - final String[] splitLine = line.split("\t"); + @Override + protected void map(final LongWritable key, final Text value, final Context context) throws IOException, InterruptedException { + final String line = value.toString(); + final String[] splitLine = line.split("\t"); - context.write(new IntWritable(Integer.valueOf(splitLine[1])), new Text(splitLine[0])); - } + context.write(new IntWritable(Integer.valueOf(splitLine[1])), new Text(splitLine[0])); + } } \ No newline at end of file diff --git a/src/main/java/mburmistrov/task2/WordSortReducer.java b/src/main/java/mburmistrov/task2/WordSortReducer.java index 7019ff1..4d87f9d 100644 --- a/src/main/java/mburmistrov/task2/WordSortReducer.java +++ b/src/main/java/mburmistrov/task2/WordSortReducer.java @@ -7,10 +7,10 @@ import java.io.IOException; public class WordSortReducer extends Reducer { - @Override - protected void reduce(IntWritable key, Iterable values, Context context) throws IOException, InterruptedException { - for (final Text value : values){ - context.write(key, value); - } + @Override + protected void reduce(IntWritable key, Iterable values, Context context) throws IOException, InterruptedException { + for (final Text value : values) { + context.write(key, value); } + } } diff --git a/src/main/java/mburmistrov/task3/StopWordProportion.java b/src/main/java/mburmistrov/task3/StopWordProportion.java index 43a05d6..f5b52c7 100644 --- a/src/main/java/mburmistrov/task3/StopWordProportion.java +++ b/src/main/java/mburmistrov/task3/StopWordProportion.java @@ -24,96 +24,96 @@ import org.apache.hadoop.util.ToolRunner; public class StopWordProportion extends Configured implements Tool { - private static Set stopWordSet; - - private static void setStopWords(String stopWordsFile){ - stopWordSet = new HashSet<>(); - try { - Files.lines(Paths.get(stopWordsFile), StandardCharsets.UTF_8).forEach(stopWordSet::add); - } catch (IOException e) { - throw new ExceptionInInitializerError(e); - } + private static Set stopWordSet; + + private static void setStopWords(String stopWordsFile) { + stopWordSet = new HashSet<>(); + try { + Files.lines(Paths.get(stopWordsFile), StandardCharsets.UTF_8).forEach(stopWordSet::add); + } catch (IOException e) { + throw new ExceptionInInitializerError(e); } + } - static class StopWordProportionMapper extends Mapper{ + static class StopWordProportionMapper extends Mapper { - @Override - protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { - final String line = value.toString(); + @Override + protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { + final String line = value.toString(); - int pos = line.indexOf(0x09); + int pos = line.indexOf(0x09); - if (pos >= 0) { - String inputString = line.substring(0, pos); - int inputCount = Integer.valueOf(line.substring(pos + 1)); + if (pos >= 0) { + String inputString = line.substring(0, pos); + int inputCount = Integer.valueOf(line.substring(pos + 1)); - context.write(new Text(inputString.toLowerCase()), new IntWritable(inputCount)); - } - } + context.write(new Text(inputString.toLowerCase()), new IntWritable(inputCount)); + } } + } - static class StopWordProportionReducer extends Reducer{ - @Override - protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { - boolean isStopWord = stopWordSet.contains(key.toString()); + static class StopWordProportionReducer extends Reducer { + @Override + protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { + boolean isStopWord = stopWordSet.contains(key.toString()); - for (final IntWritable value : values) { + for (final IntWritable value : values) { - context.getCounter(COUNTERS.WORD_COUNT).increment(value.get()); + context.getCounter(COUNTERS.WORD_COUNT).increment(value.get()); - if (isStopWord){ - context.getCounter(COUNTERS.STOP_WORD_COUNT).increment(value.get()); - } - } + if (isStopWord) { + context.getCounter(COUNTERS.STOP_WORD_COUNT).increment(value.get()); } + } } + } - @Override - public int run(String[] args) throws Exception { - final Configuration conf = this.getConf(); - final Job job = new Job(conf, "StopWordProportion"); + @Override + public int run(String[] args) throws Exception { + final Configuration conf = this.getConf(); + final Job job = new Job(conf, "StopWordProportion"); - setStopWords(args[2]); + setStopWords(args[2]); - job.setJarByClass(StopWordProportion.class); + job.setJarByClass(StopWordProportion.class); - job.setInputFormatClass(TextInputFormat.class); - job.setOutputFormatClass(TextOutputFormat.class); + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.class); - job.setMapOutputKeyClass(Text.class); - job.setMapOutputValueClass(IntWritable.class); + job.setMapOutputKeyClass(Text.class); + job.setMapOutputValueClass(IntWritable.class); - job.setMapperClass(StopWordProportionMapper.class); - job.setReducerClass(StopWordProportionReducer.class); + job.setMapperClass(StopWordProportionMapper.class); + job.setReducerClass(StopWordProportionReducer.class); - job.setOutputKeyClass(Text.class); - job.setOutputValueClass(IntWritable.class); + job.setOutputKeyClass(Text.class); + job.setOutputValueClass(IntWritable.class); - FileInputFormat.addInputPath(job, new Path(args[0])); - FileOutputFormat.setOutputPath(job, new Path(args[1])); + FileInputFormat.addInputPath(job, new Path(args[0])); + FileOutputFormat.setOutputPath(job, new Path(args[1])); - boolean success = job.waitForCompletion(true); + boolean success = job.waitForCompletion(true); - long stopWordCount = job.getCounters().findCounter(COUNTERS.STOP_WORD_COUNT).getValue(); - long wordCount = job.getCounters().findCounter(COUNTERS.WORD_COUNT).getValue(); + long stopWordCount = job.getCounters().findCounter(COUNTERS.STOP_WORD_COUNT).getValue(); + long wordCount = job.getCounters().findCounter(COUNTERS.WORD_COUNT).getValue(); - double percentProportion = stopWordCount / (double) wordCount * 100; + double percentProportion = stopWordCount / (double) wordCount * 100; - String outputPath = args[1] + "/output"; - File file = new File(outputPath); - Files.write(file.toPath(), String.valueOf(percentProportion).getBytes()); + String outputPath = args[1] + "/output"; + File file = new File(outputPath); + Files.write(file.toPath(), String.valueOf(percentProportion).getBytes()); - return success ? 0 : 1; - } + return success ? 0 : 1; + } - enum COUNTERS { - WORD_COUNT, - STOP_WORD_COUNT - } + enum COUNTERS { + WORD_COUNT, + STOP_WORD_COUNT + } - public static void main(String[] args) throws Exception{ - final int returnCode = ToolRunner.run(new Configuration(), new StopWordProportion(), args); - System.exit(returnCode); - } + public static void main(String[] args) throws Exception { + final int returnCode = ToolRunner.run(new Configuration(), new StopWordProportion(), args); + System.exit(returnCode); + } } diff --git a/src/main/java/mburmistrov/task4/NameWordProportion.java b/src/main/java/mburmistrov/task4/NameWordProportion.java new file mode 100644 index 0000000..9b4709d --- /dev/null +++ b/src/main/java/mburmistrov/task4/NameWordProportion.java @@ -0,0 +1,48 @@ +package mburmistrov.task4; + +import mburmistrov.task2.TextWCount; +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.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 NameWordProportion extends Configured implements Tool { + + @Override + public int run(String[] args) throws Exception { + final Configuration conf = this.getConf(); + final Job job = new Job(conf, "NameWordProportion"); + job.setJarByClass(NameWordProportion.class); + + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.class); + + job.setMapperClass(NameWordProportionMapper.class); + job.setReducerClass(NameWordProportionReducer.class); + + job.setMapOutputKeyClass(Text.class); + job.setMapOutputValueClass(TextWCount.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 NameWordProportion(), args); + System.exit(returnCode); + + } +} diff --git a/src/main/java/mburmistrov/task4/NameWordProportionMapper.java b/src/main/java/mburmistrov/task4/NameWordProportionMapper.java new file mode 100644 index 0000000..0a4ef10 --- /dev/null +++ b/src/main/java/mburmistrov/task4/NameWordProportionMapper.java @@ -0,0 +1,25 @@ +package mburmistrov.task4; + +import mburmistrov.task2.TextWCount; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapreduce.Mapper; + +import java.io.IOException; + +public class NameWordProportionMapper 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); + + if (pos >= 0) { + + String inputString = line.substring(0, pos); + int inputCount = Integer.valueOf(line.substring(pos + 1)); + + context.write(new Text(inputString.toLowerCase()), new TextWCount(inputString, inputCount)); + } + } +} \ No newline at end of file diff --git a/src/main/java/mburmistrov/task4/NameWordProportionReducer.java b/src/main/java/mburmistrov/task4/NameWordProportionReducer.java new file mode 100644 index 0000000..27ad483 --- /dev/null +++ b/src/main/java/mburmistrov/task4/NameWordProportionReducer.java @@ -0,0 +1,45 @@ +package mburmistrov.task4; + +import java.io.IOException; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import mburmistrov.task2.TextWCount; +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapreduce.Reducer; + +public class NameWordProportionReducer extends Reducer { + private static final Pattern nameWordPattern = Pattern.compile("^[A-Z][a-z0-9]*$"); + + @Override + protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { + int allCount = 0; + int correctCount = 0; + String correctText = null; + double correctPercent = 0.995; + + for (final TextWCount value : values) { + + allCount += value.getCount(); + + if (correctText == null) { + Matcher matcher = nameWordPattern.matcher(value.getText()); + + if (matcher.matches()) { + correctText = value.getText(); + + correctCount = value.getCount(); + } + } + } + + if (correctText == null) { + return; + } + + if (correctCount / (double) allCount >= correctPercent) { + context.write(new Text(correctText), new IntWritable(correctCount)); + } + } +} \ No newline at end of file From 20b5bf2e1f2d5877721678c5a7da39cc17d9bfcd Mon Sep 17 00:00:00 2001 From: Mikhail Burmistrov Date: Thu, 23 Nov 2017 23:31:24 +0300 Subject: [PATCH 09/12] Improve task runner --- src/main/java/mburmistrov/TaskRunner.java | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/src/main/java/mburmistrov/TaskRunner.java b/src/main/java/mburmistrov/TaskRunner.java index f54bd28..5127597 100644 --- a/src/main/java/mburmistrov/TaskRunner.java +++ b/src/main/java/mburmistrov/TaskRunner.java @@ -11,46 +11,50 @@ public class TaskRunner { public static void main(String[] args) throws Exception{ + // task 1 ToolRunner.run( new Configuration(), new WordCount(), - new String[]{"input", "output/1"} + new String[]{"input", "output/1_wordCount"} ); + // task 2 ToolRunner.run( new Configuration(), new WordSort(), - new String[]{"output/1", "output/2"} + new String[]{"output/1_wordCount", "output/2_wordSort"} ); ToolRunner.run( new Configuration(), new WordCertainPosition(), - new String[]{"output/2", "output/3", "6"} + new String[]{"output/2_wordSort", "output/2_seventhWord", "6"} ); + // task 3 ToolRunner.run( new Configuration(), new StopWordProportion(), - new String[]{"output/1", "output/4", "resources/stop_words_en.txt"} + new String[]{"output/1_wordCount", "output/3_stopWordProportion", "resources/stop_words_en.txt"} ); + // task 4 ToolRunner.run( new Configuration(), new NameWordProportion(), - new String[]{"output/1", "output/5"} + new String[]{"output/1_wordCount", "output/4_nameWordProportion"} ); ToolRunner.run( new Configuration(), new WordSort(), - new String[]{"output/5", "output/6"} + new String[]{"output/4_nameWordProportion", "output/4_nameWordSort"} ); ToolRunner.run( new Configuration(), new WordCertainPosition(), - new String[]{"output/6", "output/7", "4"} + new String[]{"output/4_nameWordSort", "output/4_nameFifth", "4"} ); } From 4911d62eba97cc189cea87042ad13cf0fe45fee6 Mon Sep 17 00:00:00 2001 From: Mikhail Burmistrov Date: Sun, 7 Jan 2018 12:27:21 +0300 Subject: [PATCH 10/12] Improve task runner with task manager --- src/main/java/mburmistrov/TaskManager.java | 25 +++++++++ src/main/java/mburmistrov/TaskRunner.java | 65 +++++++++------------- 2 files changed, 50 insertions(+), 40 deletions(-) create mode 100644 src/main/java/mburmistrov/TaskManager.java diff --git a/src/main/java/mburmistrov/TaskManager.java b/src/main/java/mburmistrov/TaskManager.java new file mode 100644 index 0000000..f7e9ec1 --- /dev/null +++ b/src/main/java/mburmistrov/TaskManager.java @@ -0,0 +1,25 @@ +package mburmistrov; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +public class TaskManager { + private List stringList; + + TaskManager(String taskName) { + stringList = new ArrayList<>(); + + stringList.add(taskName); + } + + TaskManager performTask(String description, double time){ + stringList.add(String.format("%.3f", time) + "s" + " - " + description); + + return this; + } + + public String getString() throws IOException { + return String.join("\r\n", stringList); + } +} \ No newline at end of file diff --git a/src/main/java/mburmistrov/TaskRunner.java b/src/main/java/mburmistrov/TaskRunner.java index 5127597..5901e1b 100644 --- a/src/main/java/mburmistrov/TaskRunner.java +++ b/src/main/java/mburmistrov/TaskRunner.java @@ -5,58 +5,43 @@ import mburmistrov.task2.WordSort; import mburmistrov.task3.StopWordProportion; import mburmistrov.task4.NameWordProportion; +import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import org.apache.hadoop.conf.Configuration; +import java.nio.file.Files; +import java.nio.file.Paths; + public class TaskRunner { - public static void main(String[] args) throws Exception{ + private static double getTaskTime(Tool tool, String... parameters) throws Exception{ + long timeFrom = System.currentTimeMillis(); + + ToolRunner.run(new Configuration(), tool, parameters); + + long timeTo = System.currentTimeMillis(); + return (timeTo - timeFrom) / (double) 1000; + } + + public static void main(String[] args) throws Exception { // task 1 - ToolRunner.run( - new Configuration(), - new WordCount(), - new String[]{"input", "output/1_wordCount"} - ); + TaskManager t1 = new TaskManager("task 1").performTask("word count", getTaskTime(new WordCount(),"input", "output/1_wordCount")); // task 2 - ToolRunner.run( - new Configuration(), - new WordSort(), - new String[]{"output/1_wordCount", "output/2_wordSort"} - ); - - ToolRunner.run( - new Configuration(), - new WordCertainPosition(), - new String[]{"output/2_wordSort", "output/2_seventhWord", "6"} - ); - + TaskManager t2 = new TaskManager("task 2") + .performTask("word sort", getTaskTime(new WordSort(),"output/1_wordCount", "output/2_wordSort")) + .performTask("output seventh word", getTaskTime(new WordCertainPosition(),"output/2_wordSort", "output/2_seventhWord", "6")); + // task 3 - ToolRunner.run( - new Configuration(), - new StopWordProportion(), - new String[]{"output/1_wordCount", "output/3_stopWordProportion", "resources/stop_words_en.txt"} - ); + TaskManager t3 = new TaskManager("task 3").performTask("stop word proportion", getTaskTime(new StopWordProportion(),"output/1_wordCount", "output/3_stopWordProportion", "resources/stop_words_en.txt")); // task 4 - ToolRunner.run( - new Configuration(), - new NameWordProportion(), - new String[]{"output/1_wordCount", "output/4_nameWordProportion"} - ); - - ToolRunner.run( - new Configuration(), - new WordSort(), - new String[]{"output/4_nameWordProportion", "output/4_nameWordSort"} - ); - - ToolRunner.run( - new Configuration(), - new WordCertainPosition(), - new String[]{"output/4_nameWordSort", "output/4_nameFifth", "4"} - ); + TaskManager t4 = new TaskManager("task 4") + .performTask("name word proportion", getTaskTime(new NameWordProportion(),"output/1_wordCount", "output/4_nameWordProportion")) + .performTask("name word sort", getTaskTime(new WordSort(),"output/4_nameWordProportion", "output/4_nameWordSort")) + .performTask("name fifth", getTaskTime(new WordCertainPosition(),"output/4_nameWordSort", "output/4_nameFifth", "4")); + Files.write(Paths.get("output/timeReport"), String.join("\r\n\r\n", t1.getString(), t2.getString(), t3.getString(), t4.getString()).getBytes()); } } From 0da9a647d0a615b9cd3209aba33cbd00406afe6d Mon Sep 17 00:00:00 2001 From: Mikhail Burmistrov Date: Wed, 10 Jan 2018 22:41:08 +0300 Subject: [PATCH 11/12] Add combiners --- src/main/java/mburmistrov/TaskRunner.java | 2 +- src/main/java/mburmistrov/task1/WordCount.java | 1 + .../java/mburmistrov/task2/WordCertainPosition.java | 8 ++++++-- .../java/mburmistrov/task3/StopWordProportion.java | 12 ++++++++++++ 4 files changed, 20 insertions(+), 3 deletions(-) diff --git a/src/main/java/mburmistrov/TaskRunner.java b/src/main/java/mburmistrov/TaskRunner.java index 5901e1b..c5317da 100644 --- a/src/main/java/mburmistrov/TaskRunner.java +++ b/src/main/java/mburmistrov/TaskRunner.java @@ -31,7 +31,7 @@ public static void main(String[] args) throws Exception { TaskManager t2 = new TaskManager("task 2") .performTask("word sort", getTaskTime(new WordSort(),"output/1_wordCount", "output/2_wordSort")) .performTask("output seventh word", getTaskTime(new WordCertainPosition(),"output/2_wordSort", "output/2_seventhWord", "6")); - + // task 3 TaskManager t3 = new TaskManager("task 3").performTask("stop word proportion", getTaskTime(new StopWordProportion(),"output/1_wordCount", "output/3_stopWordProportion", "resources/stop_words_en.txt")); diff --git a/src/main/java/mburmistrov/task1/WordCount.java b/src/main/java/mburmistrov/task1/WordCount.java index cd608c7..7145312 100644 --- a/src/main/java/mburmistrov/task1/WordCount.java +++ b/src/main/java/mburmistrov/task1/WordCount.java @@ -23,6 +23,7 @@ public class WordCount extends Configured implements Tool { job.setMapperClass(WordCountMapper.class); job.setReducerClass(WordCountReducer.class); + job.setCombinerClass(WordCountReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); diff --git a/src/main/java/mburmistrov/task2/WordCertainPosition.java b/src/main/java/mburmistrov/task2/WordCertainPosition.java index 8c3533e..4268b9a 100644 --- a/src/main/java/mburmistrov/task2/WordCertainPosition.java +++ b/src/main/java/mburmistrov/task2/WordCertainPosition.java @@ -62,7 +62,12 @@ protected void setup(Context context) throws IOException, InterruptedException { @Override protected void reduce(IntWritable key, Iterable values, Context context) throws IOException, InterruptedException { - values.forEach((textWCount -> setOfTextWithCount.add(textWCount.clone()))); + values.forEach((textWCount -> { + setOfTextWithCount.add(textWCount.clone()); + if (setOfTextWithCount.size() > vNum + 1){ + setOfTextWithCount.remove(setOfTextWithCount.last()); + } + })); } @Override @@ -110,7 +115,6 @@ public int run(String[] args) throws Exception { public static void main(String[] args) throws Exception { final int returnCode = ToolRunner.run(new Configuration(), new WordCertainPosition(), args); - System.exit(returnCode); } } diff --git a/src/main/java/mburmistrov/task3/StopWordProportion.java b/src/main/java/mburmistrov/task3/StopWordProportion.java index f5b52c7..e5d140e 100644 --- a/src/main/java/mburmistrov/task3/StopWordProportion.java +++ b/src/main/java/mburmistrov/task3/StopWordProportion.java @@ -68,6 +68,17 @@ protected void reduce(Text key, Iterable values, Context context) t } } + static class StopWordProportionCombiner extends Reducer{ + @Override + protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { + int vCount = 0; + for (final IntWritable value : values) { + vCount += value.get(); + } + context.write(key, new IntWritable(vCount)); + } + } + @Override public int run(String[] args) throws Exception { @@ -86,6 +97,7 @@ public int run(String[] args) throws Exception { job.setMapperClass(StopWordProportionMapper.class); job.setReducerClass(StopWordProportionReducer.class); + job.setCombinerClass(StopWordProportionCombiner.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); From a1e3a286b515a780120a43ab4009117ec0180db5 Mon Sep 17 00:00:00 2001 From: Mikhail Burmistrov Date: Thu, 11 Jan 2018 20:02:10 +0300 Subject: [PATCH 12/12] Add time report --- timeReport.txt | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) create mode 100644 timeReport.txt diff --git a/timeReport.txt b/timeReport.txt new file mode 100644 index 0000000..5155df3 --- /dev/null +++ b/timeReport.txt @@ -0,0 +1,27 @@ +Подсчет до добавления комбайнеров: + +Задание №1 +34047,536 сек. + +Задание №2 +54,232 сек. + +Задание №3 +92,131 сек. + +Задание №4 +158,679 сек. + +Подсчет после добавления комбайнеров: + +Задание №1 +4205,871 сек. + +Задание №2 +60,588 сек. + +Задание №3 +110,276 сек. + +Задание №4 +172,679 сек.