说明
通过 state 来实现 maxBy 功能
案例说明
温度传感器发来 温度数据，由于发送数据太频繁，所以采用以下统计：
根据 传感器（id） 分组，统计各个 传感器（id）的最大温度
发送数据格式：
{"id":"1","temp":35.3,"time":"2020-01-01 10:10:00"}
java
package state;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Properties;
public class S1ValueState {
final static Logger log= LoggerFactory.getLogger(S1ValueState.class);
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 定义kafka相关配置项
Properties pro = new Properties();
pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"consumer-group");
pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
// 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>("kafka2flink", new SimpleStringSchema(), pro);
DataStreamSource<String> dss = env.addSource(fkc);
KeyedStream<Tuple3<String, Double, String>, String> keyBy = dss
.flatMap(new MyFlatMap())
.keyBy(new KeySelector<Tuple3<String, Double, String>, String>() {
@Override
public String getKey(Tuple3<String, Double, String> value) throws Exception {
return value.f0;
}
});
// SingleOutputStreamOperator<Tuple3<String, Double, String>> maxSOSO = keyBy.maxBy(1);
// 通过状态实现 maxBy() 功能
SingleOutputStreamOperator<Tuple3<String, Double, String>> maxSOSO = keyBy.map(new RichMapFunction<Tuple3<String, Double, String>, Tuple3<String, Double, String>>() {
private ValueState<Tuple3<String, Double, String>> vs = null;
@Override
public void open(Configuration parameters) throws Exception {
// 创建 ValueStateDescriptor 对象
ValueStateDescriptor vsd = new ValueStateDescriptor("max_value",Double.class);
// 根据 ValueStateDescriptor 对象生成 ValueState，赋初始值
vs = getRuntimeContext().getState(vsd);
}
@Override
public Tuple3<String, Double, String> map(Tuple3<String, Double, String> data) throws Exception {
// 从 state 中取出数据
Tuple3<String, Double, String> dataVS = vs.value();
Tuple3<String, Double, String> ret = null;
// 如果 state中的数据为空，表示第一次运行，就返回传入的数据，并将传入数据放入 state中
if(dataVS == null){
ret = data;
vs.update(data);
}else{
// 如果传入数据，比state中数据 的 温度大，就更新，并返回该输入的数据
if(data.f1 > dataVS.f1) {
ret = data;
vs.update(data);
}else{// 否则，说明 state中数据 的 温度大，返回 state 中的数据
ret = dataVS;
}
}
return ret;
}
});
maxSOSO.print();
env.execute();
}
public static class MyFlatMap implements FlatMapFunction<String, Tuple3<String,Double,String>> {
@Override
public void flatMap(String data, Collector<Tuple3<String, Double, String>> out) throws Exception {
String id = null;
Double temp = null;
String timeStr = null;
try {
ObjectMapper om = new ObjectMapper();
JsonNode jsonNode = om.readTree(data);
JsonNode idJN = jsonNode.get("id");
JsonNode tempJN = jsonNode.get("temp");
JsonNode timeJN = jsonNode.get("time");
if (idJN != null) {
id = idJN.asText();
}
if (tempJN != null) {
temp = tempJN.asDouble();
}
if (timeJN != null) {
timeStr = timeJN.asText();
}
out.collect(Tuple3.of(id,temp,timeStr));
}catch (Exception e){
log.error("data error:"+data);
}
}
}
}
启动zk
登录 hadoop1，执行下面命令：
/program/bin/zk.sh start
启动kafka
登录 hadoop1，执行下面命令：
/program/bin/kafka.sh start
启动 flink
在 idea 中运行 flink 程序
创建生产者、指定主题
登录 hadoop1，执行下面命令：
/program/kafka_2.12-2.4.1/bin/kafka-console-producer.sh --broker-list hadoop1:9092,hadoop2:9092,hadoop3:9092 --topic kafka2flink
第一次输入
在kafka控制输入下面内容：
{"id":"1","temp":35.3,"time":"2020-01-01 10:10:00"}
执行结果
在 idea 控制台中打印结果如下：
1> {"id":"1","temp":35.3,"time":"2020-01-01 10:10:00"}
解释：因为只有一条数据，所以就输出该条数据
第二次输入
在kafka控制输入下面内容：
{"id":"1","temp":33.1,"time":"2020-01-01 10:12:00"}
执行结果
在 idea 控制台中打印结果如下：
1> {"id":"1","temp":35.3,"time":"2020-01-01 10:10:00"}
解释： 由于第二次输入的数据，id 与第一次输入相同，分为一组，但 temp 较小，所以输出结果是 第一次的数据
第三次输入
在kafka控制输入下面内容：
{"id":"1","temp":42.5,"time":"2020-01-01 10:14:00"}
执行结果
1> {"id":"1","temp":42.5,"time":"2020-01-01 10:14:00"}
解释：由于第三次输入数据，id 与之前输入相同，分为一组，但 temp 最大，所以输出结果是 本次的数据