【flink】flink實作wordCount(java)

文章目錄

• 前言
• 一、wordCount
• • 1、有界wordCount
  • 2、無界wordCount
  • 3、pom.xml

前言

之前學習了spark,現在正在學習flink，感覺很類似，記錄下flink的學習過程，請大家參考。

一、wordCount

1、有界wordCount

代碼如下（示例）：

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class WordCount {

	public static void main(String[] args) throws Exception{
		// 建立執行環境
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

		// 從檔案中讀取資料
		String inputPath = "E:\\webapps\\test\\flink-demo\\data\\1.txt";
		//這裡和spark很相似了
		DataSet<String> inputDataSet = env.readTextFile (inputPath);

		// 對資料集進行處理，按空格分詞展開，轉換成(word, 1)二進制組進行統計
		DataSet<Tuple2<String, Integer>> resultSet = inputDataSet.flatMap(new MyFlatMapper())
				.groupBy(0)   // 按照第一個位置的word分組
				.sum(1);    // 将第二個位置上的資料求和
		resultSet.print();
	}

	// 自定義類，實作FlatMapFunction接口
	public static class MyFlatMapper implements FlatMapFunction<String, Tuple2<String, Integer>>
	{
		public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
			// 按空格分詞
			String[] words = value.split(" ");
			// 周遊所有word，包成二進制組輸出
			for (String word : words) {
				out.collect(new Tuple2<String,Integer>(word, 1));
			}
		}
	}
}
           

2、無界wordCount

代碼如下（示例）：

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
public class WordCount2
{
	public static void main(String[] args) throws Exception{
		// 建立流處理執行環境
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		String host = "localhost"; 
		int port = 7777; 
		// 從socket文本流讀取資料
		DataStream<String> inputDataStream = env.socketTextStream(host, port);
		//處理無界資料的時候，沒有groupBy方法，groupBy可以了解為當資料全加載完成之後進行分組
		// 無界是沒有辦法加載全部資料，是以替換為keyBy方法
		DataStream<Tuple2<String, Integer>> resultStream = inputDataStream.flatMap(new WordCount.MyFlatMapper())
				.keyBy (0).sum (1);
		resultStream.print();
		// 執行任務
		env.execute();
	}
}
           

無界啟動時，使用nc工具進行測試，通過nc工具進行socket資訊傳輸。

啟動指令

nc -l -p 7777

或者

nc -lk 7777

，可能由于nc版本不同或者執行環境不同，指令可能

如果出現

local listen fuxored: INVAL

異常，使用

nc -l -p 7777

即可。

3、pom.xml

<dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.10.1</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.12</artifactId>
            <version>1.10.1</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>