Flink具有特殊类DataSet并DataStream在程序中表示数据。您可以将它们视为可以包含重复项的不可变数据集合。在DataSet数据有限的情况下，对于一个DataStream元素的数量可以是无界的。
这些集合在某些关键方面与常规Java集合不同。首先，它们是不可变的，这意味着一旦创建它们就无法添加或删除元素。你也不能简单地检查里面的元素。
集合最初通过在弗林克程序添加源创建和新的集合从这些通过将它们使用API方法如衍生map，filter等等。
Flink程序看起来像是转换数据集合的常规程序。每个程序包含相同的基本部分：
1.获取execution environment,
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
2.加载/创建初始化数据
DataStream<String> text = env.readTextFile("file:///path/to/file");
3.指定此数据的转换
val mapped = input.map { x => x.toInt }
4.指定放置计算结果的位置
writeAsText(String path)
print()
5.触发程序执行
在local模式下执行程序
execute()
将程序达成jar运行在线上
./bin/flink run \
-m node21:8081 \
./examples/batch/WordCount.jar \
--input hdfs:///user/admin/input/wc.txt \
--output hdfs:///user/admin/output2 \
二. Wordcount案例 1.Scala代码package com.xyg.streaming import org.apache.flink.api.java.utils.ParameterTool import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment import org.apache.flink.streaming.api.windowing.time.Time /** * Author: Mr.Deng * Date: 2018/10/15 * Desc: */ object SocketWindowWordCountScala { def main(args: Array[String]) : Unit = { // 定义一个数据类型保存单词出现的次数 case class WordWithCount(word: String, count: Long) // port 表示需要连接的端口 val port: Int = try { ParameterTool.fromArgs(args).getInt("port") } catch { case e: Exception => { System.err.println("No port specified. Please run 'SocketWindowWordCount --port <port>'") return } } // 获取运行环境 val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment // 连接此socket获取输入数据 val text = env.socketTextStream("node21", port, '\n') //需要加上这一行隐式转换 否则在调用flatmap方法的时候会报错 import org.apache.flink.api.scala._ // 解析数据, 分组, 窗口化, 并且聚合求SUM val windowCounts = text .flatMap { w => w.split("\\s") } .map { w => WordWithCount(w, 1) } .keyBy("word") .timeWindow(Time.seconds(5), Time.seconds(1)) .sum("count") // 打印输出并设置使用一个并行度 windowCounts.print().setParallelism(1) env.execute("Socket Window WordCount") } }
2.Java代码package com.xyg.streaming; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.java.utils.ParameterTool; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.util.Collector; /** * Author: Mr.Deng * Date: 2018/10/15 * Desc: 使用flink对指定窗口内的数据进行实时统计，最终把结果打印出来 * 先在node21机器上执行nc -l 9000 */ public class StreamingWindowWordCountJava { public static void main(String[] args) throws Exception { //定义socket的端口号 int port; try{ ParameterTool parameterTool = ParameterTool.fromArgs(args); port = parameterTool.getInt("port"); }catch (Exception e){ System.err.println("没有指定port参数，使用默认值9000"); port = 9000; } //获取运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //连接socket获取输入的数据 DataStreamSource<String> text = env.socketTextStream("node21", port, "\n"); //计算数据 DataStream<WordWithCount> windowCount = text.flatMap(new FlatMapFunction<String, WordWithCount>() { public void flatMap(String value, Collector<WordWithCount> out) throws Exception { String[] splits = value.split("\\s"); for (String word:splits) { out.collect(new WordWithCount(word,1L)); } } })//打平操作，把每行的单词转为<word,count>类型的数据 //针对相同的word数据进行分组 .keyBy("word") //指定计算数据的窗口大小和滑动窗口大小 .timeWindow(Time.seconds(2),Time.seconds(1)) .sum("count"); //把数据打印到控制台,使用一个并行度 windowCount.print().setParallelism(1); //注意：因为flink是懒加载的，所以必须调用execute方法，上面的代码才会执行 env.execute("streaming word count"); } /** * 主要为了存储单词以及单词出现的次数 */ 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 "WordWithCount{" + "word='" + word + '\'' + ", count=" + count + '}'; } } }
3.运行测试首先，使用nc命令启动一个本地监听，命令是：
[admin@node21 ~]$ nc -l 9000