大數(shù)據(jù)HelloWorld-Flink實(shí)現(xiàn)WordCount

mvn archetype:generate                \

   -DarchetypeGroupId=org.apache.flink       \

   -DarchetypeArtifactId=flink-quickstart-java   \

   -DarchetypeVersion=1.8.0

mvn archetype:generate                \

   -DarchetypeGroupId=org.apache.flink       \

   -DarchetypeArtifactId=flink-quickstart-scala   \

   -DarchetypeVersion=1.8.0

├── pom.xml
└── src
  └── main
    ├── resources
    │  └── log4j.properties
    └── scala/java
      └── org
        └── myorg
          └── quickstart
            ├── BatchJob.scala
            └── StreamingJob.scala

package jar;

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

public class BatchJob {

	public static void main(String[] args) throws Exception {
		// set up the batch execution environment
		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		//讀取目錄下的文件
		DataSourceString> data = env.readTextFile("/opt/Server_Packets/log/ServerLog_1_runtime.log");
		//把文件中的內(nèi)容按照空格進(jìn)行拆分為 word,1  1 是為了能夠在下面進(jìn)行計(jì)算.
		data.flatMap(new FlatMapFunctionString, Tuple2String, Integer>>() {
			@Override
			public void flatMap(String s, CollectorTuple2String, Integer>> collector) throws Exception {
				for (String word : s.split(" ")){
					collector.collect(new Tuple2>(word,1));
				}
			}
		})
		// 按照元組中的第1位進(jìn)行分組
		.groupBy(0)
		// 分組的元組的計(jì)算方式為 value +value 也就是剛才的 同樣的詞 把 1+1
		.reduce(new ReduceFunctionTuple2String, Integer>>() {
			@Override
			public Tuple2String, Integer> reduce(Tuple2String, Integer> t2, Tuple2String, Integer> t1) throws Exception {
				return new Tuple2>(t1.f0,t1.f1+ t2.f1);
			}
		})
		//輸出結(jié)果
		.print();
	}
}