Flink DataStream API-有解無憂

1. API基本概念

Flink程式可以對分布式集合進行轉換（例如： filtering, mapping, updating state, joining, grouping, defining windows, aggregating）

集合最初是從源創建的（例如，從檔案、kafka主題或本地記憶體集合中讀取）

結果通過sink回傳，例如，可以將資料寫入(分布式)檔案，或者寫入標準輸出(例如，命令列終端)

根據資料源的型別（有界或無界資料源），可以撰寫批處理程式或流處理程式，其中使用DataSet API進行批處理，并使用DataStream API進行流處理，

Flink有特殊的類DataSet和DataStream來表示程式中的資料，在DataSet的情況下，資料是有限的，而對于DataStream，元素的數量可以是無限的，

Flink程式看起來像轉換資料集合的常規程式，每個程式都包含相同的基本部分：

獲取一個執行環境
加載/創建初始資料
指定資料上的轉換
指定計算結果放在哪里
觸發程式執行

為了方便演示，先創建一個專案，可以從maven模板創建，例如：

mvn archetype:generate \
      -DarchetypeGroupId=org.apache.flink \
      -DarchetypeArtifactId=flink-quickstart-java \
      -DarchetypeVersion=1.10.0 \
      -DgroupId=com.cjs.example \
      -DartifactId=flink-quickstart \
      -Dversion=1.0.0-SNAPSHOT \
      -Dpackage=com.cjs.example.flink \
      -DinteractiveMode=false

也可以直接創建SpringBoot專案，自行引入依賴：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-java</artifactId>
    <version>1.10.0</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java_2.11</artifactId>
    <version>1.10.0</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka-0.10_2.11</artifactId>
    <version>1.10.0</version>
</dependency>

StreamExecutionEnvironment是所有Flink程式的基礎，你可以在StreamExecutionEnvironment上使用以下靜態方法獲得一個：

getExecutionEnvironment()

createLocalEnvironment()

createRemoteEnvironment(String host, int port, String... jarFiles)

通常，只需要使用getExecutionEnvironment()即可，因為該方法會根據背景關系自動推斷出當前的執行環境

從檔案中讀取資料，例如：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

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

對DataStream應用轉換，例如：

DataStream<String> input = ...;

DataStream<Integer> parsed = input.map(new MapFunction<String, Integer>() {
    @Override
    public Integer map(String value) {
        return Integer.parseInt(value);
    }
});

通過創建一個sink將結果輸出，例如：

writeAsText(String path)

print()

最后，呼叫StreamExecutionEnvironment上的execute()執行：

//  Triggers the program execution
env.execute();

//  Triggers the program execution asynchronously
final JobClient jobClient = env.executeAsync();
final JobExecutionResult jobExecutionResult = jobClient.getJobExecutionResult(userClassloader).get();

下面通過單詞統計的例子來加深對這一流程的理解，WordCount程式之于大資料就相當于是HelloWorld之于Java，哈哈哈

package com.cjs.example.flink;

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;

/**
 * Map-Reduce思想
 * 先分組，再求和
 * @author ChengJianSheng
 * @date 2020-05-26
 */
public class WordCount {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSet<String> text = env.readTextFile("/Users/asdf/Desktop/input.txt");
        DataSet<Tuple2<String, Integer>> counts =
                // split up the lines in pairs (2-tuples) containing: (word,1)
                text.flatMap(new Tokenizer())
                        // group by the tuple field "0" and sum up tuple field "1"
                        .groupBy(0)
                        .sum(1);
        
        counts.writeAsCsv("/Users/asdf/Desktop/aaa", "\n", " ");
        env.execute();
    }

    static class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            // normalize and split the line
            String[] tokens = value.toLowerCase().split("\\W+");

            // emit the pairs
            for (String token : tokens) {
                if (token.length() > 0) {
                    out.collect(new Tuple2<>(token, 1));
                }
            }
        }
    }
}

為Tuple定義keys

Python中也有Tuple（元組）

DataStream<Tuple3<Integer,String,Long>> input = // [...]
KeyedStream<Tuple3<Integer,String,Long>,Tuple> keyed = input.keyBy(0)

元組按第一個欄位(整數型別的欄位)分組

還可以使用POJO的屬性來定義keys，例如：

// some ordinary POJO (Plain old Java Object)
public class WC {
  public String word;
  public int count;
}
DataStream<WC> words = // [...]
DataStream<WC> wordCounts = words.keyBy("word").window(/*window specification*/);

先來了解一下KeyedStream

因此可以通過KeySelector方法來自定義

// some ordinary POJO
public class WC {public String word; public int count;}
DataStream<WC> words = // [...]
KeyedStream<WC> keyed = words
  .keyBy(new KeySelector<WC, String>() {
     public String getKey(WC wc) { return wc.word; }
   });

如何指定轉換方法呢？

方式一：匿名內部類

data.map(new MapFunction<String, Integer> () {
    public Integer map(String value) { return Integer.parseInt(value); }
});

方式二：Lamda

data.filter(s -> s.startsWith("http://"));
data.reduce((i1,i2) -> i1 + i2);

2. DataStream API

下面這個例子，每10秒鐘統計一次來自Web Socket的單詞次數

package com.cjs.example.flink;

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 WindowWordCount {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Tuple2<String, Integer>> dataStream = env.socketTextStream("localhost", 9999)
                .flatMap(new Splitter())
                .keyBy(0)
                .timeWindow(Time.seconds(10))
                .sum(1);

        dataStream.print();

        env.execute("Window WordCount");
    }

    static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            String[] words = value.split("\\W+");
            for (String word : words) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }
}

為了運行此程式，首先要在終端啟動一個監聽

nc -lk 9999

https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/datastream_api.html

轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/3123.html

標籤：大數據

上一篇：一、vmware搭建虛擬機集群（IP，防火墻，PATH環境變數）

下一篇：Spark Streaming，Flink，Storm，Kafka Streams，Samza：如何選擇流處理框架