ИЕП/К2 Јун 2022

Овај рок није решен. Помозите SI Wiki тако што ћете га решити.

Други колоквијум у јулском року 2022. године одржан је 17. септембра. Трајао је 90 минута и била је дозвољена употреба материјала са предавања.

Поставка

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

<Researcher><TAB><Work>{;<Work>}

Где поље <Researcher> представља идентификатор истраживача, а поље <Work> представља идентификатор рада, након кога долази знак : и листа идентификатора истраживача који су такође учествовали на том раду а који су раздвојени знаком ,. У листи идентификатора се сигурно налази идентификатор датог истраживача.

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

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

Istrazivaci1.java

package main;

import java.io.Serializable;
import java.util.Arrays;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

import org.apache.log4j.Level;
import org.apache.log4j.LogManager;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;

import main.Istrazivaci2.Work;
import scala.Tuple4;

public class Istrazivaci1 {

	static class Researcher implements Serializable {
		private String name;
		private List<Work> works = new LinkedList<>();

		public Researcher(String line) {
			String[] data = line.split("\t");
			this.name = data[0];

			if (data.length <= 1)
				return;

			for (String workString : data[1].split(";")) {
				this.works.add(new Work(workString));
			}

		}

		@Override
		public String toString() {
			return "Researcher [name=" + name + ", works=" + works + "]";
		}

		public String getName() {
			return name;
		}

		public void setName(String name) {
			this.name = name;
		}

		public List<Work> getWorks() {
			return works;
		}

		public void setWorks(List<Work> works) {
			this.works = works;
		}
	}

	static class Work implements Serializable {

		private String title;
		private List<String> coauthors = new LinkedList<>();

		public Work(String workString) {
			String[] data = workString.split(":");
			if (data.length <= 1)
				return;
			this.title = data[0];

			coauthors.addAll(Arrays.asList(data[1].split(",")));

		}

		public String getTitle() {
			return title;
		}

		public void setTitle(String title) {
			this.title = title;
		}

		public List<String> getCoauthors() {
			return coauthors;
		}

		public void setCoauthors(List<String> coauthors) {
			this.coauthors = coauthors;
		}

		@Override
		public String toString() {
			return "Work [title=" + title + ", coauthors=" + coauthors + "]";
		}

	}

	public static void main(String[] args) {
		SparkConf conf = new SparkConf().setAppName("istrazivaci1").setMaster("local");
		LogManager.getRootLogger().setLevel(Level.WARN); // iskljuci nepotrebne poruke
		try (JavaSparkContext sc = new JavaSparkContext(conf)) {
			zadatak(sc, "resources/Researchers_V1.txt", 3);
		}
	}

	public static void zadatak(JavaSparkContext sc, String filename, int N) {
		var result = sc.textFile(filename).map(line -> new Researcher(line)).filter(researcher -> researcher.works.size() >= N)
				.map(researcher -> {
					Set<String> uniqueResearchers = new HashSet<>();
					for (Work work : researcher.works)
						uniqueResearchers.addAll(work.getCoauthors());
					
					int cnt = uniqueResearchers.size() - 1; // ne ukljucuj i samog istrazivaca jer se nalazi u koautorima
					return new Tuple4<>(cnt, cnt, cnt, 1);
				})
				.reduce((s1, s2) -> {
					return new Tuple4<Integer, Integer, Integer, Integer>(Math.min(s1._1(), s2._1()),
							Math.max(s1._2(), s2._2()), s1._3() + s2._3(), s1._4() + s2._4());
				});
		
		int minCoauthors = result._1();
		int maxCoauthors = result._2();
		int sumCoauthors = result._3();
		int countCoauthors = result._4();
		System.out.println("Min: " + minCoauthors+", Max: " +maxCoauthors +", Avg: "+sumCoauthors *1.0/countCoauthors);
		
	}

}

Istrazivaci2.java

package spark_iep;

import java.util.Arrays;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

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

public class Istrazivaci2 {

	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("istrazivaci.txt");
			//obrada ulaznih podataka
			String R = "1";
			//primer clana liste - prvi ja autor, drugo skup svih sa kojima je saradjivao
			//(Zika, (Srbljan, Milicev, Zika, Zaki))
			JavaPairRDD<String, HashSet<String>> formatirani = ulazniPodaci.mapToPair(
					s->{
						String[] podaci = s.split("\t");
						if(podaci.length==1) return new Tuple2<>(podaci[0], new HashSet<String>());
						String[] radovi = podaci[1].split(";");
						HashSet<String> kolaboratori = new HashSet<>();
						for(String rad:radovi) {
							String[] razbijeni = rad.split(":");
							if(razbijeni.length==1) continue;
							String[] koautori = razbijeni[1].split(",");
							HashSet<String> setKoAutora = new HashSet<>(Arrays.asList(koautori));
							kolaboratori.addAll(setKoAutora);
						}
						return new Tuple2<>(podaci[0],kolaboratori);
					}
					);

			Tuple2<String, HashSet<String>>  sviSaKojimaJeIstrazivaoR = formatirani.filter(s->s._1.equals(R)).first();
			HashSet<String> kolaboratoriR = sviSaKojimaJeIstrazivaoR._2();
			
			
			JavaPairRDD<Integer, String> finalni = formatirani
			//da isfiltriramo sve koji su saradjivali sa R
			.filter(s->!(s._2.contains(R)))
			.mapToPair(s->{
				HashSet<String> datiPodaci = s._2;
				//menja datiPodaci tako da sadrzi samo one koje deli sa R
				datiPodaci.retainAll(kolaboratoriR);
				//sada imamo torku ciji je prvi clan broj zajednickih a drugi oznaka tog istrazivaca
				return new Tuple2<Integer, String>(datiPodaci.size(), s._1);
				
			})
			//sortiramo da bismo dobili one sa najvise, false da bi se sortiralno opadajuce
			.sortByKey(false)
			;
			
			try {
				//ispisuje sve sa MAX, moglo potencijalno i samo 1
				Tuple2<Integer, String> res = finalni.first();
				List<Tuple2<Integer, String>> finn = finalni.filter(s->s._1().equals(res._1())).collect();
				System.out.println("Max broj je  "+res._1()+" a saradnici su:");
				for(Tuple2<Integer, String> r:finn) {
					System.out.println("Istrazivac "+r._2());
				}
			}catch(Exception e) {
				System.out.println("Nema takvog za istrazivaca:"+R);
			}
			
			
			
			
			
		}
	}
	
}

Решење испод не би било вредновано са максималним бројем поена јер користи groupBy што је скупа операција због непотребног слања веће количине података преко мреже.

package main;

import java.io.Serializable;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;

import scala.Tuple2;

import org.apache.log4j.LogManager;
import org.apache.log4j.Level;

public class Istrazivaci2 {

	static class Researcher implements Serializable {
		private String name;
		private List<Work> works = new LinkedList<>();

		public Researcher(String line) {
			String[] data = line.split("\t");
			this.name = data[0];

			if (data.length <= 1)
				return;

			for (String workString : data[1].split(";")) {
				this.works.add(new Work(workString));
			}

		}

		@Override
		public String toString() {
			return "Researcher [name=" + name + ", works=" + works + "]";
		}

		public String getName() {
			return name;
		}

		public void setName(String name) {
			this.name = name;
		}

		public List<Work> getWorks() {
			return works;
		}

		public void setWorks(List<Work> works) {
			this.works = works;
		}
	}

	static class Work implements Serializable {

		private String title;
		private List<String> coauthors = new LinkedList<>();

		public Work(String workString) {
			String[] data = workString.split(":");
			if (data.length <= 1)
				return;
			this.title = data[0];

			coauthors.addAll(Arrays.asList(data[1].split(",")));

		}

		public String getTitle() {
			return title;
		}

		public void setTitle(String title) {
			this.title = title;
		}

		public List<String> getCoauthors() {
			return coauthors;
		}

		public void setCoauthors(List<String> coauthors) {
			this.coauthors = coauthors;
		}

		@Override
		public String toString() {
			return "Work [title=" + title + ", coauthors=" + coauthors + "]";
		}

	}

	static class TupleComparator
			implements Comparator<Tuple2<Integer, Iterable<Tuple2<String, Integer>>>>, Serializable {

		@Override
		public int compare(Tuple2<Integer, Iterable<Tuple2<String, Integer>>> o1,
				Tuple2<Integer, Iterable<Tuple2<String, Integer>>> o2) {

			return o1._1() - o2._1();
		}

	}

	public static void main(String[] args) {
		SparkConf conf = new SparkConf().setAppName("istrazivaci2").setMaster("local");
		LogManager.getRootLogger().setLevel(Level.WARN); // iskljuci nepotrebne poruke
		try (JavaSparkContext sc = new JavaSparkContext(conf)) {
			zadatak(sc, "resources/Researchers_V1.txt", "1");
		}
	}

	public static void zadatak(JavaSparkContext sc, String filename, String researcher) {
		var initialCoauthors = sc.textFile(filename).map(Researcher::new).filter(r -> r.getName().equals(researcher))
				.flatMap(r -> {
					Set<String> coauthorsSet = new HashSet<String>();

					for (Work work : r.getWorks()) {
						coauthorsSet.addAll(work.getCoauthors());
					}
					coauthorsSet.remove(r.getName());
					return coauthorsSet.iterator();
				});

		Set<String> initialCoauthorsSet = new HashSet<>(initialCoauthors.collect());

		var result = sc.textFile(filename).map(Researcher::new)
				// izbaci researchera i sve povezane sa njegovim "komsijama"
				.filter(r -> !r.getName().equals(researcher) && !initialCoauthorsSet.contains(r.getName())) 
				.map(r -> {
					Set<String> coauthorsSet = new HashSet<String>();

					for (Work work : r.getWorks()) {
						coauthorsSet.addAll(work.getCoauthors());
					}
					coauthorsSet.remove(r.getName());

					return new Tuple2<>(r.getName(), coauthorsSet);
				})
				// izbaci istrazivaca ako je povezan sa researcherom
				.filter(pair->!pair._2().contains(researcher)) 
				.map(pair -> {
					String researcherName = pair._1();
					Set<String> coauthorsSet = pair._2();

					Set<String> intersection = new HashSet(initialCoauthorsSet);
					intersection.retainAll(coauthorsSet);
					int numberConnections = intersection.size();

					return new Tuple2<>(researcherName, numberConnections);
				}).groupBy(pair -> pair._2()).max(new TupleComparator());

		System.out.println("Max number of connections: " + result._1());
		System.out.println("Authors with max number of connections: ");
		result._2().forEach(pair -> System.out.println(pair._1()));

	}
}

Провера

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