Study Flink Data Streaming Exercise Code

This post is based on [Flink exercise materials here](http://training.data-artisans.com/).

<!-- 
I am a very beginner of Scala and Flink. So I checked [Flink exercise materials here](http://training.data-artisans.com/) and worked on Data Streaming Exercises. There are detail instructions and you can also find souce code, so it is not difficult to just finish the materials, but on the other hand,  fully understanding its technical mechanism is confusing. I just wanted to know how flink works, so just following instruction was easy. But I haven't understood what is really going on inside of Flink yet and don't know how to use it in practice.
-->

## Environment
OS: Ubuntu 16.04
Kafka: kafka_2.10-0.10.2.0
Elasticsearch: elasticsearch-2.4.3
Kibana: kibana-4.6.4-linux-x86_64
Flink: flink-1.3.1-bin-hadoop27-scala_2.10 
Java: openjdk version "1.8.0_131"

```
ubuntu@ubuntu:~$ java -version
openjdk version "1.8.0_131"
OpenJDK Runtime Environment (build 1.8.0_131-8u131-b11-0ubuntu1.16.04.2-b11)
OpenJDK 64-Bit Server VM (build 25.131-b11, mixed mode)
```

IDE: IntelliJ
Build Tool: Apache Maven 3.5.0

```
buntu@ubuntu:~$ mvn --version
Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426; 2017-04-03T12:39:06-07:00)
```


## Generate a new Flink project
To create a new Flink Scala project, run the following command.

```
ubuntu@ubuntu:~/workspace$ mvn archetype:generate                             \
    -DarchetypeGroupId=org.apache.flink            \
    -DarchetypeArtifactId=flink-quickstart-scala   \
    -DarchetypeVersion=1.3.0                       \
    -DgroupId=org.apache.flink.quickstart          \
    -DartifactId=flink-scala-project               \
    -Dversion=0.1                                  \
    -Dpackage=org.apache.flink.quickstart          \
    -DinteractiveMode=false
```

After the installation, you can find a new project directory.

```
ubuntu@ubuntu:~/workspace$ ls
flink-scala-project
```

Next, clone "flink-training-exercises" project and install it

```
ubuntu@ubuntu:~/workspace$ git clone https://github.com/dataArtisans/flink-training-exercises.git
ubuntu@ubuntu:~/workspace$ cd flink-training-exercises
ubuntu@ubuntu:~/workspace/flink-training-exercises$ mvn clean install
```

After the installation, add the following dependencies  to "~/workspace/flink-scala-project/pom.xml" file.

```
ubuntu@ubuntu:~/workspace/flink-scala-project$ diff -u ~/workspace/backups/flink-scala-project_backup/pom.xml  ./pom.xml 
--- /home/ubuntu/workspace/backups/flink-scala-project_backup/pom.xml 2017-07-01 04:41:21.979384836 -0700
+++ ./pom.xml 2017-07-01 18:03:26.341580985 -0700
@@ -92,6 +92,11 @@
    <artifactId>flink-clients_2.10</artifactId>
    <version>${flink.version}</version>
   </dependency>
+  <dependency>
+   <groupId>com.data-artisans</groupId>
+   <artifactId>flink-training-exercises</artifactId>
+   <version>0.10.0</version>
+  </dependency>
```

Finally, run "mvn clean package" command under /flink-scala-project

```
ubuntu@ubuntu:~/workspace/flink-scala-project$ mvn clean package
```

#### *Memo: What is the differene between "mvn clean install" and "mvn clean package"*
When you use "mvn clean install", it put necessary packages in your local repository so that other projects can refer the local repository. 
In the above steps, "mvn clean install" is used for "training-exercising-project". It means, other projects can refer its package later. On the other hand, "mvn clean package" is used for  "quickstart-project". It means "quickstart-project" refers to the package of "training-exercising-project". In order to refer the local repository, maven dependency is added in pom.xml before "mvn clean package".

Reference: https://stackoverflow.com/questions/16602017/how-are-mvn-clean-package-and-mvn-clean-install-different


## Grab the overview of Scala programming

### Taxi Ride Cleansing
Start from this exercise [Taxi Ride Cleansing](http://dataartisans.github.io/flink-training/exercises/rideCleansing.html).
(You can find the source code [here](https://github.com/dataArtisans/flink-training-exercises/blob/master/src/main/scala/com/dataartisans/flinktraining/exercises/datastream_scala/basics/RideCleansing.scala))
>The task of the “Taxi Ride Cleansing” exercise is to cleanse a stream of TaxiRide events by removing events that do not start or end in New York City.

##### 1. Pass input data to "TaxiRideSource" which generates a stream of TaxiRide events

```
val rides = env.addSource(new TaxiRideSource(input, maxDelay, speed))
```

##### 2. Filter out ride events that do not start and end in NYC by using "GeoUtils.isInNYC" function can be called within a FilterFunction to check if a location is in the New York City area.

```
    val filteredRides = rides.filter(r => GeoUtils.isInNYC(r.startLon, r.startLat) && GeoUtils.isInNYC(r.endLon, r.endLat))
```

### Popular Places
Next exercise is [Popular Places](http://training.data-artisans.com/exercises/popularPlaces.html)
> The task of the “Popular Places” exercise is to identify popular places from the taxi ride data stream.

##### 1. Pass input data as like in "Taxi Ride Cleansing"

```
val rides = env.addSource(new TaxiRideSource(input, maxDelay, speed))
```

##### 2. After filtering out the events outside of NYC, map "Start events -> their departure location" and  "end events -> their destination location

```
    val popularPlaces = rides
      // remove all rides which are not within NYC
      .filter { r => GeoUtils.isInNYC(r.startLon, r.startLat) && GeoUtils.isInNYC(r.endLon, r.endLat) }
      // match ride to grid cell and event type (start or end)
      .map(new GridCellMatcher)
******omitted******      
```

Here, "GridCellMatcher" is defined as below.

```
  class GridCellMatcher extends MapFunction[TaxiRide, (Int, Boolean)] {
    def map(taxiRide: TaxiRide): (Int, Boolean) = {
      if (taxiRide.isStart) {
        // get grid cell id for start location
        val gridId: Int = GeoUtils.mapToGridCell(taxiRide.startLon, taxiRide.startLat)
        (gridId, true)
      } else {
        // get grid cell id for end location
        val gridId: Int = GeoUtils.mapToGridCell(taxiRide.endLon, taxiRide.endLat)
        (gridId, false)
      }
    }
  }
```

##### 3. Then, identify popular places which exceeds the threshod defined by time window of 15 minutes triggers every five minutes. 
> This is done by counting every five minutes the number of taxi rides that started and ended in the same area within the last 15 minutes. 

```
    val popularPlaces = rides
******omitted******          
      .timeWindow(Time.minutes(15), Time.minutes(5))
      // count events in window
      .apply{ (key: (Int, Boolean), window, vals, out: Collector[(Int, Long, Boolean, Int)]) =>
      out.collect( (key._1, window.getEnd, key._2, vals.size) )
    }
      // filter by popularity threshold
      .filter( c => { c._4 >= popThreshold } )
******omitted******      
```