# Laboratorium – MapReduce

Do wykonania ćwiczeń należy skopiować repozytorium:
```shell
git clone https://git.wmi.amu.edu.pl/bigdata/apache_hadoop
```
Celem ćwiczenia jest zaprezentowanie aplikacji w oparciu o algorytm MapReduce z wykorzystaniem:



* Hadoop Streaming
* Apache Hive
* Apache Pig
* Apache Spark

WordCount jest „odpowiednikiem Hello World” w świecie Big Data. Ćwiczenie prezentuje algorytm WordCount z wykorzystaniem różnych narzędzi.

Aby wykonać ćwiczenia, należy skopiować folder _books_ do systemu HDFS:
```
hdfs dfs -mkdir tmp
hdfs dfs -copyFromLocal ~/apache_hadoop/mr/books tmp/books
```
## 1.WordCount – Hadoop Streaming
Hadoop streaming umożliwia użytkownikom wykorzystanie mappera i reducera napisanego w dowolnym języku programowania. Jedynym wymaganiem jest obecność interpretera na każdym z węzłów.



### 1.1.Python


#### 1.1.1.Mapper i Reducer
Mapper i reducer napisane w języku Python znajdują się w folderze _~/apache_hadoop/mr/python_



#### 1.1.2.Uruchomienie algorytmu
Aplikację można uruchomić poprzez wykonanie komendy:
```
bash ~/apache_hadoop/mr/python/wordcount.sh
```
Uruchom aplikację i wyjaśnij co jest wynikiem działania tego algorytmu?



#### 1.1.3.Dane wyjściowe
Pliki zawierające wynik działania algorytmu znajdują się w folderze _tmp/python/output_ w systemie HDFS.

Analizując pliki w tym folderze, odpowiedz na pytanie ile reducerów zostało użytych podczas przebiegu aplikacji?



#### 1.1.4.Modyfikacja
Zmodyfikuj skrypt _~/apache_hadoop/mr/python/wordcount.sh_ tak, aby użyte zostały 4 reducery.



## 2.WordCount – Hive
Uruchom klienta Hive w konsoli Linux.



### 2.1.HiveQL


#### 2.1.1.Ustawienie parametrów
Ustaw parametr USERNAME tak, aby wskazywał nazwę użytkownika i utwórz schemat bazy:

```sql
set USERNAME=<nazwa_użytkownika>
create database if not exists ${hiveconf:USERNAME};
use ${hiveconf:USERNAME};
```

#### 2.1.2.Utworzenie tabeli doc
Utwórz tabelę przechowującą dokumenty tekstowe w postaci wierszowej:

```sql
create table ${hiveconf:USERNAME}.doc(
    text string
) row format delimited fields terminated by '\n'
stored as textfile;
```



#### 2.1.3.Ładowanie danych do tabeli
Utwórz zewnętrzną tabelę zawierającą dane z folderu _tmp/books_

```sql
create external table ${hiveconf:USERNAME}.doc_(
        text string
) row format delimited fields terminated by '\n'
  stored as textfile location '/user/${hiveconf:USERNAME}/tmp/books/';
```

Następnie należy skopiować dane z tabeli _doc__ do _doc_:

```sql
insert into ${hiveconf:USERNAME}.doc select * from ${hiveconf:USERNAME}.doc_;
```

#### 2.1.4.Utworzenie widoku
Kolejnym krokiem jest utworzenie widoku, zawierającego słowa wydzielone z krotek tabeli doc:

```sql
CREATE VIEW ${hiveconf:USERNAME}.words 
AS SELECT cast(word as string) as word FROM (
  SELECT explode(split(text, ' ')) AS word FROM ${hiveconf:USERNAME}.doc
) doc;
```


#### 2.1.5.Obliczenie ilości słów:
Należy wykonać polecenie:

```sql
select word, count(*) from ${hiveconf:USERNAME}.words group by word;
```


#### 2.1.6.Pominięcie widoku
Obliczenie ilości słów jest również możliwe z pominięciem utworzenia widoku, a z wykorzystaniem podzapytania:


    SELECT word, count(*) from (SELECT explode(split(text, ' ')) AS ${hiveconf:USERNAME}.word FROM doc) words group by word; 




## 3.WordCount – Pig
Uruchom Pig w konsoli.



### 3.1.Pig Latin


#### 3.1.11.Ładowanie pliku
Zdefiniuj zmienną books, ładującą tekst książek:

```
books = LOAD 'tmp/books' USING TextLoader() AS (line:chararray);
```


#### 3.1.12.Wydzielenie słów

Zdefiniuj zmienną words wydzielającą słowa z tekstu: 
```
words = FOREACH books GENERATE FLATTEN(TOKENIZE(line)) as word;
```


#### 3.1.13.Grupowanie

Pogrupój słowa: 
```
grouped = GROUP words BY word;
```

#### 3.1.14.Obliczanie ilości słów

Policz słowa: 
```
wordcount = FOREACH grouped GENERATE group, COUNT(words);
```


#### 3.1.15.Wyświetlenie rezultatów
Wyświetl wynik działania algorytmu:

```
    DUMP wordcount;
```



## 4.WordCount Spark


### 4.1.Scala
Uruchom spark-shell.



#### 4.1.1.Ładowanie pliku
Zdefiniuj zmienną ładującą plik z HDFS:

```
val text_file = sc.textFile("tmp/books/")
```

#### 4.1.2.Definicja funkcji MapReduce
Spark korzysta z zalet języka Scala i umożliwia utworzenie prostej funkcji:

```
val counts = text_file.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)
```


#### 4.1.3.Wyświetlenie wyników
Aby wyświetlić rezultaty wykonaj:

```
counts.collect().foreach(println) 
```



#### 4.1.4.Zapis rezultatu do pliku
Aby zapisać rezultat do pliku wykonaj:

```
counts.saveAsTextFile("tmp/spark-outputs")
```


#### 4.1.5.Zapis alternatywny
Przy tworzeniu bardziej skomplikowanych funkcji przydatny może okazać się zapis, w którym parametry wejściowe funkcji przyjmują określone nazwy lub chcemy czytać z lokalnego systemu plików/udziału NFS:

```
sc.textFile("file:///home/<nazwa_uzytkownika>/apache_hadoop/mr/books")
val counts2 = text_file.
    flatMap(txt => txt.split(" ")).
    map(word => (word, 1)).
    reduceByKey((a, b) => a + b)
counts2.collect().foreach(println) 
counts2.saveAsTextFile("tmp/spark-outputs")
```



### 5.1.PySpark
Uruchom pyspark.



#### 5.1.1.Ładowanie pliku
Zdefiniuj zmienną ładującą plik z HDFS:

```
    text_file = sc.textFile("hdfs:///tmp/books/*")
```


#### 5.1.2.Definicja funkcji MapReduce
Spark korzysta z zalet języka Scala i umożliwia utworzenie prostej funkcji:

```
    counts = text_file.flatMap(lambda line: line.split(" ")) \ \
             .map(lambda word: (word, 1)) \ \
             .reduceByKey(lambda a, b: a + b)
```


#### 5.1.3.Wyświetlenie wyników
Aby wyświetlić rezultaty wykonaj:

```
    counts.collect()
```


#### 5.1.4.Zapis rezultatu do pliku
Aby zapisać rezultat do pliku wykonaj:

```
    counts.saveAsTextFile("tmp/pyspark-outputs")
```

## 6. Modyfikacje (zaliczenie)

Czy potrafisz dokonać modyfikacji? Przykładowe pomysły:
- posortuj słowa według ich długości
- wyczyść nieznaczące znaki i sprowadź słowa do małych liter
- policz wystąpienia słów o określonej ilości znaków