Shulou(Shulou.com)06/02 Report--

This article shows you how to develop Flink real-time processing applications, the content is concise and easy to understand, can definitely brighten your eyes, through the detailed introduction of this article, I hope you can get something.

Using Flink + java to realize the requirement environment

JDK:1.8

Maven:3.6.1 (minimum Maven 3.0.4)

Use the springboot-flink-train project from the previous section

Development steps

Step 1: create a streaming context

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment ()

Step 2: read the data and use socket stream to read the data

DataStreamSource text = env.socketTextStream ("192.168.152.45", 9999)

Step 3: transform

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

Here we use comma separation, and then unlike batch processing, we use keyBy (0) instead of groupBy (0). Timewindow indicates how often it is executed.

Step 4: execute

Env.execute ("StreamingWCJavaApp")

The overall code is as follows:

/ * use Java API to develop real-time processing applications for Flink * wc statistics are derived from socket * / public class StreamingWCJava02App {public static void main (String [] args) throws Exception {/ / get parameters int port; try {ParameterTool tool = ParameterTool.fromArgs (args); port = tool.getInt ("port") } catch (Exception e) {System.out.println ("Port is not set, default port 9999"); port = 9999;} / / step1: get stream processing context StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment (); / / step2: read data DataStreamSource text = env.socketTextStream ("192.168.152.45", port) / / step3: transform text.flatMap (new FlatMapFunction () {@ Override public void flatMap (String value, Collector out) throws Exception {String [] tokens = value.toLowerCase () .split (",") For (String token: tokens) {if (token.length () > 0) {out.collect (new Tuple2 (token, 1));}) .keyby (0) .timeWindow (Time.seconds (5)) .sum (1) .print () Env.execute ("StreamingWCJavaApp");}} run

First run the command on 192.168.152.45

Nc-l 9999

Then run the main method. Enter on the nc of 192.168.152.45

Abc,def,abc,ddd

The output in the idea console is as follows:

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

The "4 >" before this indicates the degree of parallelism. We can set setParallelism (1) to ignore this problem. As follows:

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

The print result of the console is as follows:

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

Such a simple demo is a success!

Refactoring code

In the above code, localhost and port need to be passed in as parameters.

The code is as follows:

/ / get parameters int port; try {ParameterTool tool = ParameterTool.fromArgs (args); port = tool.getInt ("port");} catch (Exception e) {System.out.println ("Port is not set, default port 9999"); port = 9999;}

Use the ParameterTool provided by Flink to receive parameters.

We can specify the parameter list at run time, where the key must start with "-" or "-".

At run time, configuration parameters:

In this way, you can pass parameters from the outside world.

Use Flink + Scala to achieve requirements

Next, use the Scala method to create a new StreamingWCScalaApp in the project springboot-flink-train-scala, as follows:

/ * use Scala to develop real-time processing applications for Flink * / object StreamingWCScalaApp {def main (args: Array [String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironment / / introduce implicit conversion import org.apache.flink.api.scala._ val text = env.socketTextStream ("192.168.152.45", 9999) text.flatMap (_ .split (",")) .map ((_ ) .keyby (0) .timeWindow (Time.seconds (5)) .sum (1) .print () .setParallelism (1) env.execute ("StreamingWCScalaApp") }}

This approach is more concise than the java implementation.

The above is how Flink developers deal with applications in real time. Have you learned any knowledge or skills? If you want to learn more skills or enrich your knowledge reserve, you are welcome to follow the industry information channel.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.