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

Download and install flink:

Upload compressed package: flink-1.7.2-bin-scala_2.12.tgz

Decompress: tar-zxvf / flink-1.7.2-bin-scala_2.12.tgz-C.. / hone

Copy the extracted file to the child node:

Scp-r / home/flink-1.7.2/ root@slave1:/home/

Scp-r / home/flink-1.7.2/ root@slave2:/home/

Modify the configuration file: select a master node and configure conf/flink-conf.yaml

Vi conf/flink-conf.yaml

Set the jobmanager.rpc.address configuration item to the IP or hostname of the node

Jobmanager.rpc.address: 10.108.4.202

Then add the child node configuration:

In all nodes: under the flink directory: vi conf/slaves

Add all child nodes ip and then save

Start the local flink cluster:

Cd to the flink directory

. / bin/start-cluster.sh

View webui:ip:8081

Start listening: nc-lk 9000

When reporting that the nc command does not exist (yum install nc)

Then execute the test jar:

Stop the flink cluster: bin/stop-cluster.sh

Submit tasks in a cluster: under the flink directory

. / bin/flink run-m yarn-cluster-c com.demo.florian.WordCount $DEMO_DIR/target/flink-demo-1.0-SNAPSHOT.jar-- port 9000

Create a new maven program

Pom.xml dependencies are as follows:

Then create a new TestSocketWindowWordCount class with the following code

Then start the flink cluster-> create a new listener: nc-lk 6666

Then start the TestSocketWindowWordCount class

Enter the code on the linux monitoring page

Observe that there is statistical output in the idea console.

-pom.xml starts-

Org.apache.flink

Flink-java

1.9.0

Compile

Org.apache.flink

Flink-streaming-java_2.11

1.9.0

Compile

Org.apache.flink

Flink-streaming-scala_2.11

1.6.2

Org.apache.flink

Flink-clients_2.11

1.6.2

-pom.xml ends-

-TestSocketWindowWordCount starts-

Package com.gyb

Import org.apache.flink.api.common.functions.FlatMapFunction

Import org.apache.flink.api.common.functions.ReduceFunction

Import org.apache.flink.streaming.api.datastream.DataStream

Import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator

Import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment

Import org.apache.flink.streaming.api.windowing.time.Time

Import org.apache.flink.util.Collector

Import javax.xml.soap.Text

Public class TestSocketWindowWordCount {

Public static void main (String args []) {

String hostname = "192.168.198.130"

Int port = 6666

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment ()

DataStream text = env.socketTextStream (hostname, port, "\ n"); / / get the execution environment

SingleOutputStreamOperator windowCounts = text

.flatMap (new FlatMapFunction () {@ Overridebr/ > @ Override)

For (String word: value.split ("\ s")) {

Out.collect (new SocketWindowWordCount.WordWithCount (word, 1L))

}

}

})

.keyby ("word")

.timeWindow (Time.seconds (5), Time.seconds (5))

.reduce (new ReduceFunction () {@ Overridebr/ > @ Override

Return new SocketWindowWordCount.WordWithCount (a.word, a.count + b.count)

}

});

/ / print the results with a single thread, rather than in parallelwindowCounts.print (). SetParallelism (1); / / env.execute ("Socket Window WordCount"); try {env.execute ("Socket Window WordCount");} catch (Exception e) {e.printStackTrace ();}

}

Public static class WordWithCount {public String word; public long count; public WordWithCount () {} public WordWithCount (String word, long count) {this.word = word; this.count = count;} @ Override public String toString () {return word + ":" + count;}}

}

-TestSocketWindowWordCount ends-

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.