一、引言
咱两书接上回，上一篇文章主要介绍了DataStream API一些基本的使用，主要是针对单数据流的场景下，但是在实际的流处理场景中，常常需要对多个数据流进行合并、拆分等操作，以满足复杂的业务需求。Flink 的 DataStream API 提供了一系列强大的多流转换算子，如 union、connect 和 split 等，下面我们来详细了解一下它们的功能和用法。
二、多流转换
2.1 union 算子
union 算子的功能非常直接，就是将多个类型相同的 DataStream 合并成一个新的 DataStream 。它适用于需要将多个来源相同或相似的数据合并到一起进行统一处理的场景。例如，在电商场景中，我们可能有来自不同地区的订单流，希望将它们合并起来进行整体的销售统计分析。
下面通过一个简单的代码示例来展示 union 算子的使用：
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class UnionExample {public static void main(String[] args) throws Exception {// 创建执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 定义两个数据流DataStream<Integer> stream1 = env.fromElements(1, 2, 3);DataStream<Integer> stream2 = env.fromElements(4, 5, 6);// 使用union合并两个数据流DataStream<Integer> unionStream = stream1.union(stream2);// 打印合并后的数据流unionStream.print();// 执行任务env.execute("Union Example");}
}在这个示例中，我们首先创建了两个包含整数的数据流 stream1 和 stream2，然后使用 union 算子将它们合并成一个新的数据流 unionStream 。最后，通过 print 方法将合并后的数据流输出到控制台。合并后的数据特点是，新数据流包含了原始两个数据流中的所有元素，元素的顺序按照它们在原始数据流中的顺序依次排列。
2.2 connect 算子
connect 算子与 union 算子不同，它主要用于连接两个类型可以不同的数据流，并将它们合并成一个 ConnectedStreams 对象。这个对象允许我们在后续处理中分别对两个流的数据进行操作，从而保留流之间的差异。这种特性在需要对不同类型的数据进行关联处理，但又要保持数据类型独立性的场景中非常有用。例如，在一个监控系统中，我们可能有一个数据流表示设备的状态信息（如温度、湿度等数值型数据），另一个数据流表示设备的日志信息（字符串类型），我们可以使用 connect 算子将这两个流连接起来，以便在后续处理中综合分析设备状态和日志。
以下是使用 connect 算子的代码示例：
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;public class ConnectExample {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 定义两个不同类型的数据流DataStream<Integer> stream1 = env.fromElements(1, 2, 3);DataStream<String> stream2 = env.fromElements("a", "b", "c");// 使用connect连接两个数据流ConnectedStreams<Integer, String> connectedStreams = stream1.connect(stream2);// 对连接后的流进行处理DataStream<Object> resultStream = connectedStreams.map(new CoMapFunction<Integer, String, Object>() {@Overridepublic Object map1(Integer value) {return "Integer: " + value;}@Overridepublic Object map2(String value) {return "String: " + value;}});resultStream.print();env.execute("Connect Example");}
}在这个示例中，我们创建了一个整数类型的数据流 stream1 和一个字符串类型的数据流 stream2 。通过 connect 算子将它们连接成 ConnectedStreams 对象，然后使用 CoMapFunction 对连接后的流进行处理。CoMapFunction 包含两个方法 map1 和 map2 ，分别用于处理来自不同流的数据。最终，将处理结果输出到控制台。
2.3 split 算子
split 算子的作用与合并相反，它用于将一个 DataStream 根据某些条件拆分成多个 DataStream 。在实际应用中，我们常常需要根据数据的不同特征对数据流进行分类处理，split 算子就可以帮助我们实现这一需求。比如，在一个电商订单处理系统中，我们可以根据订单金额的大小将订单流拆分成小额订单流和大额订单流，以便对不同金额范围的订单进行不同的处理策略。
下面是使用 split 算子的代码实现：
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.ArrayList;
import java.util.List;public class SplitExample {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 定义一个包含整数的数据流DataStream<Integer> stream = env.fromElements(1, 20, 3, 40, 5, 60);// 使用split算子拆分数据流SplitStream<Integer> splitStream = stream.split(new OutputSelector<Integer>() {@Overridepublic Iterable<String> select(Integer value) {List<String> output = new ArrayList<>();if (value > 10) {output.add("large");} else {output.add("small");}return output;}});// 获取拆分后的两个数据流DataStream<Integer> largeStream = splitStream.select("large");DataStream<Integer> smallStream = splitStream.select("small");// 打印拆分后的数据流largeStream.print("Large Stream: ");smallStream.print("Small Stream: ");env.execute("Split Example");}
}在这个示例中，我们定义了一个包含整数的数据流 stream 。通过 split 算子和自定义的 OutputSelector，根据数值大小将数据流拆分成两个子流：largeStream 包含大于 10 的整数，smallStream 包含小于等于 10 的整数。最后，分别将这两个子流输出到控制台，并加上相应的标识以便区分。
三、数据下沉（Sink）
在流处理应用中，将处理后的结果数据输出到各种存储介质是非常重要的一环。Flink 提供了丰富的数据下沉（Sink）操作，支持将数据写入文件、Kafka、数据库等多种存储系统，以满足不同场景下的数据持久化和后续处理需求。
3.1 写入文件
Flink 提供了多种将数据写入文件的方法，其中常用的是 writeAsText 方法，它将数据流中的元素以文本形式写入文件。例如，我们可以将处理后的订单数据写入文件，以便后续进行数据分析或存档。
以下是一个完整的代码示例，展示如何从流数据到文件写入的全流程：
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class FileSinkExample {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 定义一个包含字符串的数据流DataStream<String> stream = env.fromElements("apple", "banana", "cherry");// 将数据流写入文件，路径为output.txtstream.writeAsText("output.txt");// 执行任务env.execute("File Sink Example");}
}在上述代码中，我们首先创建了一个包含水果名称的数据流 stream 。然后，使用 writeAsText 方法将数据流中的元素写入名为 output.txt 的文件中。在实际应用中，需要注意文件写入模式和配置，writeAsText 默认使用追加模式写入文件，如果文件已存在，新的数据会追加到文件末尾。还可以通过配置参数来指定写入模式（如覆盖模式）、缓冲区大小等。例如，通过设置 env.setBufferTimeout (1000) 可以调整缓冲区的超时时间，当缓冲区数据达到一定时间（这里是 1 秒）或大小限制时，会被写入文件。
3.2 写入 Kafka
Flink 与 Kafka 的集成非常紧密，通过 FlinkKafkaProducer 可以方便地将处理结果写入 Kafka 主题。这种方式在构建实时数据管道时非常常见，处理后的数据可以被其他系统从 Kafka 中消费，实现数据的共享和进一步处理。其原理是通过配置 Kafka 集群的地址、目标主题等参数，FlinkKafkaProducer 将 Flink 处理后的数据流转换为 Kafka 可接受的消息格式，并发送到指定的 Kafka 主题中 。
以下是连接 Kafka、配置参数以及数据写入 Kafka 的代码示例：
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;import java.util.Properties;public class KafkaSinkExample {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 定义一个包含字符串的数据流DataStream<String> stream = env.fromElements("message1", "message2", "message3");// 配置Kafka参数Properties properties = new Properties();properties.setProperty("bootstrap.servers", "localhost:9092");properties.setProperty("acks", "all");// 创建FlinkKafkaProducer，将数据写入Kafka的test-topic主题FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>("test-topic",new SimpleStringSchema(),properties);// 将数据流写入Kafkastream.addSink(kafkaProducer);// 执行任务env.execute("Kafka Sink Example");}
}在这个示例中，我们首先创建了一个包含消息的数据流 stream 。然后，配置了 Kafka 的连接参数，包括 Kafka 集群地址（bootstrap.servers）和 acks 参数（这里设置为 "all"，表示等待所有副本确认写入成功，以确保数据的可靠性）。接着，创建了 FlinkKafkaProducer 对象，指定了要写入的 Kafka 主题（test-topic）和序列化器（SimpleStringSchema，用于将字符串数据转换为 Kafka 消息格式）。最后，通过 addSink 方法将数据流写入 Kafka。
3.3 写入数据库（以 Redis 为例）
以 Redis 为例，将 Flink 处理后的数据写入 Redis 可以实现数据的快速存储和查询，适用于对数据读写性能要求较高的场景。连接 Redis 数据库的步骤主要包括引入相关依赖、创建 Jedis 连接池配置以及在 Flink 中自定义 Sink 函数来实现数据写入。
以下是自定义 RichSinkFunction 实现数据写入 Redis 的代码：
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPoolConfig;public class RedisSinkExample {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 定义一个包含字符串的数据流DataStream<String> stream = env.fromElements("key1:value1", "key2:value2", "key3:value3");// 将数据流写入Redisstream.addSink(new RedisSink());// 执行任务env.execute("Redis Sink Example");}public static class RedisSink extends RichSinkFunction<String> {private transient Jedis jedis;@Overridepublic void open(org.apache.flink.configuration.Configuration parameters) throws Exception {super.open(parameters);// 创建Jedis连接池配置JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();// 初始化Jedis连接，连接到本地Redis服务jedis = new Jedis("localhost", 6379);}@Overridepublic void invoke(String value, Context context) throws Exception {// 按冒号分割字符串，得到键值对String[] parts = value.split(":", 2);String key = parts[0];String data = parts[1];// 将数据写入Redisjedis.set(key, data);}@Overridepublic void close() throws Exception {super.close();// 关闭Jedis连接if (jedis != null) {jedis.close();}}}
}在这段代码中，我们定义了一个自定义的 RichSinkFunction，即 RedisSink 。在 open 方法中，创建了 Jedis 连接池配置，并初始化了 Jedis 连接，连接到本地的 Redis 服务（地址为localhost，端口为 6379）。在 invoke 方法中，对输入的字符串进行处理，将其按冒号分割为键值对，然后使用 jedis.set 方法将数据写入 Redis。在 close 方法中，关闭 Jedis 连接，释放资源。通过这种方式，Flink 处理后的数据流中的数据就可以成功写入 Redis 数据库。
四、总结
Flink DataStream API 的多流转换操作，如 union、connect 和 split 等算子，为我们提供了强大的工具，使他们能够灵活地处理多个数据流之间的复杂关系。通过这些算子，我们可以将不同来源的数据进行合并，对不同类型的数据进行关联处理，以及根据数据特征进行分类拆分，从而满足各种复杂业务场景下的流处理需求。
转换为3dtiles的免费工具！)
开发商城系统)
