ИЕП/К2 2022

Извор: SI Wiki
Пређи на навигацију Пређи на претрагу

Други колоквијум 2022. године одржан је 5. маја. На колоквијуму су били доступни Hadoop документација, презентација са предавања, виртуелна машина коришћена на предавању и два текстуална фајла као примери уноса (без очекиваног исписа или примера R и N параметара).

Поставка

Посматра се евиденција о положеним испитима. У једном реду се налазе идентификатор студента и листа испитима[sic] које је положио дати студент. Сваки рад[sic] садржи информације о положеном испиту као што су шифра предмета, шифра рока, и оцена коју је студент добио. За потребе наведене евиденције подаци се чувају у текстуалној датотеци на Hadoop систему. Подаци су дати у облику: 

<Student><TAB>{<Exam>{;<Exam>}}

Где поље <Student> представља идентификатор студента, а поље <Exam> садржи шифру предмета, након кога долази знак ,, па шифра рока, након кога долази знак , и на крају оцена.

  1. У програмском језику Јава саставити Map/Reduce посао који враћа статистичке податке о испитима у испитним роковима: шифру предмета, шифру рока, број студената који су полагали дати испит, минималну оцену, максималну оцену и просечну оцену. Водити рачуна о конкурентности.
  2. У програмском језику Јава саставити ланац од два Map/Reduce посла који враћа предмет[1] који је у задатом испитном R полагало највише студената, а да ни један од тих студената у том року није добио задату оцену N. Параметри R и N се прослеђује[sic] рачунарима који раде обраду. Водити рачуна о конкурентности.

Одговор[sic] се предају у виду два[sic] јава датотека (Ocene1.java и Ocene2.java).

Ocene1.java

package spark_iep;

import java.util.LinkedList;
import java.util.List;

import org.apache.spark.*;
import org.apache.spark.api.java.*;
import scala.Tuple2;
import scala.Tuple5;

public class Ocene1 {

	public static void main(String[] args) {
		SparkConf conf = new SparkConf()
				.setAppName("Ocene1")
				.setMaster("local");
		try(JavaSparkContext sc = new JavaSparkContext(conf);){
			JavaRDD<String> ulazniPodaci = sc.textFile("studenti_test.txt");
			//obrada ulaznih podataka
			List<Tuple2<String,Integer[]>> rezultat = ulazniPodaci.flatMapToPair(
					s->{
						List<Tuple2<String, Integer[]>> lista = new LinkedList<>();
						String[] podaciSvi = s.split("\t");
						//za slucaj da student nema polozene ispite
						if(podaciSvi.length==1) return lista.iterator();
						//niz podataka o ispitima za studenta
						String[] podaciIspiti = podaciSvi[1].split(";");
						for(String p:podaciIspiti) {
							//pod[0] = predmet1,rok1,6
							String[] pod = p.split(",");
							//konvertujemo ocenu u string radi dalje obrade
							Integer ocena = Integer.parseInt(pod[2]);
							//torka spremna za obradu i dodavanje u listu
							Tuple2<String, Integer[]> podatakZaListu = new Tuple2<>(pod[0]+"&"+pod[1], new Integer[] {ocena, ocena, ocena, 1});
							lista.add(podatakZaListu);
						}
						return lista.iterator();
					}
					)
					//prvi clan se koristi za max, drugi za min, treci je suma vrednosti, cetvrti brojac vrednosti
					.reduceByKey((a,b)->new Integer[] {Math.max(a[0], b[0]), Math.min(a[1], b[1]), a[2]+b[2], a[3]+b[3]}).collect();
			
			for(Tuple2<String, Integer[]> r:rezultat) {
				System.out.println("Predmet&Rok: "+r._1()+", max:"+r._2()[0]+", min:"+r._2()[1]+", avg:"+Math.round(r._2()[2]*1.0/r._2()[3]));
				
			}
			
		}
	}
	
}

Ocene2.java

Провера

Следећи садржај датотеке која се прослеђује као први аргумент оба програма може се користити за тестирање: 

pera	predmet1,jun2020,9;predmet2,jun2020,10;predmet3,jun2020,9;predmet1,jul2020,10;predmet3,jul2020,10
mika	predmet1,jun2020,6;predmet2,jun2020,6;predmet3,jun2020,7;predmet1,jul2020,6
zika	predmet1,jun2020,8
jovan

(додати табулатор на крај последњег реда ручно уколико се не ископира).

На колоквијуму су биле доступне Students_V0.txt и Students_V1.txt датотеке за тестирање решења.

Ocene1 & Ocene2 - Hadoop (старо градиво)

package rs.etf.iep.mapreduce;

import java.io.File;

import org.apache.commons.io.FileUtils;
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 Ocene1 {
	private static final int MIN_INVALID_GRADE = 3;
	private static final int MAX_INVALID_GRADE = 11;

	private static Text formatMapRow(int studentCount, int minGrade, int maxGrade, int gradeSum, double avgGrade) {
		return new Text(studentCount + "\t" + minGrade + "\t" + maxGrade + "\t" + gradeSum + "\t" + avgGrade);
	}

	public static class Map 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(",");
					String examCode = examSplit[0];
					String examName = examSplit[1];
					int grade = Integer.parseInt(examSplit[2]);
					Text emitKey = new Text(examCode + "\t" + examName);
					Text emitValue = formatMapRow(1, grade, grade, grade, grade);
					context.write(emitKey, emitValue);
				}
			} catch (Exception e) {
				// Хватају се све грешке јер Hadoop понекад може да их не испише.
				e.printStackTrace();
			}
		}
	}

	public static class Reduce extends Reducer<Text, Text, Text, Text> {
		@Override
		public void reduce(Text key, Iterable<Text> values, Context context) {
			try {
				int studentCount = 0;
				int gradeSum = 0;
				int minGrade = MAX_INVALID_GRADE;
				int maxGrade = MIN_INVALID_GRADE;
				for (Text value : values) {
					String[] valueSplit = value.toString().split("\t");
					studentCount += Integer.parseInt(valueSplit[0]);
					gradeSum += Integer.parseInt(valueSplit[3]);
					int currMinGrade = Integer.parseInt(valueSplit[1]);
					if (currMinGrade < minGrade) {
						minGrade = currMinGrade;
					}
					int currMaxGrade = Integer.parseInt(valueSplit[2]);
					if (currMaxGrade > maxGrade) {
						maxGrade = currMaxGrade;
					}
				}
				context.write(key, formatMapRow(studentCount, minGrade, maxGrade, gradeSum,
						((double) gradeSum) / ((double) studentCount)));
			} catch (Exception e) {
				// Хватају се све грешке јер Hadoop понекад може да их не испише.
				e.printStackTrace();
			}
		}
	}

	public static void main(String[] args) throws Exception {
		FileUtils.deleteDirectory(new File(args[1]));

		Job job = Job.getInstance();
		job.setJarByClass(Ocene1.class);
		job.setJobName("ocene1");

		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

		job.setMapperClass(Map.class);
		job.setReducerClass(Reduce.class);
		job.setCombinerClass(Reduce.class);

		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		job.waitForCompletion(true);
	}
}



package rs.etf.iep.mapreduce;

import java.io.File;

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 {
	private static final int MIN_INVALID_GRADE = 3;

	public static class Map1 extends Mapper<LongWritable, Text, Text, Text> {
		private String R;

		@Override
		public void setup(Context context) {
			R = context.getConfiguration().get("R");
		}

		@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(",");
					if (examSplit[1].equals(R)) {
						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> {
		private int N;

		@Override
		public void setup(Context context) {
			N = context.getConfiguration().getInt("N", MIN_INVALID_GRADE);
		}

		@Override
		public void reduce(Text key, Iterable<Text> values, Context context) {
			try {
				int studentCount = 0;
				int grade = MIN_INVALID_GRADE;
				for (Text value : values) {
					String[] valueSplit = value.toString().split("\t");
					grade = Integer.parseInt(valueSplit[0]);
					if (grade == N) {
						return;
					}
					studentCount += Integer.parseInt(valueSplit[1]);
				}
				context.write(key, new Text(grade + "\t" + studentCount));
			} catch (Exception e) {
				// Хватају се све грешке јер Hadoop понекад може да их не испише.
				e.printStackTrace();
			}
		}
	}

	public static void job1(String[] args) throws Exception {
		Configuration conf = new Configuration();
		conf.set("R", args[2]);
		conf.setInt("N", Integer.parseInt(args[3]));

		Job job = Job.getInstance(conf, "ocene2-1");
		job.setJarByClass(Ocene2.class);

		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");

		@Override
		public void map(LongWritable key, Text value, Context context) {
			try {
				context.write(text, value);
			} 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 {
				int maxStudentCount = 0;
				String subject = null;
				for (Text value : values) {
					String[] valueSplit = value.toString().split("\t");
					int studentCount = Integer.parseInt(valueSplit[2]);
					if (studentCount > maxStudentCount) {
						maxStudentCount = studentCount;
						subject = valueSplit[0];
					}
				}
				context.write(new Text(subject), new Text(String.valueOf(maxStudentCount)));
			} catch (Exception e) {
				// Хватају се све грешке јер Hadoop понекад може да их не испише.
				e.printStackTrace();
			}
		}
	}

	public static void job2(String[] args) throws Exception {
		Job job = Job.getInstance();
		job.setJobName("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.setCombinerClass(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);
	}
}

Напомене

  1. ↑ Уколико их има више, вратити било који. Није гарантовано да овај предмет постоји.
Преузето из „https://siwiki.rs/w/index.php?title=ИЕП/К2_2022&oldid=7520”