下載安裝flink：
上傳壓縮包：flink-1.7.2-bin-scala_2.12.tgz
解壓：tar -zxvf /flink-1.7.2-bin-scala_2.12.tgz -C ../hone
復制解壓文件到子節點：
scp -r /home/flink-1.7.2/ root@slave1:/home/
scp -r /home/flink-1.7.2/ root@slave2:/home/
修改配置文件：選擇一個master節點，配置conf/flink-conf.yaml
vi conf/flink-conf.yaml
設置jobmanager.rpc.address 配置項為該節點的IP 或者主機名
jobmanager.rpc.address: 10.108.4.202
然后添加子節點配置：
在所有的節點中：flink目錄下：vi conf/slaves
添加所有子節點ip然后保存
啟動本地的flink集群：
cd 到flink目錄下
./bin/start-cluster.sh
查看webui:ip:8081
啟動監聽：nc -lk 9000
當報nc命令不存在時（yum install nc）
然后執行測試jar：
停止flink集群：bin/stop-cluster.sh
以集群方式提交任務：在flink目錄下
./bin/flink run -m yarn-cluster -c com.demo.florian.WordCount $DEMO_DIR/target/flink-demo-1.0-SNAPSHOT.jar --port 9000

新建maven程序
pom.xml依賴如下：
然后新建一個TestSocketWindowWordCount類具體代碼如下
然后啟動flink集群->新建一個監聽:nc -lk 6666
然后啟動TestSocketWindowWordCount類
在linux監聽頁面輸入代碼
觀察在idea控制臺就有統計的輸出
-------pom.xml開始-----------


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結束-----------
-------TestSocketWindowWordCount開始------------------
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");//獲取執行環境
SingleOutputStreamOperator windowCounts = text
.flatMap(new FlatMapFunction() {@Override
br/>@Override
out) {
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() {@Override
br/>@Override
return new SocketWindowWordCount.WordWithCount(a.word, a.count + b.count);
}
});

// print the results with a single thread, rather than in parallel
windowCounts.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結束------------------

網站題目：flink偽分布式搭建及其本地idea測flink連接
網頁鏈接：http://m.jcarcd.cn/article/iesgdh.html