Flink入門（五）——DataSet Api編程指南

Apache Flink

Apache Flink 是一個兼顧高吞吐、低延遲、高性能的分散式處理框架。在實時計算崛起的今天，Flink正在飛速發展。由於性能的優勢和兼顧批處理，流處理的特性，Flink可能正在顛覆整個大數據的生態。

DataSet API

首先要想運行Flink，我們需要下載並解壓Flink的二進位包，下載地址如下：https://flink.apache.org/downloads.html

我們可以選擇Flink與Scala結合版本，這裡我們選擇最新的1.9版本Apache Flink 1.9.0 for Scala 2.12進行下載。

下載成功後，在windows系統中可以通過Windows的bat文件或者Cygwin來運行Flink。

在linux系統中分為單機，集群和Hadoop等多種情況。

請參考：Flink入門（三）——環境與部署

Flink的編程模型，Flink提供了不同的抽象級別以開發流式或者批處理應用，本文我們來介紹DataSet API ，Flink最常用的批處理編程模型。

Flink中的DataSet程式是實現數據集轉換的常規程式（例如，Filter，映射，連接，分組）。數據集最初是從某些來源創建的（例如，通過讀取文件或從本地集合創建）。結果通過接收器返回，接收器可以例如將數據寫入（分散式）文件或標準輸出（例如命令行終端）。Flink程式可以在各種環境中運行，獨立運行或嵌入其他程式中。執行可以在本地JVM中執行，也可以在許多電腦的集群上執行。

示常式序

以下程式是WordCount的完整工作示例。您可以複製並粘貼代碼以在本地運行它。

Java

public class WordCountExample {
    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?");

        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));
            }
        }
    }
}

Scala

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

object WordCount {
  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?")

    val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
      .map { (_, 1) }
      .groupBy(0)
      .sum(1)

    counts.print()
  }
}

數據集轉換

數據轉換將一個或多個DataSet轉換為新的DataSet。程式可以將多個轉換組合到複雜的程式集中。

DataSet API 中最重要的就是這些運算元，我們將數據接入後，通過這些運算元對數據進行處理，得到我們想要的結果。

Java版運算元如下：

數據源

數據源創建初始數據集，例如來自文件或Java集合。創建數據集的一般機制是在InputFormat後面抽象的 。Flink附帶了幾種內置格式，可以從通用文件格式創建數據集。他們中的許多人在ExecutionEnvironment上都有快捷方法。

基於文件的：

• readTextFile(path)/ TextInputFormat- 按行讀取文件並將其作為字元串返回。
• readTextFileWithValue(path)/ TextValueInputFormat- 按行讀取文件並將它們作為StringValues返回。StringValues是可變字元串。
• readCsvFile(path)/ CsvInputFormat- 解析逗號（或其他字元）分隔欄位的文件。返回元組或POJO的DataSet。支持基本java類型及其Value對應作為欄位類型。
• readFileOfPrimitives(path, Class)/ PrimitiveInputFormat- 解析新行（或其他字元序列）分隔的原始數據類型（如String或）的文件Integer。
• readFileOfPrimitives(path, delimiter, Class)/ PrimitiveInputFormat- 解析新行（或其他字元序列）分隔的原始數據類型的文件，例如String或Integer使用給定的分隔符。
• readSequenceFile(Key, Value, path)/ SequenceFileInputFormat- 創建一個JobConf並從類型為SequenceFileInputFormat，Key class和Value類的指定路徑中讀取文件，並將它們作為Tuple2 <Key，Value>返回。

基於集合：

• fromCollection(Collection) - 從Java Java.util.Collection創建數據集。集合中的所有數據元必須屬於同一類型。
• fromCollection(Iterator, Class) - 從迭代器創建數據集。該類指定迭代器返回的數據元的數據類型。
• fromElements(T ...) - 根據給定的對象序列創建數據集。所有對象必須屬於同一類型。
• fromParallelCollection(SplittableIterator, Class) - 並行地從迭代器創建數據集。該類指定迭代器返回的數據元的數據類型。
• generateSequence(from, to) - 並行生成給定間隔中的數字序列。

通用：

• readFile(inputFormat, path)/ FileInputFormat- 接受文件輸入格式。
• createInput(inputFormat)/ InputFormat- 接受通用輸入格式。

例子

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

// read text file from local files system
DataSet<String> localLines = env.readTextFile("file:///path/to/my/textfile");

// read text file from a HDFS running at nnHost:nnPort
DataSet<String> hdfsLines = env.readTextFile("hdfs://nnHost:nnPort/path/to/my/textfile");

// read a CSV file with three fields
DataSet<Tuple3<Integer, String, Double>> csvInput = env.readCsvFile("hdfs:///the/CSV/file")
                           .types(Integer.class, String.class, Double.class);

// read a CSV file with five fields, taking only two of them
DataSet<Tuple2<String, Double>> csvInput = env.readCsvFile("hdfs:///the/CSV/file")
                               .includeFields("10010")  // take the first and the fourth field
                           .types(String.class, Double.class);

// read a CSV file with three fields into a POJO (Person.class) with corresponding fields
DataSet<Person>> csvInput = env.readCsvFile("hdfs:///the/CSV/file")
                         .pojoType(Person.class, "name", "age", "zipcode");

// read a file from the specified path of type SequenceFileInputFormat
DataSet<Tuple2<IntWritable, Text>> tuples =
 env.readSequenceFile(IntWritable.class, Text.class, "hdfs://nnHost:nnPort/path/to/file");

// creates a set from some given elements
DataSet<String> value = env.fromElements("Foo", "bar", "foobar", "fubar");

// generate a number sequence
DataSet<Long> numbers = env.generateSequence(1, 10000000);

// Read data from a relational database using the JDBC input format
DataSet<Tuple2<String, Integer> dbData =
    env.createInput(
      JDBCInputFormat.buildJDBCInputFormat()
                     .setDrivername("org.apache.derby.jdbc.EmbeddedDriver")
                     .setDBUrl("jdbc:derby:memory:persons")
                     .setQuery("select name, age from persons")
                     .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO))
                     .finish()
    );

// Note: Flink's program compiler needs to infer the data types of the data items which are returned
// by an InputFormat. If this information cannot be automatically inferred, it is necessary to
// manually provide the type information as shown in the examples above.

收集數據源和接收器

通過創建輸入文件和讀取輸出文件來完成分析程式的輸入並檢查其輸出是很麻煩的。Flink具有特殊的數據源和接收器，由Java集合支持以簡化測試。一旦程式經過測試，源和接收器可以很容易地被讀取/寫入外部數據存儲（如HDFS）的源和接收器替換。

在開發中，我們經常直接使用接收器對數據源進行接收。

final ExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment();

// Create a DataSet from a list of elements
DataSet<Integer> myInts = env.fromElements(1, 2, 3, 4, 5);

// Create a DataSet from any Java collection
List<Tuple2<String, Integer>> data = ...
DataSet<Tuple2<String, Integer>> myTuples = env.fromCollection(data);

// Create a DataSet from an Iterator
Iterator<Long> longIt = ...
DataSet<Long> myLongs = env.fromCollection(longIt, Long.class);

廣播變數

除了常規的 運算元操作輸入之外，廣播變數還允許您為 運算元操作的所有並行實例提供數據集。這對於輔助數據集或與數據相關的參數化非常有用。然後，運算元可以將數據集作為集合訪問。

// 1. The DataSet to be broadcast
DataSet<Integer> toBroadcast = env.fromElements(1, 2, 3);

DataSet<String> data = env.fromElements("a", "b");

data.map(new RichMapFunction<String, String>() {
    @Override
    public void open(Configuration parameters) throws Exception {
      // 3. Access the broadcast DataSet as a Collection
      Collection<Integer> broadcastSet = getRuntimeContext().getBroadcastVariable("broadcastSetName");
    }


    @Override
    public String map(String value) throws Exception {
        ...
    }
}).withBroadcastSet(toBroadcast, "broadcastSetName"); // 2. Broadcast the DataSet

分散式緩存

Flink提供了一個分散式緩存，類似於Apache Hadoop，可以在本地訪問用戶函數的並行實例。此函數可用於共用包含靜態外部數據的文件，如字典或機器學習的回歸模型。

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

// register a file from HDFS
env.registerCachedFile("hdfs:///path/to/your/file", "hdfsFile")

// register a local executable file (script, executable, ...)
env.registerCachedFile("file:///path/to/exec/file", "localExecFile", true)

// define your program and execute
...
DataSet<String> input = ...
DataSet<Integer> result = input.map(new MyMapper());
...
env.execute();

以上就是DataSet API 的使用，其實和spark非常的相似，我們將數據接入後，可以利用各種運算元對數據進行處理。

Flink Demo代碼

Flink系列文章：

Flink入門（一）——Apache Flink介紹
Flink入門（二）——Flink架構介紹

Flink入門（三）——環境與部署

Flink入門（四）——編程模型

更多實時計算,Flink,Kafka等相關技術博文，歡迎關註實時流式計算