IEP/K2 2022

Izvor: SI Wiki
< ИЕП
Datum izmene: 21. april 2024. u 00:16; autor: Aleksa (razgovor | doprinosi) (ispravka da lepo filtrira.)
Pređi na navigaciju Pređi na pretragu

Drugi kolokvijum 2022. godine održan je 5. maja. Na kolokvijumu su bili dostupni Hadoop dokumentacija, prezentacija sa predavanja, virtuelna mašina korišćena na predavanju i dva tekstualna fajla kao primeri unosa (bez očekivanog ispisa ili primera R i N parametara).

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.

  1. 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, minimalnu ocenu, maksimalnu ocenu i prosečnu ocenu. Voditi računa o konkurentnosti.
  2. U programskom jeziku Java sastaviti lanac od dva Map/Reduce posla koji vraća predmet[1] koji je u zadatom ispitnom R polagalo najviše studenata, a da ni jedan od tih studenata u tom roku nije dobio zadatu ocenu N. Parametri R i N se prosleđuje[sic] 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 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:"+(r._2()[2]*1.0/r._2()[3]));
				
			}
			
		}
	}
	
}

Ocene2.java

package spark_iep;

import java.util.Arrays;
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 Ocene2 {
	
	//predmet zadat rok R, polagalo najvise studenata, a da nema ocene N

	public static void main(String[] args) {
		SparkConf conf = new SparkConf()
				.setAppName("Ocene1")
				.setMaster("local");
		try(JavaSparkContext sc = new JavaSparkContext(conf);){
			String rok = "jun2020";
			String zadataOcena = "10";
			JavaRDD<String> ulazniPodaci = sc.textFile("studenti_test.txt");
			//obrada ulaznih podataka
			//Tuple2<Integer,String[]>
			List<Tuple2<Integer,String[]>>  rezultat = ulazniPodaci.flatMapToPair(
					s->{
						List<Tuple2<String[], String>> 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
							//torka spremna za obradu i dodavanje u listu
							Integer ocena = Integer.parseInt(pod[2]);
							//([predmet,rok], ocena)
							Tuple2<String[], String> podatakZaListu = new Tuple2<>(new String[] {pod[0],pod[1]}, pod[2]);
							lista.add(podatakZaListu);
						}
						return lista.iterator();
					}
					)
					//isfiltriraj rok
			 		.filter(s->s._1[1].equals(rok))
			 		//reformatiraj kljuc predmet, vrednost ocena
			 		.mapToPair(s->new Tuple2<String, String>(s._1[0], s._2))
                    //ni jedna ocena u roku nije zadata ocena
			 		.filter(s->!(s._2.equals(zadataOcena)))
			 		//pravimoStringOcena ocena1,ocena2...
			 		.reduceByKey((a,b)->(a+";"+b))
			 		//reformatiraj da bude ([predmet1,ocena1;ocena2...],brojOcena)
			 		.mapToPair(s->new Tuple2<Integer, String[]>(s._2.split(";").length, new String[] {s._1, s._2}))
			 		//sortiraj po broju ocena
			 		.sortByKey(false)
			 		//vrati sve, ovako zbog lakse obrade ako nema
			 		.collect()
					;
			if(rezultat.size()==0) {
				System.out.println("Nema jbg");
			}else {
				System.out.println("Trazeni predmet je "+rezultat.get(0)._2[0]);
			}
			
			
		}
	}
	
}

Provera

Sledeći sadržaj datoteke koja se prosleđuje kao prvi argument oba programa može se koristiti za testiranje: 

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

(dodati tabulator na kraj poslednjeg reda ručno ukoliko se ne iskopira).

Na kolokvijumu su bile dostupne Students_V0.txt i Students_V1.txt datoteke za testiranje rešenja.

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

Napomene

  1. ↑ Ukoliko ih ima više, vratiti bilo koji. Nije garantovano da ovaj predmet postoji.
Preuzeto iz „https://siwiki.rs/w/index.php?title=ИЕП/К2_2022&oldid=7541”