IEP/K2 Septembar 2021
Drugi kolokvijum u septembarskom roku 2021. godine održan je 17. septembra.
Postavka
Posmatra se evidencija o položenim ispitima. U jednom redu se nalaze identifikator studenta i lista ispitima[sic] koje je položio dati student. Svaki rad[sic] sadrži informacije o položenom ispitu kao što su šifra predmeta, šifra roka, i ocena koju je student dobio. Za potrebe navedene evidencije podaci se čuvaju u tekstualnoj datoteci na Hadoop sistemu. Podaci su dati u obliku:
<Student><TAB>{<Exam>{;<Exam>}}
Gde polje <Student> predstavlja identifikator studenta, a polje <Exam> sadrži šifru predmeta, nakon koga dolazi znak ,, pa šifra roka, nakon koga dolazi znak , i na kraju ocena.
- U programskom jeziku Java sastaviti Map/Reduce posao koji vraća statističke podatke o ispitima u ispitnim rokovima: šifru predmeta, šifru roka, broj studenata koji su polagali dati ispit, broj studenata koji su dobili ocenu 6, broj studenata koji su dobili ocenu 7, broj studenata koji su dobili ocenu 8, broj studenata koji su dobili ocenu 9, broj studenata koji su dobili ocenu 10. Voditi računa o konkurentnosti.
- U programskom jeziku Java sastaviti lanac od dva Map/Reduce posla koji vraća spisak predmeta sa najvišim prosekom (MAX) (prosek nije po roku nego od svih koji su ga ikada polagali), pri čemu je svaki od predmeta položilo barem studenata (, parametar koji se prosleđuje računarima koji rade obradu). Voditi računa o konkurentnosti.
Odgovor[sic] se predaju u vidu dva[sic] java datoteka (Ocene1.java i Ocene2.java).
Ocene1.java
package k2_sept_2021;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;
import java.io.Serializable;
import org.apache.log4j.LogManager;
import org.apache.log4j.Level;
import java.util.List;
import java.util.LinkedList;
public class Ocene1 {
static class Exam implements Serializable {
String subject;
String examTerm;
int grade;
public Exam(String examString) {
String[] data = examString.split(",");
this.subject = data[0];
this.examTerm = data[1];
this.grade = Integer.parseInt(data[2]);
}
public String getSubject() {
return subject;
}
public void setSubject(String subject) {
this.subject = subject;
}
public String getExamTerm() {
return examTerm;
}
public void setExamTerm(String examTerm) {
this.examTerm = examTerm;
}
public int getGrade() {
return grade;
}
public void setGrade(int grade) {
this.grade = grade;
}
@Override
public String toString() {
return "Exam [subject=" + subject + ", examTerm=" + examTerm + ", grade=" + grade + "]";
}
}
static class Student implements Serializable {
private String name;
private List<Exam> exams = new LinkedList<>();
public Student(String line) {
String[] data = line.split("\t");
this.name = data[0];
if (data.length <= 1)
return;
for (String examString : data[1].split(";")) {
exams.add(new Exam(examString));
}
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public List<Exam> getExams() {
return exams;
}
public void setExams(List<Exam> exams) {
this.exams = exams;
}
}
public static void main(String[] args) {
SparkConf conf = new SparkConf().setAppName("ocene1").setMaster("local");
LogManager.getRootLogger().setLevel(Level.WARN); // iskljuci nepotrebne poruke
try (JavaSparkContext sc = new JavaSparkContext(conf)) {
zadatak(sc, "resources/students_test.tsv");
}
}
public static void zadatak(JavaSparkContext sc, String filename) {
var result = sc.textFile(filename).map(Student::new).flatMapToPair(student -> {
System.out.println(student);
List<Tuple2<String, int[]>> gradeInfo = new LinkedList<Tuple2<String, int[]>>();
for (Exam exam : student.getExams()) {
int[] gradeArray = new int[6]; // {0,0,0,0,0,0} - Broj polaganja, broj 6, broj 7, broj 8 , broj 9, broj 10
String examName = exam.getSubject() + " - " + exam.getExamTerm();
if (exam.getGrade() <= 5)
continue;
gradeArray[0] = 1;
gradeArray[exam.getGrade() - 5]++;
gradeInfo.add(new Tuple2<>(examName, gradeArray));
}
return gradeInfo.iterator();
}).reduceByKey((gradeInfo1, gradeInfo2) -> {
int[] resultGrades = new int[6];
for (int i = 0; i < gradeInfo1.length; i++) {
resultGrades[i] = gradeInfo1[i] + gradeInfo2[i];
}
return resultGrades;
}).collectAsMap();
for (String examName : result.keySet()) {
int[] gi = result.get(examName);
System.out.println("Ispit: " + examName + " Br polaganja: "+ gi[0] + " Br 6:" + gi[1] + " Br 7:" + gi[2] + " Br 8:" + gi[3] + " Br 9:"
+ gi[4] + " Br 10:" + gi[5]);
}
}
}
Ocene2.java
package rs.etf.iep.mapreduce;
import java.io.File;
import java.util.LinkedList;
import java.util.List;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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;
public class Ocene2 {
public static final int DEFAULT_N = 1;
public static class Map1 extends Mapper<LongWritable, Text, Text, Text> {
@Override
public void map(LongWritable key, Text value, Context context) {
try {
String[] split = value.toString().split("\t");
if (split.length <= 1) {
// Студент нема ниједан испит.
return;
}
for (String exam : split[1].split(";")) {
String[] examSplit = exam.split(",");
context.write(new Text(examSplit[0]), new Text(examSplit[2] + "\t1"));
}
} catch (Exception e) {
// Хватају се све грешке јер Hadoop понекад може да их не испише.
e.printStackTrace();
}
}
}
public static class Reduce1 extends Reducer<Text, Text, Text, Text> {
@Override
public void reduce(Text key, Iterable<Text> values, Context context) {
try {
int sum = 0;
int count = 0;
for (Text value : values) {
String[] valueSplit = value.toString().split("\t");
sum += Integer.parseInt(valueSplit[0]);
count += Integer.parseInt(valueSplit[1]);
}
context.write(key, new Text(sum + "\t" + count));
} catch (Exception e) {
// Хватају се све грешке јер Hadoop понекад може да их не испише.
e.printStackTrace();
}
}
}
public static void job1(String[] args) throws Exception {
Job job = Job.getInstance();
job.setJarByClass(Ocene2.class);
job.setJobName("ocene2-1");
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
job.setMapperClass(Map1.class);
job.setReducerClass(Reduce1.class);
job.setCombinerClass(Reduce1.class);
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path("ocene2-temp"));
job.waitForCompletion(true);
}
public static class Map2 extends Mapper<LongWritable, Text, Text, Text> {
private static final Text text = new Text("ocene2-text");
private int N;
@Override
public void setup(Context context) {
N = context.getConfiguration().getInt("N", DEFAULT_N);
}
@Override
public void map(LongWritable key, Text value, Context context) {
try {
String[] splitValue = value.toString().split("\t");
double sum = Double.parseDouble(splitValue[1]);
double count = Double.parseDouble(splitValue[2]);
if (count >= N) {
context.write(text, new Text(splitValue[0] + "\t" + (sum / count)));
}
} catch (Exception e) {
// Хватају се све грешке јер Hadoop понекад може да их не испише.
e.printStackTrace();
}
}
}
public static class Reduce2 extends Reducer<Text, Text, Text, Text> {
@Override
public void reduce(Text key, Iterable<Text> values, Context context) {
try {
double maxAvg = 0;
List<String> list = new LinkedList<>();
for (Text value : values) {
String[] valueSplit = value.toString().split("\t");
double avg = Double.parseDouble(valueSplit[1]);
if (avg > maxAvg) {
list.clear();
maxAvg = avg;
}
if (avg == maxAvg) {
list.add(value.toString());
}
}
for (String value : list) {
context.write(key, new Text(value));
}
} catch (Exception e) {
// Хватају се све грешке јер Hadoop понекад може да их не испише.
e.printStackTrace();
}
}
}
public static void job2(String[] args) throws Exception {
Configuration conf = new Configuration();
conf.setInt("N", Integer.parseInt(args[2]));
Job job = Job.getInstance(conf, "ocene2-2");
job.setJarByClass(Ocene2.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
job.setMapperClass(Map2.class);
job.setReducerClass(Reduce2.class);
FileInputFormat.setInputPaths(job, new Path("ocene2-temp"));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.waitForCompletion(true);
}
public static void main(String[] args) throws Exception {
FileUtils.deleteDirectory(new File(args[1]));
FileUtils.deleteDirectory(new File("ocene2-temp"));
job1(args);
job2(args);
}
}
Provera
Sledeći sadržaj datoteke koja se prosleđuje kao prvi argument oba programa može se koristiti za testiranje:
Pera Peric predmet1,rok1,6;predmet2,rok1,10;predmet1,rok2,9 Marko Markovic predmet1,rok1,8;predmet2,rok1,6;predmet3,rok3,9
Na kolokvijumu su bile dostupne Students_V0.txt i Students_V1.txt datoteke za testiranje rešenja.