1、一个简单的开始
- 创建一个maven项目，pom里添加flink依赖，可以参考 https://ashiamd.github.io/docsify-notes/#/study/BigData/Flink/%E5%B0%9A%E7%A1%85%E8%B0%B7Flink%E5%85%A5%E9%97%A8%E5%88%B0%E5%AE%9E%E6%88%98-%E5%AD%A6%E4%B9%A0%E7%AC%94%E8%AE%B0?id=_2-%e5%bf%ab%e9%80%9f%e4%b8%8a%e6%89%8b
- 创建如下代码，并在命令行窗口输入nc -lk 7777，然后启动程序，再在命令行窗口输入一些词即可开始统计
- StreamContextEnvironment.getExecutionEnvironment();创建执行环境
- env.execute();有这个才会执行
package BaseOperationTest;
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.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamContextEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
public class WordCount {
public static void main(String[] args) throws Exception {
// 创建执行环境
StreamExecutionEnvironment env = StreamContextEnvironment.getExecutionEnvironment();
// 从文件中读取数据
// String inputPath = "/Users/kaiker/Documents/projects/flink_study/src/main/resources/hello.txt";
// DataStream<String> inputDataStream = env.readTextFile(inputPath);
DataStream<String> inputDataStream = env.socketTextStream("localhost", 7777);
// 对数据集进行处理，按空格分词展开，转换成(word, 1)二元组进行统计
// 按照第一个位置的word分组
// 按照第二个位置上的数据求和
DataStream<Tuple2<String, Integer>> resultSet = inputDataStream.flatMap(new MyFlatMapper())
.keyBy(0)
.sum(1);
resultSet.print();
env.execute();
}
// 自定义类，实现FlatMapFunction接口
public static class MyFlatMapper implements FlatMapFunction<String, Tuple2<String, Integer>> {
@Override
public void flatMap(String s, Collector<Tuple2<String, Integer>> out) throws Exception {
// 按空格分词
String[] words = s.split(" ");
// 遍历所有word，包成二元组输出
for (String str : words) {
out.collect(new Tuple2<>(str, 1));
}
}
}
}
2、转换操作
2.2 基于KeyedStream的转换
keyBy
通过指定键值的方式将一个DataStream转化为KeyedStream。流中的事件会根据各自键值被分到不同的分区
KeyBy分区操作
DataStream<SensorReading> inputDataStream = env.socketTextStream("localhost", 7777);
keyStream = inputDataStream.keyBy(SensorReading::getId);
滚动聚合
sum() min() max() 不断获取数据，更新获取到的聚合值
minBy() maxBy() sumBy() 不断获取数据，更新获取到的聚合值，非聚合的值会使用最新的那一条更新
Reduce
每个到来的事件都会和reduce结果进行一次组合，reduce不会改变数据类型
接收两个参数，一个记录现有（或者说已经聚合的）记录，一个记录新来的记录，然后对两者进行聚合。
DataStream<SensorReading> resultStream = keyedStream.reduce(
(curSensor,newSensor)->new SensorReading(curSensor.getId(),newSensor.getTimestamp(), Math.max(curSensor.getTemperature(), newSensor.getTemperature()))
);
2.3 多流转换
Union
DataStream.union可以合并多条类型相同的DataStream，生成一个新的类型相同的DataStream
union
Connect
- connect会连接起两条流，默认情况下，不会使两条输入流的事件之间产生任何关联。但是可以结合keyBy等方法使用，让两条流有一定关联
- 有coMap\coFlatMap等方法可以分别对两条流进行处理
3、类型
- 原始类型
- Java和Scala Tuple，Flink提供了Java元组的高效实现，最多可包含25个字段
- Scala case类
- POJO类，如果一个类有公共类、有一个无参构造器、每个字段都有getter和setter、字段类型可以支持，则可以被分析为POJO
- 其他特殊类型
为数据类型创建类型信息
TypeInformation<Integer> intType = Types.INT
显式提供类型信息
有时候一些必要的信息可能无法提取，比如Java的类型擦除
可以通过实现ResultTypeQueryable接口来提供类型，也可以使用returns()方法来显示指定某算子的返回类型
DataStream<Person> persons = tuples
.map(t -> new Person(t.f0, t.f1)).returns(Types..POJO(Person.class));
4、富函数
- DataStreamAPI中所有转换函数都有对应的富函数
- 富函数可以在处理第一条数据之前进行初始化操作，获取到一些上下文信息
public static class MyMapFunction extends RichMapFunction<SensorReading, Tuple2<Integer, String>> {
@Override public Tuple2<Integer, String> map(SensorReading value) throws Exception {
return new Tuple2<>(getRuntimeContext().getIndexOfThisSubtask(), value.getId());
}
@Override public void open(Configuration parameters) throws Exception {
System.out.println(getRuntimeContext.getIndexOfThisSubtask); // 以下可以做一些初始化工作，例如建立一个和HDFS的连接
}
@Override public void close() throws Exception {
System.out.println("my map close"); // 以下做一些清理工作，例如断开和HDFS的连接
}
}
网友评论