From 621b5f51a350c99edbe7231d8f06760037c59b30 Mon Sep 17 00:00:00 2001 From: tFNiYaFF Date: Mon, 13 Nov 2017 05:25:07 +0300 Subject: [PATCH 1/2] First --- pom.xml | 8 +- src/main/java/tFNiYaFF/NameCount.java | 121 ++++++++++++++++++ .../WordCount.java | 62 ++++++++- 3 files changed, 183 insertions(+), 8 deletions(-) create mode 100644 src/main/java/tFNiYaFF/NameCount.java rename src/main/java/{pritykovskaya => tFNiYaFF}/WordCount.java (56%) diff --git a/pom.xml b/pom.xml index 16acaa5..d2b9764 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.WordCount @@ -38,8 +38,8 @@ org.apache.maven.plugins maven-compiler-plugin - 1.6 - 1.6 + 1.7 + 1.7 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/pritykovskaya/WordCount.java b/src/main/java/tFNiYaFF/WordCount.java similarity index 56% rename from src/main/java/pritykovskaya/WordCount.java rename to src/main/java/tFNiYaFF/WordCount.java index b047511..50cfc55 100644 --- a/src/main/java/pritykovskaya/WordCount.java +++ b/src/main/java/tFNiYaFF/WordCount.java @@ -1,8 +1,8 @@ -package pritykovskaya; +package tFNiYaFF; -import java.io.IOException; -import java.util.StringTokenizer; +import java.io.*; +import java.util.*; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; @@ -23,6 +23,32 @@ public class WordCount 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(); @@ -41,6 +67,8 @@ public static class MyMapper extends Mapper { + private Map countMap = new HashMap<>(); + @Override public void reduce(final Text key, final Iterable values, final Context context) throws IOException, InterruptedException { @@ -48,7 +76,33 @@ public void reduce(final Text key, final Iterable values, final Con for (final IntWritable val : values) { sum += val.get(); } - context.write(key, new IntWritable(sum)); + countMap.put(new Text(key),new IntWritable(sum)); + } + + @Override + protected void cleanup(Context context) throws IOException, InterruptedException { + File file = new File("stop_words_en.txt"); + InputStreamReader isr = new InputStreamReader(new FileInputStream(file)); + BufferedReader br = new BufferedReader(isr); + ArrayList stopWords = new ArrayList<>(); + String line; + while ((line = br.readLine()) != null) { + stopWords.add(line); + } + Map sortedMap = sortByValues(countMap); + int counter = 0; + int stopWordsCounter = 0; + int sum = 0; + for (Text key: sortedMap.keySet()) { + if(stopWords.contains(key.toString())){ + stopWordsCounter+=sortedMap.get(key).get(); + } + sum+= sortedMap.get(key).get(); + if(++counter==7) { + context.write(key, sortedMap.get(key)); + } + } + context.write(new Text( "stop% "), new IntWritable(stopWordsCounter/sum)); } } From f7261eacbb18d821f1a46a3fee60ded89b3e0b8f Mon Sep 17 00:00:00 2001 From: tFNiYaFF Date: Mon, 27 Nov 2017 15:10:57 +0300 Subject: [PATCH 2/2] third hw --- 2017-big-data.iml | 79 ++++++++++ answers.txt | 6 +- pom.xml | 2 +- report.txt | 14 ++ src/main/java/tFNiYaFF/Main.java | 105 +++++++++++++ src/main/java/tFNiYaFF/NValue.java | 117 +++++++++++++++ src/main/java/tFNiYaFF/NamesPercent.java | 99 ++++++++++++ src/main/java/tFNiYaFF/StopWordsCount.java | 139 +++++++++++++++++ .../java/tFNiYaFF/TextWithCountWriteble.java | 76 ++++++++++ src/main/java/tFNiYaFF/WordCount.java | 142 ++++++------------ src/main/java/tFNiYaFF/WordsOrder.java | 86 +++++++++++ 11 files changed, 762 insertions(+), 103 deletions(-) create mode 100644 2017-big-data.iml create mode 100644 report.txt create mode 100644 src/main/java/tFNiYaFF/Main.java create mode 100644 src/main/java/tFNiYaFF/NValue.java create mode 100644 src/main/java/tFNiYaFF/NamesPercent.java create mode 100644 src/main/java/tFNiYaFF/StopWordsCount.java create mode 100644 src/main/java/tFNiYaFF/TextWithCountWriteble.java create mode 100644 src/main/java/tFNiYaFF/WordsOrder.java 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 d2b9764..fc61ac2 100644 --- a/pom.xml +++ b/pom.xml @@ -29,7 +29,7 @@ - tFNiYaFF.WordCount + tFNiYaFF.Main 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/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/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 index 50cfc55..7f9f796 100644 --- a/src/main/java/tFNiYaFF/WordCount.java +++ b/src/main/java/tFNiYaFF/WordCount.java @@ -1,9 +1,5 @@ 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; @@ -20,115 +16,63 @@ 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 { - 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()); + context.write(word, ONE); + } } - } } - 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()); - context.write(word, ONE); - } - } - } - - - public static class MyReducer extends Reducer { - private Map countMap = new HashMap<>(); + 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(); - } - countMap.put(new Text(key),new IntWritable(sum)); - } - - @Override - protected void cleanup(Context context) throws IOException, InterruptedException { - File file = new File("stop_words_en.txt"); - InputStreamReader isr = new InputStreamReader(new FileInputStream(file)); - BufferedReader br = new BufferedReader(isr); - ArrayList stopWords = new ArrayList<>(); - String line; - while ((line = br.readLine()) != null) { - stopWords.add(line); - } - Map sortedMap = sortByValues(countMap); - int counter = 0; - int stopWordsCounter = 0; - int sum = 0; - for (Text key: sortedMap.keySet()) { - if(stopWords.contains(key.toString())){ - stopWordsCounter+=sortedMap.get(key).get(); - } - sum+= sortedMap.get(key).get(); - if(++counter==7) { - context.write(key, sortedMap.get(key)); + @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)); } - } - context.write(new Text( "stop% "), new IntWritable(stopWordsCounter/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); - @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.setMapperClass(MyMapper.class); + job.setReducerClass(MyReducer.class); + job.setCombinerClass(MyReducer.class); - job.setOutputKeyClass(Text.class); - job.setOutputValueClass(IntWritable.class); + job.setOutputKeyClass(Text.class); + job.setOutputValueClass(IntWritable.class); - job.setInputFormatClass(TextInputFormat.class); - job.setOutputFormatClass(TextOutputFormat.class); + job.setInputFormatClass(TextInputFormat.class); + job.setOutputFormatClass(TextOutputFormat.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(final String[] args) throws Exception { - final int returnCode = ToolRunner.run(new Configuration(), new WordCount(), args); - System.exit(returnCode); - } + 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); + } +}