俺の中で話題のFlinkを試してみる。

# Flinkとは
Spark使いをドキドキさせる分散処理ソフト。Sparkと同じapache仲間。

知っている方からすると今更感が強いだろうが、
最近知ったのだからしょうがない。

本家サイトは[こちら](https://flink.apache.org/)。
どれ位ドキドキさせるかというと、

![flink_01.png](https://qiita-image-store.s3.amazonaws.com/0/42445/64b3e88c-df03-b001-d860-5bb20a5264d9.png)

既に"spark vs"でググると3位にランクインしている。

これは触っておかねば。

# 最終的に何が出来上がるか

``kafkaで何かしらを飛ばして、elasticsearchに何かしらを投げるStack。``

Simple is Simple。

stackはdocker-composeで作る予定。
イメージはDockerfileから作成する。
docker hubには本家libraryもあるのだが、癖を知りたい(何のかは聞かないでほしい)。

細かい所は良い感じで進める。

# 実践

## まずはFlink単体で試す。


単体で試すというのはLocal Flink Clusterを試すって事ね。

こんな感じでDockerイメージを作って

``` Dockerfile
FROM openjdk:8-jdk

ENV FLINK_VERSION=1.3.0 \
  FLINK_SCALA_VERSION=2.11 \
  FLINK_HADOOP_VERSION=27 \
  FLINK_HOME=/opt/flink
RUN apt-get update \
  && apt-get -y upgrade \
  && apt-get -y install \
       apt-transport-https \
       curl \
  && curl -OL http://ftp.tsukuba.wide.ad.jp/software/apache/flink/flink-${FLINK_VERSION}/flink-${FLINK_VERSION}-bin-hadoop${FLINK_HADOOP_VERSION}-scala_${FLINK_SCALA_VERSION}.tgz \
  && tar fx flink-${FLINK_VERSION}-bin-hadoop${FLINK_HADOOP_VERSION}-scala_${FLINK_SCALA_VERSION}.tgz \
  && rm flink-${FLINK_VERSION}-bin-hadoop${FLINK_HADOOP_VERSION}-scala_${FLINK_SCALA_VERSION}.tgz \
  && mv flink-${FLINK_VERSION} ${FLINK_HOME}

COPY ./docker-entrypoint.sh /docker-entrypoint.sh

ENTRYPOINT /docker-entrypoint.sh
```

entrypointはこんな感じ

``` bash
#!/bin/bash

set -e

${FLINK_HOME}/bin/start-local.sh

tail -f /dev/null
```

こんな感じでbuild & 実行

``` bash
sudo docker build -t miyatama/flink:1.13.0 .
sudo docker run -it -p 8081:8081 miyatama/flink:1.13.0
```

結果。
![flink_02.png](https://qiita-image-store.s3.amazonaws.com/0/42445/7674ba51-18e7-d134-ce3e-8673b58f0e9f.png)

まぁ、キレイ。

## お決まりのwordcountを試す。

SparkにしてもFlinkにしても分散処理はWordCountが大好きだ。
何故かは知らんが、大好きだ。

公式Exampleは[こちら](https://github.com/apache/flink/blob/master/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/socket/SocketWindowWordCount.scala)

QuickStartの中には

>You can find the complete source code for this SocketWindowWordCount example in scala and java on GitHub.

と書いてある。探したら2つあった。
何がどう違うのか見てもいないけど、~~QuickStartの方を使う。~~exampleの方を使う。
使うと言っても、flink-projectをcloneしてsrc差し替えてビルドするだけ。

先ほどのイメージにmaven入れてビルドして再度実行。
ついでに酢豚入れる。

``` dockerfile
FROM openjdk:8-jdk

ENV FLINK_VERSION=1.3.0 \
  FLINK_SCALA_VERSION=2.11 \
  FLINK_HADOOP_VERSION=27 \
  FLINK_HOME=/opt/flink \
  SBT_VERSION=0.13.15 \
  MAVEN_VERSION=3.5.0

RUN  apt-get update \
  && apt-get -y upgrade \
  && apt-get -y install \
       apt-transport-https \
       curl \
       git \
  && echo "deb https://dl.bintray.com/sbt/debian /" | tee -a /etc/apt/sources.list.d/sbt.list \
  && apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv 2EE0EA64E40A89B84B2DF73499E82A75642AC823 \
  && apt-get update \
  && apt-get -y install \
       sbt=${SBT_VERSION} \
  && curl -OL http://ftp.tsukuba.wide.ad.jp/software/apache/flink/flink-${FLINK_VERSION}/flink-${FLINK_VERSION}-bin-hadoop${FLINK_HADOOP_VERSION}-scala_${FLINK_SCALA_VERSION}.tgz \
  && tar fx flink-${FLINK_VERSION}-bin-hadoop${FLINK_HADOOP_VERSION}-scala_${FLINK_SCALA_VERSION}.tgz \
  && rm flink-${FLINK_VERSION}-bin-hadoop${FLINK_HADOOP_VERSION}-scala_${FLINK_SCALA_VERSION}.tgz \
  && mv flink-${FLINK_VERSION} ${FLINK_HOME}  \
  && curl -OL http://ftp.yz.yamagata-u.ac.jp/pub/network/apache/maven/maven-3/${MAVEN_VERSION}/binaries/apache-maven-${MAVEN_VERSION}-bin.tar.gz \
  && tar fx apache-maven-${MAVEN_VERSION}-bin.tar.gz \
  && rm -f apache-maven-${MAVEN_VERSION}-bin.tar.gz \
  && mv apache-maven-${MAVEN_VERSION} /opt/apache-maven-${MAVEN_VERSION}

ENV PATH=/opt/apache-maven-${MAVEN_VERSION}/bin:${PATH}

RUN mkdir /src \
  && cd /src \
  && git clone https://github.com/apache/flink.git \
  && cd ./flink/flink-quickstart/flink-quickstart-scala \
  && mvn package \
  && cd / \
  && git clone https://github.com/tillrohrmann/flink-project.git \
  && git clone https://github.com/apache/flink.git \
  && mv flink-project app \
  && rm -fr /app/src/main/scala/org/example/* \
  && \cp -f /flink/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/socket/SocketWindowWordCount.scala /app/src/main/scala/org/example/SocketWindowWordCount.scala \
  && rm -fr /flink \
  && cd /app \
  && sed -i -e 's/"Flink Project"/"example"/g' build.sbt \
  && sed -i -e 's/0.1-SNAPSHOT/0.1/g' build.sbt \
  && sed -i -e 's/org.example.Job/org.example.SocketWindowWordCount/g' build.sbt \
  && sed -i -e 's/org.apache.flink.streaming.scala.examples.socket/org.example/g' /app/src/main/scala/org/example/SocketWindowWordCount.scala \
  && sbt assembly

COPY ./docker-entrypoint.sh /docker-entrypoint.sh

ENTRYPOINT /docker-entrypoint.sh
```

酢豚のビルドは長いのでハンドスピナーとかコーヒーとかで時間を潰すといい

scala慣れしていないと、何をどうビルドしたら良いのかさっぱりだろう。
さっぱりな人は、[ここ](https://ci.apache.org/projects/flink/flink-docs-release-1.3/quickstart/scala_api_quickstart.html)をザックリ読むといい。

entrypointに追加
``` bash
 + $FLINK_HOME/bin/flink run /app/target/scala-2.11/example-assembly-0.1.jar --port 9000 --host localhost
```

centos7の人はポートを開けておく。firewalldを切っている人は特に問題ないはず。

``` bash
sudo firewall-cmd --permanent --zone=pulic --add-port=9000/tcp
sudo firewall-cmd --reload
```

実行して確認(ビルドは割愛)

``` bash
sudo docker run -it -p 8081:8081 -p 9000:9000 miyatama/flink:1.13.0
```

コンテナの中に入ってhelloを一通り打つ。
![flink03.png](https://qiita-image-store.s3.amazonaws.com/0/42445/5812a6ef-0b50-d934-b25c-694e6247bd92.png)

結果
![flink04.png](https://qiita-image-store.s3.amazonaws.com/0/42445/c3a6d601-caa7-2d98-918f-9da219f3aa2c.png)


うん、WordCountだ。

kafkaのkすら登場してないけど、
長くなったので一旦切る。

# reference

 + [Flink vs. Spark](https://www.slideshare.net/sbaltagi/flink-vs-spark)
 + [Flink](https://flink.apache.org/)
 + [sbt Reference](http://www.scala-sbt.org/0.13/docs/Setup.html)
 + [maven installation](https://maven.apache.org/install.html)