Flink開發(fā)怎樣進(jìn)行實時處理應(yīng)用程序

本篇文章為大家展示了Flink開發(fā)怎樣進(jìn)行實時處理應(yīng)用程序，內(nèi)容簡明扼要并且容易理解，絕對能使你眼前一亮，通過這篇文章的詳細(xì)介紹希望你能有所收獲。

使用Flink + java實現(xiàn)需求

環(huán)境

JDK：1.8

Maven：3.6.1（最低Maven 3.0.4）

使用上一節(jié)中的springboot-flink-train項目

開發(fā)步驟

第一步：創(chuàng)建流處理上下文環(huán)境

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

第二步：讀取數(shù)據(jù)，使用socket流方式讀取數(shù)據(jù)

DataStreamSource<String> text = env.socketTextStream("192.168.152.45", 9999);

第三步：transform

        text.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] tokens = value.toLowerCase().split(",");
                for(String token: tokens) {
                    if(token.length() > 0) {
                        out.collect(new Tuple2<String, Integer>(token, 1));
                    }
                }
            }
        }).keyBy(0).timeWindow(Time.seconds(5)).sum(1).print();

這里我們使用逗號分隔，然后跟批處理不同的是，這里使用keyBy(0)，而不是groupBy(0)。timewindow表示每隔多久執(zhí)行一次。

第四步：執(zhí)行

env.execute("StreamingWCJavaApp");

整體代碼如下：

/**
 * 使用Java API來開發(fā)Flink的實時處理應(yīng)用程序
 * wc統(tǒng)計的數(shù)據(jù)源自socket
 */
public class StreamingWCJava02App {

    public static void main(String[] args) throws Exception {

        // 獲取參數(shù)
        int port;
        try{
            ParameterTool tool = ParameterTool.fromArgs(args);
            port = tool.getInt("port");
        } catch (Exception e) {
            System.out.println("端口未設(shè)置, 使用默認(rèn)端口9999");
            port = 9999;
        }


        // step1: 獲取流處理上下文環(huán)境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // step2: 讀取數(shù)據(jù)
        DataStreamSource<String> text = env.socketTextStream("192.168.152.45", port);
        // step3: transform
        text.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] tokens = value.toLowerCase().split(",");
                for(String token: tokens) {
                    if(token.length() > 0) {
                        out.collect(new Tuple2<String, Integer>(token, 1));
                    }
                }
            }
        }).keyBy(0).timeWindow(Time.seconds(5)).sum(1).print();
        env.execute("StreamingWCJavaApp");
    }

}

運行

首先在192.168.152.45上運行命令

nc -l 9999

然后在運行main方法。在192.168.152.45的nc上輸入

abc,def,abc,ddd

在idea控制臺輸出如下：

4> (abc,2)
1> (def,1)
4> (ddd,1)

這個前面的"4>"表示并行度。我們可以設(shè)置setParallelism(1)來忽略這個問題。如下所示：

        text.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] tokens = value.toLowerCase().split(",");
                for(String token: tokens) {
                    if(token.length() > 0) {
                        out.collect(new Tuple2<String, Integer>(token, 1));
                    }
                }
            }
        }).keyBy(0).timeWindow(Time.seconds(5)).sum(1).print().setParallelism(1);

這樣控制臺的打印結(jié)果如下：

(abc,2)
(ddd,1)
(def,1)

這樣一個簡單的demo就成功了！

重構(gòu)代碼

上面的代碼中l(wèi)ocalhost與port需要用參數(shù)傳遞進(jìn)來。

代碼如下：

        // 獲取參數(shù)
        int port;
        try{
            ParameterTool tool = ParameterTool.fromArgs(args);
            port = tool.getInt("port");
        } catch (Exception e) {
            System.out.println("端口未設(shè)置, 使用默認(rèn)端口9999");
            port = 9999;
        }

使用Flink提供的ParameterTool來接收參數(shù)。

我們在運行時就可以指定參數(shù)列表了，其中的key必須以“-”或者“--”開頭。

在運行時，配置參數(shù)：

這樣運行就可以從外界傳遞參數(shù)了

使用Flink + Scala實現(xiàn)需求

接下來使用Scala方式實現(xiàn)，在項目springboot-flink-train-scala中新建StreamingWCScalaApp，內(nèi)容如下：

/**
  * 使用Scala開發(fā)Flink的實時處理應(yīng)用程序
  */
object StreamingWCScalaApp {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 引入隱式轉(zhuǎn)換
    import org.apache.flink.api.scala._

    val text = env.socketTextStream("192.168.152.45", 9999)
    text.flatMap(_.split(","))
        .map((_,1))
        .keyBy(0)
        .timeWindow(Time.seconds(5))
        .sum(1)
        .print()
        .setParallelism(1)

    env.execute("StreamingWCScalaApp");
  }
}

這種方式比java實現(xiàn)更加簡潔。

上述內(nèi)容就是Flink開發(fā)怎樣進(jìn)行實時處理應(yīng)用程序，你們學(xué)到知識或技能了嗎？如果還想學(xué)到更多技能或者豐富自己的知識儲備，歡迎關(guān)注億速云行業(yè)資訊頻道。