Flinkとは

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

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

本家サイトはこちら。
どれ位ドキドキさせるかというと、

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

これは触っておかねば。

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

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

Simple is Simple。

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

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

実践

まずはFlink単体で試す。

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

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

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はこんな感じ

# !/bin/bash

set -e

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

tail -f /dev/null

こんな感じでbuild & 実行

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

結果。

まぁ、キレイ。

お決まりのwordcountを試す。

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

公式Exampleはこちら

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入れてビルドして再度実行。
ついでに酢豚入れる。

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慣れしていないと、何をどうビルドしたら良いのかさっぱりだろう。
さっぱりな人は、ここをザックリ読むといい。

entrypointに追加

 + $FLINK_HOME/bin/flink run /app/target/scala-2.11/example-assembly-0.1.jar --port 9000 --host localhost

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

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

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

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

コンテナの中に入ってhelloを一通り打つ。

結果

うん、WordCountだ。

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

reference