说明
flatMap 算子的功能相当于 map算子 + filter算子 ：
- 输入都是数据流中的每个数据
- 输出可以是 零个、一个 或 多个元素
如下图，输入是包含圆形或正方形的列表，flatMap 过滤掉圆形，正方形列表被展平，以单个元素的形式输出。
应用场景
- 将数据展平
- 支持
map()算子功能 - 实现过滤
- 该算子可以替代
map()和filter() - 用
try catch捕获异常，即使报错，仍然可继续运行。map()算子则不行
api
filter() 方法
通过 source 调用 flatMap() 方法，如下：
public <R> SingleOutputStreamOperator<R> flatMap(FlatMapFunction<T, R> flatMapper)
需要传入 FlatMapFunction 类对象
FlatMapFunction
源码如下：
// 通过注解可知，支持lambda表达式
@FunctionalInterface
@Public
public interface FlatMapFunction<T, O> extends Function, Serializable {
void flatMap(T var1, Collector<O> var2) throws Exception;
}
泛型解释：
- T：传入的数据类型
- O：输出的数据类型
flatMap() 方法形参：
Collector<O> var2：通过该对象的collect()方法，收集需要的数据，丢弃不满足条件的数据
案例
从kafka接收数据，一条数据 的格式是 数组，内部元素是 json格式字符串：
[
{"name":"李雷","url":"/jianpan.html","time":"2020-09-01 10:00:00"},
{"name":"","url":"/jianpan.html","time":"2020-09-01 10:00:00"}
]
为了便于后续操作处理，需要将一条数据 拆分，并转成 java对象（UserLog），也要 判断用户名是否为空，为空就丢弃
主类
package flatmap;
import filter.UserLog;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JavaType;
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.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.apache.log4j.Logger;
import java.util.List;
import java.util.Properties;
public class StdFlatMap {
static final Logger LOGGER= Logger.getLogger(StdFlatMap.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);
SingleOutputStreamOperator<UserLog> soso = dss.flatMap(new MyFlatMap());
soso.print();
env.execute();
}
public static class MyFlatMap implements FlatMapFunction<String, UserLog> {
@Override
public void flatMap(String s, Collector<UserLog> collector) throws Exception {
ObjectMapper mapper = new ObjectMapper();
JavaType javaType = mapper.getTypeFactory().constructParametricType(List.class, UserLog.class);
// 捕获异常后，可继续运行。map() 算子则不行
try {
List<UserLog> list=mapper.readValue(s,javaType);
for(UserLog item:list){
String name = item.getName();
if(name != null && !"".equals(name)){
collector.collect(item);
}
}
} catch (Exception e) {
LOGGER.error("数据格式不正确，数据："+s,e);
}
}
}
}
启动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
然后输入：
[{"name":"李雷","url":"/jianpan.html","time":"2020-09-01 10:00:00"},{"name":"","url":"/jianpan.html","time":"2020-09-01 10:00:00"},{"name":"lili","url":"/dior.html","time":"2020-09-01 10:00:00"}]
执行结果
在 idea 控制台中打印结果如下：
5> UserLog{name='����', url='/jianpan.html', time=Tue Sep 01 18:00:00 CST 2020}
5> UserLog{name='lili', url='/dior.html', time=Tue Sep 01 18:00:00 CST 2020}
传输是 json数组，返回多个对象，并且过滤 name 是空的数据
参考：
https://cloud.tencent.com/developer/article/1559993