Flink（三）Flink開發IDEA環境搭建與測試

一.IDEA開發環境

1.pom文件設置

    <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <encoding>UTF-8</encoding>
        <scala.version>2.11.12</scala.version>
        <scala.binary.version>2.11</scala.binary.version>
        <hadoop.version>2.7.6</hadoop.version>
        <flink.version>1.6.1</flink.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka-0.10_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.38</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.22</version>
        </dependency>
    </dependencies>
    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <testSourceDirectory>src/test/scala</testSourceDirectory>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.0</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                        <configuration>
                            <args>
                                <!-- <arg>-make:transitive</arg> -->
                                <arg>-dependencyfile</arg>
                                <arg>${project.build.directory}/.scala_dependencies</arg>
                            </args>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <version>2.18.1</version>
                <configuration>
                    <useFile>false</useFile>
                    <disableXmlReport>true</disableXmlReport>
                    <includes>
                        <include>**/*Test.*</include>
                        <include>**/*Suite.*</include>
                    </includes>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.0.0</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>org.apache.spark.WordCount</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

2.flink開發流程

Flink具有特殊類DataSet並DataStream在程序中表示數據。您可以將它們視為可以包含重復項的不可變數據集合。在DataSet數據有限的情況下，對於一個DataStream元素的數量可以是無界的。

這些集合在某些關鍵方面與常規Java集合不同。首先，它們是不可變的，這意味着一旦創建它們就無法添加或刪除元素。你也不能簡單地檢查里面的元素。

集合最初通過在弗林克程序添加源創建和新的集合從這些通過將它們使用API方法如衍生map，filter等等。

Flink程序看起來像是轉換數據集合的常規程序。每個程序包含相同的基本部分：

1.獲取execution environment,

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

2.加載/創建初始化數據

DataStream<String> text = env.readTextFile("file:///path/to/file");

3.指定此數據的轉換

val mapped = input.map { x => x.toInt }

4.指定放置計算結果的位置

writeAsText(String path)

print()

5.觸發程序執行

在local模式下執行程序

execute()

將程序達成jar運行在線上

./bin/flink run \

-m node21:8081 \

./examples/batch/WordCount.jar \

--input  hdfs:///user/admin/input/wc.txt \

--output  hdfs:///user/admin/output2  \

二. Wordcount案例

1.Scala代碼

package com.xyg.streaming

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
/**
  * Author: Mr.Deng
  * Date: 2018/10/15
  * Desc:
  */
object SocketWindowWordCountScala {
  def main(args: Array[String]) : Unit = {
    // 定義一個數據類型保存單詞出現的次數
    case class WordWithCount(word: String, count: Long)
    // port 表示需要連接的端口
    val port: Int = try {
      ParameterTool.fromArgs(args).getInt("port")
    } catch {
      case e: Exception => {
        System.err.println("No port specified. Please run 'SocketWindowWordCount --port <port>'")
        return
      }
    }
    // 獲取運行環境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 連接此socket獲取輸入數據
    val text = env.socketTextStream("node21", port, '\n')
    //需要加上這一行隱式轉換 否則在調用flatmap方法的時候會報錯
    import org.apache.flink.api.scala._
    // 解析數據, 分組, 窗口化, 並且聚合求SUM
    val windowCounts = text
      .flatMap { w => w.split("\\s") }
      .map { w => WordWithCount(w, 1) }
      .keyBy("word")
      .timeWindow(Time.seconds(5), Time.seconds(1))
      .sum("count")
    // 打印輸出並設置使用一個並行度
    windowCounts.print().setParallelism(1)
    env.execute("Socket Window WordCount")
  }
}

2.Java代碼

package com.xyg.streaming;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * Author: Mr.Deng
 * Date: 2018/10/15
 * Desc: 使用flink對指定窗口內的數據進行實時統計，最終把結果打印出來
 *       先在node21機器上執行nc -l 9000
 */
public class StreamingWindowWordCountJava {
    public static void main(String[] args) throws Exception {
    //定義socket的端口號
    int port;
    try{
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        port = parameterTool.getInt("port");
    }catch (Exception e){
        System.err.println("沒有指定port參數，使用默認值9000");
        port = 9000;
    }
    //獲取運行環境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    //連接socket獲取輸入的數據
    DataStreamSource<String> text = env.socketTextStream("node21", port, "\n");
    //計算數據
    DataStream<WordWithCount> windowCount = text.flatMap(new FlatMapFunction<String, WordWithCount>() {
        public void flatMap(String value, Collector<WordWithCount> out) throws Exception {
            String[] splits = value.split("\\s");
            for (String word:splits) {
                out.collect(new WordWithCount(word,1L));
            }
        }
    })//打平操作，把每行的單詞轉為<word,count>類型的數據
            //針對相同的word數據進行分組
            .keyBy("word")
            //指定計算數據的窗口大小和滑動窗口大小
            .timeWindow(Time.seconds(2),Time.seconds(1))
            .sum("count");
    //把數據打印到控制台,使用一個並行度
    windowCount.print().setParallelism(1);
    //注意：因為flink是懶加載的，所以必須調用execute方法，上面的代碼才會執行
    env.execute("streaming word count");
}

    /**
     * 主要為了存儲單詞以及單詞出現的次數
     */
    public static class WordWithCount{
        public String word;
        public long count;
        public WordWithCount(){}
        public WordWithCount(String word, long count) {
            this.word = word;
            this.count = count;
        }

        @Override
        public String toString() {
            return "WordWithCount{" +
                    "word='" + word + '\'' +
                    ", count=" + count +
                    '}';
        }
    }

}

3.運行測試

首先，使用nc命令啟動一個本地監聽，命令是：

[admin@node21 ~]$ nc -l 9000

通過netstat命令觀察9000端口。 netstat -anlp | grep 9000，啟動監聽如果報錯：-bash: nc: command not found，請先安裝nc，在線安裝命令：yum -y install nc。

然后，IDEA上運行flink官方案例程序

node21上輸入

IDEA控制台輸出如下

4.集群測試

這里單機測試官方案例

[admin@node21 flink-1.6.1]$ pwd
/opt/flink-1.6.1
[admin@node21 flink-1.6.1]$ ./bin/start-cluster.sh 
Starting cluster.
Starting standalonesession daemon on host node21.
Starting taskexecutor daemon on host node21.
[admin@node21 flink-1.6.1]$ jps
2100 StandaloneSessionClusterEntrypoint
2518 TaskManagerRunner
2584 Jps
[admin@node21 flink-1.6.1]$ ./bin/flink run examples/streaming/SocketWindowWordCount.jar --port 9000

程序連接到套接字並等待輸入。您可以檢查Web界面以驗證作業是否按預期運行：

單詞在5秒的時間窗口（處理時間，翻滾窗口）中計算並打印到stdout。監視TaskManager的輸出文件並寫入一些文本nc（輸入在點擊后逐行發送到Flink）：

三. 使用IDEA開發離線程序

Dataset是flink的常用程序，數據集通過source進行初始化，例如讀取文件或者序列化集合，然后通過transformation（filtering、mapping、joining、grouping）將數據集轉成，然后通過sink進行存儲，既可以寫入hdfs這種分布式文件系統，也可以打印控制台，flink可以有很多種運行方式，如local、flink集群、yarn等.

1. scala程序

package com.xyg.batch

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala._

/**
  * Author: Mr.Deng
  * Date: 2018/10/19
  * Desc:
  */
object WordCountScala{
  def main(args: Array[String]) {
    //初始化環境
    val env = ExecutionEnvironment.getExecutionEnvironment
    //從字符串中加載數據
    val text = env.fromElements(
      "Who's there?",
      "I think I hear them. Stand, ho! Who's there?")
    //分割字符串、匯總tuple、按照key進行分組、統計分組后word個數
    val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
      .map { (_, 1) }
      .groupBy(0)
      .sum(1)
    //打印
    counts.print()
  }
}

2. java程序

package com.xyg.batch;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * Author: Mr.Deng
 * Date: 2018/10/19
 * Desc:
 */
public class WordCountJava {
    public static void main(String[] args) throws Exception {
        //構建環境
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //通過字符串構建數據集
        DataSet<String> text = env.fromElements(
                "Who's there?",
                "I think I hear them. Stand, ho! Who's there?");
        //分割字符串、按照key進行分組、統計相同的key個數
        DataSet<Tuple2<String, Integer>> wordCounts = text
                .flatMap(new LineSplitter())
                .groupBy(0)
                .sum(1);
        //打印
        wordCounts.print();
    }
    //分割字符串的方法
    public static class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String line, Collector<Tuple2<String, Integer>> out) {
            for (String word : line.split(" ")) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }
}

3.運行

原文：https://www.cnblogs.com/frankdeng/p/9760015.html

--------------------------------------------------------------

開發環境准備

Flink 可以運行在 Linux, Max OS X, 或者是 Windows 上。為了開發 Flink 應用程序，在本地機器上需要有 Java 8.x 和 maven 環境。

如果有 Java 8 環境，運行下面的命令會輸出如下版本信息：

$ java -version
java version "1.8.0_65"
Java(TM) SE Runtime Environment (build 1.8.0_65-b17)
Java HotSpot(TM) 64-Bit Server VM (build 25.65-b01, mixed mode)

如果有 maven 環境，運行下面的命令會輸出如下版本信息：

$ mvn -version
Apache Maven 3.5.4 (1edded0938998edf8bf061f1ceb3cfdeccf443fe; 2018-06-18T02:33:14+08:00)
Maven home: /Users/wuchong/dev/maven
Java version: 1.8.0_65, vendor: Oracle Corporation, runtime: /Library/Java/JavaVirtualMachines/jdk1.8.0_65.jdk/Contents/Home/jre
Default locale: zh_CN, platform encoding: UTF-8
OS name: "mac os x", version: "10.13.6", arch: "x86_64", family: "mac"

另外我們推薦使用 ItelliJ IDEA (社區免費版已夠用)作為 Flink 應用程序的開發 IDE。Eclipse 雖然也可以，但是 Eclipse 在 Scala 和 Java 混合型項目下會有些已知問題，所以不太推薦 Eclipse。下一章節，我們會介紹如何創建一個 Flink 工程並將其導入 ItelliJ IDEA。

創建 Maven 項目

我們將使用 Flink Maven Archetype 來創建我們的項目結構和一些初始的默認依賴。在你的工作目錄下，運行如下命令來創建項目：

mvn archetype:generate \
    -DarchetypeGroupId=org.apache.flink \
    -DarchetypeArtifactId=flink-quickstart-java \
    -DarchetypeVersion=1.6.1 \
    -DgroupId=my-flink-project \
    -DartifactId=my-flink-project \
    -Dversion=0.1 \
    -Dpackage=myflink \
    -DinteractiveMode=false

 

mvn archetype:generate -DarchetypeGroupId=org.apache.flink -DarchetypeArtifactId=flink-quickstart-java -DarchetypeVersion=1.6.1 -DgroupId=my-flink-project -DartifactId=my-flink-project -Dversion=0.1 -Dpackage=myflink -DinteractiveMode=false

你可以編輯上面的 groupId, artifactId, package 成你喜歡的路徑。使用上面的參數，Maven 將自動為你創建如下所示的項目結構：

$ tree my-flink-project
my-flink-project
├── pom.xml
└── src
    └── main
        ├── java
        │   └── myflink
        │       ├── BatchJob.java
        │       └── StreamingJob.java
        └── resources
            └── log4j.properties

我們的 pom.xml 文件已經包含了所需的 Flink 依賴，並且在 src/main/java 下有幾個示例程序框架。接下來我們將開始編寫第一個 Flink 程序。

編寫 Flink 程序

啟動 IntelliJ IDEA，選擇 “Import Project”(導入項目)，選擇 my-flink-project 根目錄下的 pom.xml。根據引導，完成項目導入。

在 src/main/java/myflink 下創建 SocketWindowWordCount.java 文件：

package myflink;
public class SocketWindowWordCount {
  public static void main(String[] args) throws Exception {
  }
}

現在這程序還很基礎，我們會一步步往里面填代碼。注意下文中我們不會將 import 語句也寫出來，因為 IDE 會自動將他們添加上去。在本節末尾，我會將完整的代碼展示出來，如果你想跳過下面的步驟，可以直接將最后的完整代碼粘到編輯器中。

Flink 程序的第一步是創建一個 StreamExecutionEnvironment 。這是一個入口類，可以用來設置參數和創建數據源以及提交任務。所以讓我們把它添加到 main 函數中：

StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();

下一步我們將創建一個從本地端口號 9000 的 socket 中讀取數據的數據源：

DataStream text = env.socketTextStream("localhost", 9000, "\n");

這創建了一個字符串類型的 DataStream。DataStream 是 Flink 中做流處理的核心 API，上面定義了非常多常見的操作(如，過濾、轉換、聚合、窗口、關聯等)。在本示例中，我們感興趣的是每個單詞在特定時間窗口中出現的次數，比如說5秒窗口。為此，我們首先要將字符串數據解析成單詞和次數(使用Tuple2表示)，第一個字段是單詞，第二個字段是次數，次數初始值都設置成了1。我們實現了一個 flatmap 來做解析的工作，因為一行數據中可能有多個單詞。

DataStream<Tuple2<String, Integer>> wordCounts = text
        .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
          @Override
          public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            for (String word : value.split("\\s")) {
              out.collect(Tuple2.of(word, 1));
            }
          }
        });

接着我們將數據流按照單詞字段(即0號索引字段)做分組，這里可以簡單地使用 keyBy(int index) 方法，得到一個以單詞為 key 的Tuple2數據流。然后我們可以在流上指定想要的窗口，並根據窗口中的數據計算結果。在我們的例子中，我們想要每5秒聚合一次單詞數，每個窗口都是從零開始統計的。

DataStream<Tuple2<String, Integer>> windowCounts = wordCounts
        .keyBy(0)
        .timeWindow(Time.seconds(5))
        .sum(1);

第二個調用的 .timeWindow() 指定我們想要5秒的翻滾窗口(Tumble)。第三個調用為每個key每個窗口指定了sum聚合函數，在我們的例子中是按照次數字段(即1號索引字段)相加。得到的結果數據流，將每5秒輸出一次這5秒內每個單詞出現的次數。

最后一件事就是將數據流打印到控制台，並開始執行：

windowCounts.print().setParallelism(1);
env.execute("Socket Window WordCount");

最后的 env.execute 調用是啟動實際Flink作業所必需的。所有算子操作(例如創建源、聚合、打印)只是構建了內部算子操作的圖形。只有在execute()被調用時才會在提交到集群上或本地計算機上執行。

下面是完整的代碼，部分代碼經過簡化(代碼在 GitHub 上也能訪問到)：

package myflink;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
public class SocketWindowWordCount {
  public static void main(String[] args) throws Exception {
    // 創建 execution environment
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // 通過連接 socket 獲取輸入數據，這里連接到本地9000端口，如果9000端口已被占用，請換一個端口
    DataStream<String> text = env.socketTextStream("localhost", 9000, "\n");
    // 解析數據，按 word 分組，開窗，聚合
    DataStream<Tuple2<String, Integer>> windowCounts = text
        .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
          @Override
          public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            for (String word : value.split("\\s")) {
              out.collect(Tuple2.of(word, 1));
            }
          }
        })
        .keyBy(0)
        .timeWindow(Time.seconds(5))
        .sum(1);
    // 將結果打印到控制台，注意這里使用的是單線程打印，而非多線程
    windowCounts.print().setParallelism(1);
    env.execute("Socket Window WordCount");
  }
}

運行程序

要運行示例程序，首先我們在終端啟動 netcat 獲得輸入流：

nc -lk 9000

如果是 Windows 平台，可以通過 https://nmap.org/ncat/ 安裝 ncat 然后運行：

ncat -lk 9000

然后直接運行SocketWindowWordCount的 main 方法。

只需要在 netcat 控制台輸入單詞，就能在 SocketWindowWordCount 的輸出控制台看到每個單詞的詞頻統計。如果想看到大於1的計數，請在5秒內反復鍵入相同的單詞。