说明

根据规则 对数据进行 筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。

api

filter() 方法

通过 source 调用 filter() 方法，filter() 方法如下：

public SingleOutputStreamOperator<T> filter(FilterFunction<T> filter)

需要传入 FilterFunction 类对象

FilterFunction

源码如下：

// 通过注解可知，支持lambda表达式
@FunctionalInterface
@Public
public interface FilterFunction<T> extends Function, Serializable {
    boolean filter(T var1) throws Exception;
}

案例

上接 flink1.12.x Transformation-map算子，如果用户名是 null 或者是 空字符串，就无法分析该用户的行为喜好，就丢弃该数据

主类

package filter;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
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.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

public class TestFilter {
    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> map = dss.map(new MyMap());

        SingleOutputStreamOperator<UserLog> filter = map.filter(new MyFilter());

        filter.print();

        env.execute();
    }

    /**
     * 第一个泛型，表示原数据的类型
     * 第二个泛型，表示转换后的数据类型
     */
    public static class MyMap implements MapFunction<String, UserLog>{

        @Override
        public UserLog map(String s) throws Exception {
            ObjectMapper mapper = new ObjectMapper();
            UserLog result = mapper.readValue(s, UserLog.class);
            return result;
        }
    }
    // 如果用户名是空，就丢弃
    public static class MyFilter implements FilterFunction<UserLog>{

        @Override
        public boolean filter(UserLog userLog) throws Exception {
            String name = userLog.getName();
            if(name == null || "".equals(name)) {
                return false;
            }else{
                return true;
            }
        }
    }
}

kafka

在生产者中，输入下面内容：

{"name":"李雷","url":"/jianpan.html","time":"2020-09-01 10:00:00"}

{"name":"","url":"/jianpan.html","time":"2020-09-01 10:00:00"}

执行结果

在 idea 控制台中打印结果如下：

5> UserLog{name='����', url='/jianpan.html', time=Tue Sep 01 18:00:00 CST 2020}

说明 name 是空字符串的数据，已经被丢弃

原文出处：/show_1IX435YAk9Li.html