189 8069 5689
这篇文章主要介绍“Flink数据流DataStream和DataSet怎么使用”，在日常操作中，相信很多人在Flink数据流DataStream和DataSet怎么使用问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”Flink数据流DataStream和DataSet怎么使用”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！
创新互联-专业网站定制、快速模板网站建设、高性价比阿尔山网站开发、企业建站全套包干低至880元,成熟完善的模板库,直接使用。一站式阿尔山网站制作公司更省心,省钱,快速模板网站建设找我们，业务覆盖阿尔山地区。费用合理售后完善，10多年实体公司更值得信赖。
Flink主要用来处理数据流，所以从抽象上来看就是对数据流的处理，正如前面大数据开发-Flink-体系结构 && 运行架构提到写Flink程序实际上就是在写DataSource、Transformation、Sink.
DataSource是程序的数据源输入，可以通过StreamExecutionEnvironment.addSource(sourceFuntion)为程序 添加一个数据源
Transformation是具体的操作，它对一个或多个输入数据源进行计算处理，比如Map、FlatMap和Filter等操作
Sink是程序的输出，它可以把Transformation处理之后的数据输出到指定的存储介质中
Flink针对DataStream提供了两种实现方式的数据源，可以归纳为以下四种：
基于文件
readTextFile(path) 读取文本文件，文件遵循TextInputFormat逐行读取规则并返回
基于Socket
socketTextStream 从Socket中读取数据，元素可以通过一个分隔符分开
基于集合
fromCollection(Collection) 通过Java的Collection集合创建一个数据流，集合中的所有元素必须是相同类型的，需要注意的是，如果集合里面的元素要识别为POJO，需要满足下面的条件
总结：上面的要求其实就是为了让Flink可以方便地序列化和反序列化这些对象为数据流
该类有共有的无参构造方法
该类是共有且独立的（没有非静态内部类）
类（及父类）中所有的不被static、transient修饰的属性要么有公有的（且不被final修饰），要么是包含公有的getter和setter方法，这些方法遵循java bean命名规范
自定义Source
使用StreamExecutionEnvironment.addSource(sourceFunction)将一个流式数据源加到程序中，具体这个sourceFunction 是为非并行源implements SourceFunction，或者为并行源 implements ParallelSourceFunction接口，或者extends RichParallelSourceFunction，对于自定义Source，Sink, Flink内置了下面几种Connector
| 连接器 | 是否提供Source支持 | 是否提供Sink支持 |
|---|---|---|
| Apache Kafka | 是 | 是 |
| ElasticSearch | 否 | 是 |
| HDFS | 否 | 是 |
| Twitter Streaming PI | 是 | 否 |
对于Source的使用，其实较简单，这里给一个较常用的自定义Source的KafaSource的使用例子。更多相关源码可以查看：
package com.hoult.stream;
public class SourceFromKafka {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
String topic = "animalN";
Properties props = new Properties();
props.put("bootstrap.servers", "linux121:9092");
FlinkKafkaConsumer consumer = new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), props);
DataStreamSource data = env.addSource(consumer);
SingleOutputStreamOperator> maped = data.map(new MapFunction>() {
@Override
public Tuple2 map(String value) throws Exception {
System.out.println(value);
Tuple2 t = new Tuple2(0l,0l);
String[] split = value.split(",");
try{
t = new Tuple2(Long.valueOf(split[0]), Long.valueOf(split[1]));
} catch (Exception e) {
e.printStackTrace();
}
return t;
}
});
KeyedStream, Long> keyed = maped.keyBy(value -> value.f0);
//按照key分组策略，对流式数据调用状态化处理
SingleOutputStreamOperator> flatMaped = keyed.flatMap(new RichFlatMapFunction, Tuple2>() {
ValueState> sumState;
@Override
public void open(Configuration parameters) throws Exception {
//在open方法中做出State
ValueStateDescriptor> descriptor = new ValueStateDescriptor<>(
"average",
TypeInformation.of(new TypeHint>() {
}),
Tuple2.of(0L, 0L)
);
sumState = getRuntimeContext().getState(descriptor);
// super.open(parameters);
}
@Override
public void flatMap(Tuple2 value, Collector> out) throws Exception {
//在flatMap方法中，更新State
Tuple2 currentSum = sumState.value();
currentSum.f0 += 1;
currentSum.f1 += value.f1;
sumState.update(currentSum);
out.collect(currentSum);
/*if (currentSum.f0 == 2) {
long avarage = currentSum.f1 / currentSum.f0;
out.collect(new Tuple2<>(value.f0, avarage));
sumState.clear();
}*/
}
});
flatMaped.print();
env.execute();
}
} 对于Transformation ，Flink提供了很多的算子，
map
DataStream → DataStream Takes one element and produces one element. A map function that doubles the values of the input stream:
