阅读量:0
在 Flink 中，可以使用 filter() 函数对数据进行过滤。filter() 函数接收一个 FilterFunction 类型的参数，该参数定义了过滤条件。
以下是一个简单的示例，演示如何使用 Flink 对数据流进行过滤：
import org.apache.flink.api.common.functions.FilterFunction; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer; public class FlinkFilterExample { public static void main(String[] args) throws Exception { // 创建 Flink 执行环境 final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 从 Kafka 读取数据 FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("input-topic", new SimpleStringSchema(), properties); DataStream<String> stream = env.addSource(consumer); // 定义过滤条件 FilterFunction<String> filterFunction = new FilterFunction<String>() { @Override public boolean filter(String value) throws Exception { return value.contains("filter"); } }; // 使用 filter 函数过滤数据 DataStream<String> filteredStream = stream.filter(filterFunction); // 将过滤后的数据写入 Kafka FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>("output-topic", new SimpleStringSchema(), properties); filteredStream.addSink(producer); // 执行 Flink 任务 env.execute("Flink Filter Example"); } } 在上述示例中，我们首先从 Kafka 读取数据，然后定义了一个过滤条件，该条件只保留包含 “filter” 的字符串。接下来，我们使用 filter() 函数对数据流进行过滤，并将过滤后的数据写入 Kafka。最后，我们执行 Flink 任务。
