Flink pseudo-distributed Construction and its Local idea Test flink connection
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-
