大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型配置与使用附带案例1：消费Kafka写到Redis

news2024/11/15 13:56:40

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

Hadoop（已更完）
HDFS（已更完）
MapReduce（已更完）
Hive（已更完）
Flume（已更完）
Sqoop（已更完）
Zookeeper（已更完）
HBase（已更完）
Redis （已更完）
Kafka（已更完）
Spark（已更完）
Flink（正在更新！）

章节内容

上节我们完成了如下的内容：

Flink DataStream Transformation
FlatMap Window Aggregations Reduce 等等等函数

在这里插入图片描述

Sink

Flink 的 Sink 是指数据流处理过程中最终输出数据的组件。在 Apache Flink 中，数据流从 Source 读取后经过一系列的转换操作，最后会被写入到 Sink 中。Sink 是 Flink 流式处理应用的终点，决定了处理后的数据如何保存或传输。

基本概念

Flink 的 Sink 是用来将流处理的数据写入外部存储系统的，比如数据库、文件系统、消息队列等。Sink 接口提供了一种灵活的方式来定义数据的输出格式和存储目标。Flink 提供了多个内置的 Sink 连接器，用户也可以根据需求自定义 Sink。

常见类型

Flink 提供了多种内置的 Sink，可以将数据输出到多种不同的系统中。以下是一些常见的 Flink Sink：

File Sink：将数据输出到文件系统，支持多种文件格式，如文本文件、CSV、Parquet 等。
Kafka Sink：将数据输出到 Kafka 主题，用于构建流式数据管道。
Elasticsearch Sink：将数据写入 Elasticsearch 索引，适用于实时数据搜索和分析。
JDBC Sink：将数据写入关系型数据库，如 MySQL、PostgreSQL 等。
HDFS Sink：将数据存储在 Hadoop 分布式文件系统中，适用于大规模数据的长期存储。
Cassandra Sink：将数据写入 Cassandra 数据库，适用于大规模的 NoSQL 数据存储

配置与使用

要在 Flink 应用中使用 Sink，需要通过 DataStream 的 addSink 方法来配置和添加 Sink。例如，将数据写入 Kafka 的简单配置如下：

DataStream<String> dataStream = // 数据处理逻辑
dataStream.addSink(new FlinkKafkaProducer<>(
    "localhost:9092",         // Kafka broker 地址
    "output-topic",           // 输出的 Kafka 主题
    new SimpleStringSchema()   // 数据序列化格式
));

同样，配置 JDBC Sink 的方式如下：

dataStream.addSink(JdbcSink.sink(
    "INSERT INTO my_table (column1, column2) VALUES (?, ?)",
    (statement, value) -> {
        statement.setString(1, value.f0);
        statement.setInt(2, value.f1);
    },
    JdbcExecutionOptions.builder()
        .withBatchSize(1000)
        .withBatchIntervalMs(200)
        .build(),
    new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:mysql://localhost:3306/mydb")
        .withDriverName("com.mysql.jdbc.Driver")
        .withUsername("user")
        .withPassword("password")
        .build()
));

自定义 Sink

除了使用内置的 Sink，Flink 还允许开发者实现自定义 Sink。通过实现 SinkFunction 接口或扩展 RichSinkFunction 类，开发者可以定义自己所需的 Sink。自定义 Sink 通常用于需要特殊处理或集成尚不支持的外部系统。

例如，自定义一个简单的控制台打印 Sink：

public class PrintSinkFunction<T> extends RichSinkFunction<T> {
    @Override
    public void invoke(T value, Context context) {
        System.out.println(value);
    }
}

Sink 的容错机制

Flink 提供了精确一次 (Exactly-Once) 和至少一次 (At-Least-Once) 的容错语义，具体取决于 Sink 的类型及其配置。例如，Kafka Sink 通常支持精确一次语义，而某些文件系统 Sink 可能只支持至少一次语义。通过启用 Flink 的 Checkpointing 机制，Sink 可以在发生故障时从最近的检查点恢复，从而保证数据的一致性。

Sink 的并行度

Flink 的 Sink 通常是并行的，默认情况下与上游操作的并行度一致。用户可以通过 setParallelism 方法来手动调整 Sink 的并行度。注意，对于一些 Sink，如文件系统 Sink，并行度越高，生成的文件数也越多。

生命周期

Flink 的 Sink 在执行时会经历以下几个阶段：

打开 (open)：初始化资源，如数据库连接、文件句柄等。
写入 (invoke)：将每一条数据写入目标存储系统。
关闭 (close)：关闭资源，确保数据完整写入和资源的正确释放。

简单示例

以下是一个将处理后的数据流写入文本文件的完整示例：

DataStream<String> dataStream = // 数据处理逻辑
StreamingFileSink<String> sink = StreamingFileSink
    .forRowFormat(new Path("/output/path"), new SimpleStringEncoder<String>("UTF-8"))
    .build();

dataStream.addSink(sink);

案例1：数据写入Redis

添加依赖

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-redis_2.11</artifactId>
  <version>1.1.5</version>
</dependency>

编写代码

消费Kafka 计算之后写入到 Redis中。
Source(Kafka) -> Sink(Redis)

package icu.wzk;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.util.Properties;

public class StreamFromKafka {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 配置信息
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "h121.wzk.icu:9092");

        // Kafka
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
                "flink_test",
                new SimpleStringSchema(),
                properties
        );
        DataStreamSource<String> data = env.getJavaEnv().addSource(consumer);

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = data
                .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                        String[] words = value.split(" ");
                        for (String word: words) {
                            out.collect(new Tuple2<>(word, 1));
                        }
                    }
                });
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = wordAndOne
                .keyBy(new KeySelector<Tuple2<String, Integer>, Object>() {
                    @Override
                    public Object getKey(Tuple2<String, Integer> value) throws Exception {
                        return value.f0;
                    }
                })
                .sum(1);
        result.print();
        env.execute("StreamFromKafka");
    }

}