如何在flink中傳遞參數

眾所周知，flink作為流計算引擎，處理源源不斷的數據是其本意，但是在處理數據的過程中，往往可能需要一些參數的傳遞，那么有哪些方法進行參數的傳遞？在什么時候使用？這里嘗試進行簡單的總結。

• 使用configuration

　　在main函數中定義變量

// Class in Flink to store parameters
Configuration configuration = new Configuration();
configuration.setString("genre", "Action");

lines.filter(new FilterGenreWithParameters())
        // Pass parameters to a function
        .withParameters(configuration)
        .print();

　　使用參數的function需要繼承自一個rich的function，這樣才可以在open方法中獲取相應的參數。

class FilterGenreWithParameters extends RichFilterFunction<Tuple3<Long, String, String>> {

    String genre;

    @Override
    public void open(Configuration parameters) throws Exception {
        // Read the parameter
        genre = parameters.getString("genre", "");
    }

    @Override
    public boolean filter(Tuple3<Long, String, String> movie) throws Exception {
        String[] genres = movie.f2.split("\\|");

        return Stream.of(genres).anyMatch(g -> g.equals(genre));
    }
}

 

• 使用ParameterTool

使用configuration雖然傳遞了參數，但顯然不夠動態，每次參數改變，都涉及到程序的變更，既然main函數能夠接受參數，flink自然也提供了相應的承接的機制，即ParameterTool。

如果使用ParameterTool，則在參數傳遞上如下

public static void main(String... args) {
    // Read command line arguments
    ParameterTool parameterTool = ParameterTool.fromArgs(args);
    
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.getConfig().setGlobalJobParameters(parameterTool);
...

// This function will be able to read these global parameters
lines.filter(new FilterGenreWithGlobalEnv())
                .print();
}

如上面代碼，使用parameterTool來承接main函數的參數，通過env來設置全局變量來進行分發，那么在繼承了rich函數的邏輯中就可以使用這個全局參數。

class FilterGenreWithGlobalEnv extends RichFilterFunction<Tuple3<Long, String, String>> {

    @Override
    public boolean filter(Tuple3<Long, String, String> movie) throws Exception {
        String[] genres = movie.f2.split("\\|");
        // Get global parameters
        ParameterTool parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
        // Read parameter
        String genre = parameterTool.get("genre");

        return Stream.of(genres).anyMatch(g -> g.equals(genre));
    }
}

 

• 使用broadcast變量

在上面使用configuration和parametertool進行參數傳遞會很方便，但是也僅僅適用於少量參數的傳遞，如果有比較大量的數據傳遞，flink則提供了另外的方式來進行，其中之一即是broadcast，這個也是在其他計算引擎中廣泛使用的方法之一。

DataSet<Integer> toBroadcast = env.fromElements(1, 2, 3);
// Get a dataset with words to ignore
DataSet<String> wordsToIgnore = ...

data.map(new RichFlatMapFunction<String, String>() {

    // A collection to store words. This will be stored in memory
    // of a task manager
    Collection<String> wordsToIgnore;

    @Override
    public void open(Configuration parameters) throws Exception {
        // Read a collection of words to ignore
        wordsToIgnore = getRuntimeContext().getBroadcastVariable("wordsToIgnore");
    }


    @Override
    public String map(String line, Collector<String> out) throws Exception {
        String[] words = line.split("\\W+");
        for (String word : words)
            // Use the collection of words to ignore
            if (wordsToIgnore.contains(word))
                out.collect(new Tuple2<>(word, 1));
    }
    // Pass a dataset via a broadcast variable
}).withBroadcastSet(wordsToIgnore, "wordsToIgnore");

在第3行定義了需要進行廣播的數據集，在第27行指定了將此數據集進行廣播的目的地。

廣播的變量會保存在tm的內存中，這個也必然會使用tm有限的內存空間，也因此不能廣播太大量的數據。

那么，對於數據量更大的廣播需要，要如何進行？flink也提供了緩存文件的機制，如下。

• 使用distributedCache

首先還是需要在定義dag圖的時候指定緩存文件：

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

// Register a file from HDFS
env.registerCachedFile("hdfs:///path/to/file", "machineLearningModel")

...

env.execute()

flink本身支持指定本地的緩存文件，但一般而言，建議指定分布式存儲比如hdfs上的文件，並為其指定一個名稱。

使用起來也很簡單，在rich函數的open方法中進行獲取。

class MyClassifier extends RichMapFunction<String, Integer> {

    @Override
    public void open(Configuration config) {
      File machineLearningModel = getRuntimeContext().getDistributedCache().getFile("machineLearningModel");
      ...
    }

    @Override
    public Integer map(String value) throws Exception {
      ...
    }
}

上面的代碼忽略了對文件內容的處理。

在上面的幾個方法中，應該說參數本身都是static的，不會變化，那么如果參數本身隨着時間也會發生變化，怎么辦？

嗯，那就用connectStream，其實也是流的聚合了。

• 使用connectStream

使用ConnectedStream的前提當然是需要有一個動態的流，比如在主數據之外，還有一些規則數據，這些規則數據會通過Restful服務來發布，假如我們的主數據來自於kafka，

那么，就可以如下：

DataStreamSource<String> input = (DataStreamSource) KafkaStreamFactory
                .getKafka08Stream(env, srcCluster, srcTopic, srcGroup);

DataStream<Tuple2<String, String>> appkeyMeta = env.addSource(new AppKeySourceFunction(), "appkey")

ConnectedStreams<String, Tuple2<String, String>> connectedStreams = input.connect(appkeyMeta.broadcast());

DataStream<String> cleanData = connectedStreams.flatMap(new DataCleanFlatMapFunction())

其實可以看到，上面的代碼中做了四件事，首先在第1行定義了獲取主數據的流，在第4行定義了獲取規則數據的流，在AppKeySourceFunction中實現了讀取Restful的邏輯，

在第6行實現了將規則數據廣播到主數據中去，最后在第8行實現了從connectedStream中得到經過處理的數據。其中的關鍵即在於DataCleanFlatMapFunction。

public class DataCleanFlatMapFunction extends RichCoFlatMapFunction<String, Tuple2<String, String>, String>{

public void flatMap1(String s, Collector<String> collector){...}

public void flatMap2(Tuple2<String, String> s, Collector<String> collector) {...}


}

這是一段縮減的代碼，關鍵在於第一行，首先這個函數需要實現RichCoFlatMapFunction這個抽象類，其次在類實現中，flatMap2會承接規則函數，flatMap1會承接主函數。

 

 

當然，參數可以從client發送到task，有時候也需要從task發回到client，一般這里就會使用accumulator。

這里先看一個簡單的例子，實現單詞的計數以及處理文本的記錄數：

DataSet<String> lines = ...

// Word count algorithm
lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
    @Override
    public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
        String[] words = line.split("\\W+");
        for (String word : words) {
            out.collect(new Tuple2<>(word, 1));
        }
    }
})
.groupBy(0)
.sum(1)
.print();

// Count a number of lines in the text to process
int linesCount = lines.count()
System.out.println(linesCount);

上面的代碼中，第14行實現了單詞的計算，第18行實現了處理記錄的行數，但很可惜，這里會產生兩個job，僅僅第18行一句代碼，就會產生一個job，無疑是不高效的。

flink提供了accumulator來實現數據的回傳，亦即從tm傳回到JM。

flink本身提供了一些內置的accumulator:

• IntCounter, LongCounter, DoubleCounter – allows summing together int, long, double values sent from task managers
• AverageAccumulator – calculates an average of double values
• LongMaximum, LongMinimum, IntMaximum, IntMinimum, DoubleMaximum, DoubleMinimum – accumulators to determine maximum and minimum values for different types
• Histogram – used to computed distribution of values from task managers

首先需要定義一個accumulator，然后在某個自定義函數中來注冊它，這樣在客戶端就可以獲取相應的的值。

lines.flatMap(new RichFlatMapFunction<String, Tuple2<String, Integer>>() {

    // Create an accumulator
    private IntCounter linesNum = new IntCounter();

    @Override
    public void open(Configuration parameters) throws Exception {
        // Register accumulator
        getRuntimeContext().addAccumulator("linesNum", linesNum);
    }

    @Override
    public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
        String[] words = line.split("\\W+");
        for (String word : words) {
            out.collect(new Tuple2<>(word, 1));
        }
        
        // Increment after each line is processed
        linesNum.add(1);
    }
})
.groupBy(0)
.sum(1)
.print();

// Get accumulator result
int linesNum = env.getLastJobExecutionResult().getAccumulatorResult("linesNum");
System.out.println(linesNum);

當然，如果內置的accumulator不能滿足需求，可以自定義accumulator，只需要繼承兩個接口之一即可，Accumulator或者SimpleAccumulato。

 

上面介紹了幾種參數傳遞的方式，在日常的使用中，可能不僅僅是使用其中一種，或許是某些的組合，比如通過parametertool來傳遞hdfs的路徑，再通過filecache來讀取緩存。