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 = https://www.cnblogs.com/tree1123/p/env.fromElements("Foo", "bar", "foobar", "fubar");

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

// Read data from a relational database using the JDBC input format
DataSet 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 = https://www.cnblogs.com/tree1123/p/...
DataSet> myTuples = env.fromCollection(data);

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

廣播變數

除了常規的 算子操作輸入之外，廣播變數還允許您為 算子操作的所有并行實體提供資料集，這對于輔助資料集或與資料相關的引數化非常有用，然后，算子可以將資料集作為集合訪問，

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

DataSet<String> data = https://www.cnblogs.com/tree1123/p/env.fromElements("a", "b");

data.map(new RichMapFunction() {
    @Override
    public void open(Configuration parameters) throws Exception {
      // 3. Access the broadcast DataSet as a Collection
      Collection 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等相關技術博文，歡迎關注實時流式計算

標籤：大數據

上一篇：HDFS的HA集群原理分析

下一篇：Hive簡介