diff --git a/2017-big-data.iml b/2017-big-data.iml new file mode 100644 index 0000000..ce21085 --- /dev/null +++ b/2017-big-data.iml @@ -0,0 +1,79 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/README.md b/README.md index 997652b..8caead4 100644 --- a/README.md +++ b/README.md @@ -1,27 +1,7 @@ # 2017-big-data Репозиторий для практических домашних заданий 2017 года курса "Введение в машинное обучение для java-разработчиков" в [Технополис](https://polis.mail.ru). -### Fork -[Форкните проект](https://help.github.com/articles/fork-a-repo/), склонируйте и добавьте `upstream`: -``` -$ git clone git@github.com:/2017-big-data.git -Cloning into '2017-big-data'... -remote: Counting objects: 3, done. -remote: Compressing objects: 100% (2/2), done. -remote: Total 3 (delta 2), reused 3 (delta 2), pack-reused 0 -Receiving objects: 100% (3/3), 10.1 KiB | 4.5 MiB/s, done. -Resolving deltas: 100% (2/2), done. -$ git remote add upstream git@github.com:polis-mail-ru/2017-big-data.git -$ git fetch upstream -From github.com:polis-mail-ru/2017-big-data - * [new branch] master -> upstream/master -``` +### Job run time -### Develop -Откройте в IDE -- [IntelliJ IDEA Community Edition](https://www.jetbrains.com/idea/) -через File->Open, выбрав pom.xml - -Пример конфигурация для запуска таски из idea можно посмотреть в файле config.png - -В своём Java package `` выполните все задания. +Время работы job WordCount без combiner составило 47m 38.473s. С использованием combiner, после добавления combiner время работы составило 31m 38.491s. diff --git a/pom.xml b/pom.xml index 16acaa5..68016de 100644 --- a/pom.xml +++ b/pom.xml @@ -29,7 +29,7 @@ - pritykovskaya.WordCount + kubrin.WordCount @@ -38,8 +38,8 @@ org.apache.maven.plugins maven-compiler-plugin - 1.6 - 1.6 + 1.8 + 1.8 diff --git a/runtime.txt b/runtime.txt new file mode 100644 index 0000000..31f7b1d --- /dev/null +++ b/runtime.txt @@ -0,0 +1,18 @@ +WordCount without combiner +real 47m38.473s +user 44m32.444s +sys 1m5.972s + + +WordCount with combiner +real 31m38.491s +user 33m37.704s +sys 0m19.676s + +WordsSort +real 0m23.659s +user 0m25.428s +sys 0m1.880s + + + diff --git a/src/main/java/META-INF/MANIFEST.MF b/src/main/java/META-INF/MANIFEST.MF new file mode 100644 index 0000000..3a5a4aa --- /dev/null +++ b/src/main/java/META-INF/MANIFEST.MF @@ -0,0 +1,3 @@ +Manifest-Version: 1.0 +Main-Class: kubrin.SortWords + diff --git a/src/main/java/kubrin/GetRow.java b/src/main/java/kubrin/GetRow.java new file mode 100644 index 0000000..0a986de --- /dev/null +++ b/src/main/java/kubrin/GetRow.java @@ -0,0 +1,118 @@ +package kubrin; +// based on implementation of a-filippo (https://github.com/a-filippo/2017-big-data) + +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 GetRow 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 TextWithCountWritable(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 (TextWithCountWritable 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, "GetRow"); + + valueNumber = Integer.valueOf(args[2]); + + job.setJarByClass(GetRow.class); + + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.class); + + job.setMapperClass(MyMapper.class); + job.setReducerClass(MyReducer.class); + + job.setMapOutputKeyClass(IntWritable.class); + job.setMapOutputValueClass(TextWithCountWritable.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 GetRow(), args); + System.exit(returnCode); + + } +} \ No newline at end of file diff --git a/src/main/java/kubrin/NameCount.java b/src/main/java/kubrin/NameCount.java new file mode 100644 index 0000000..6707340 --- /dev/null +++ b/src/main/java/kubrin/NameCount.java @@ -0,0 +1,100 @@ +package kubrin; +// based on implementation of a-filippo (https://github.com/a-filippo/2017-big-data) + +import java.io.IOException; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +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 NameCount 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('\t'); + + String inputString = line.substring(0, pos); + int inputCount = Integer.valueOf(line.substring(pos+1)); + + context.write(new Text(inputString.toLowerCase()), new TextWithCountWritable(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 TextWithCountWritable 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(NameCount.class); + + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.class); + + job.setMapperClass(MyMapper.class); + job.setReducerClass(MyReducer.class); + + job.setMapOutputKeyClass(Text.class); + job.setMapOutputValueClass(TextWithCountWritable.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 NameCount(), args); + System.exit(returnCode); + + } +} \ No newline at end of file diff --git a/src/main/java/kubrin/SortWords.java b/src/main/java/kubrin/SortWords.java new file mode 100644 index 0000000..5630099 --- /dev/null +++ b/src/main/java/kubrin/SortWords.java @@ -0,0 +1,72 @@ +package kubrin; + +import java.io.IOException; + +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 SortWords extends Configured implements Tool { + public static class MyMapper extends Mapper { + 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(); + word.set(line.split("\t")[0]); + int val = -1*Integer.parseInt(line.split("\t")[1]); + context.write(new IntWritable(val), word); + } + } + + 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); + } + } + } + + @Override + public int run(String[] args) throws Exception { + final Configuration conf = this.getConf(); + final Job job = new Job(conf, "Words Order"); + job.setJarByClass(SortWords.class); + + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.class); + + job.setMapperClass(MyMapper.class); + + job.setMapOutputKeyClass(IntWritable.class); + job.setMapOutputValueClass(Text.class); + + job.setOutputKeyClass(IntWritable.class); + job.setOutputValueClass(Text.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 SortWords(), args); + System.exit(returnCode); + } +} diff --git a/src/main/java/kubrin/StopWordsCount.java b/src/main/java/kubrin/StopWordsCount.java new file mode 100644 index 0000000..a806d28 --- /dev/null +++ b/src/main/java/kubrin/StopWordsCount.java @@ -0,0 +1,119 @@ +package kubrin; + + +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 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.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 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{ + private static final IntWritable ONE = new IntWritable(1); + private final transient Text word = new Text(); + @Override + protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { + final String line = value.toString(); + final StringTokenizer tokenizer = new StringTokenizer(line, " \n\t.,![]()-:;"); + 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 { + 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.setMapOutputKeyClass(Text.class); + job.setMapOutputValueClass(IntWritable.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 stopWordsCount = job.getCounters().findCounter(MY_COUNTERS.STOP_WORDS_COUNT).getValue(); + long wordsCount = job.getCounters().findCounter(MY_COUNTERS.WORDS_COUNT).getValue(); + double result = stopWordsCount / (double)wordsCount * 100; + + System.out.println(String.valueOf("Stop words percent in input files = " + result)); + + 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); + } +} \ No newline at end of file diff --git a/src/main/java/kubrin/TextWithCountWritable.java b/src/main/java/kubrin/TextWithCountWritable.java new file mode 100644 index 0000000..16bff0b --- /dev/null +++ b/src/main/java/kubrin/TextWithCountWritable.java @@ -0,0 +1,75 @@ +package kubrin; +// based on implementation of a-filippo (https://github.com/a-filippo/2017-big-data) + +import java.io.DataInput; +import java.io.DataOutput; +import java.io.IOException; + +import org.apache.hadoop.io.WritableComparable; + +public class TextWithCountWritable 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; + } + + TextWithCountWritable(){} + + TextWithCountWritable(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 TextWithCountWritable)){ + return false; + } + TextWithCountWritable otherMyClass = (TextWithCountWritable)other; + if (otherMyClass.count != count){ + return false; + } + if (!otherMyClass.text.equals(text)){ + return false; + } + return true; + } + + @Override + protected TextWithCountWritable clone() { + return new TextWithCountWritable(text, count); + } + + @Override + public int compareTo(TextWithCountWritable o) { + if (equals(o)){ + return 0; + } + int intCompare = Integer.compare(count, o.count); + return (intCompare == 0) ? Integer.compare(this.hashCode(), o.hashCode()) : -intCompare; + } +} diff --git a/src/main/java/pritykovskaya/WordCount.java b/src/main/java/kubrin/WordCount.java similarity index 70% rename from src/main/java/pritykovskaya/WordCount.java rename to src/main/java/kubrin/WordCount.java index b047511..8f459a6 100644 --- a/src/main/java/pritykovskaya/WordCount.java +++ b/src/main/java/kubrin/WordCount.java @@ -1,5 +1,4 @@ -package pritykovskaya; - +package kubrin; import java.io.IOException; import java.util.StringTokenizer; @@ -20,7 +19,6 @@ 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 { @@ -28,38 +26,42 @@ public static class MyMapper extends Mapper= 'a' && curToken.charAt(0) <= 'z') || + (curToken.charAt(0) >= 'A' && curToken.charAt(0) <= 'Z')) { + word.set(curToken); + context.write(word, ONE); + } } } } - - public static class MyReducer extends Reducer { + static class MyReducer extends Reducer{ @Override - public void reduce(final Text key, final Iterable values, final Context context) - throws IOException, InterruptedException { + protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { int sum = 0; - for (final IntWritable val : values) { + for (final IntWritable val : values){ sum += val.get(); } context.write(key, new IntWritable(sum)); } } - - @Override public int run(final String[] args) throws Exception { + @Override + public int run(String[] args) throws Exception { final Configuration conf = this.getConf(); - final Job job = Job.getInstance(conf, "Word Count"); + 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); @@ -73,8 +75,8 @@ public void reduce(final Text key, final Iterable values, final Con return job.waitForCompletion(true) ? 0 : 1; } - public static void main(final String[] args) throws Exception { + public static void main(String[] args) throws Exception{ final int returnCode = ToolRunner.run(new Configuration(), new WordCount(), args); System.exit(returnCode); } -} +} \ No newline at end of file