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

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
简介: 大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1:消费Kafka写到Redis

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

目前已经更新到了:

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");
    }

}

启动Kafka

启动Redis

运行代码

写入数据

查看结果

目录
打赏
0
0
0
0
100
分享
相关文章
云消息队列 Kafka 版 V3 系列荣获信通院“云原生技术创新标杆案例”
2024 年 12 月 24 日,由中国信息通信研究院(以下简称“中国信通院”)主办的“2025 中国信通院深度观察报告会:算力互联网分论坛”,在北京隆重召开。本次论坛以“算力互联网 新质生产力”为主题,全面展示中国信通院在算力互联网产业领域的研究、实践与业界共识,与产业先行者共同探索算力互联网产业未来发展的方向。会议公布了“2024 年度云原生与应用现代化标杆案例”评选结果,“云消息队列 Kafka 版 V3 系列”荣获“云原生技术创新标杆案例”。
|
20天前
|
NoSQL与Redis配置与优化
通过合理配置和优化Redis,可以显著提高其性能和可靠性。选择合适的数据结构、优化内存使用、合理设置持久化策略、使用Pipeline批量执行命令、以及采用分布式集群方案,都是提升Redis性能的重要手段。同时,定期监控和维护Redis实例,及时调整配置,能够确保系统的稳定运行。希望本文对您在Redis的配置与优化方面有所帮助。
60 23
大数据-131 - Flink CEP 案例:检测交易活跃用户、超时未交付
大数据-131 - Flink CEP 案例:检测交易活跃用户、超时未交付
106 0
探索Flink动态CEP:杭州银行的实战案例
本文由杭州银行大数据工程师唐占峰、欧阳武林撰写,介绍Flink动态CEP的定义、应用场景、技术实现及使用方式。Flink动态CEP是基于Flink的复杂事件处理库,支持在不重启服务的情况下动态更新规则,适应快速变化的业务需求。文章详细阐述了其在反洗钱、反欺诈和实时营销等金融领域的应用,并展示了某金融机构的实际应用案例。通过动态CEP,用户可以实时调整规则,提高系统的灵活性和响应速度,降低维护成本。文中还提供了具体的代码示例和技术细节,帮助读者理解和使用Flink动态CEP。
445 2
探索Flink动态CEP:杭州银行的实战案例
|
21天前
|
NoSQL与Redis配置与优化
通过合理配置和优化Redis,可以显著提高其性能和可靠性。选择合适的数据结构、优化内存使用、合理设置持久化策略、使用Pipeline批量执行命令、以及采用分布式集群方案,都是提升Redis性能的重要手段。
39 7
Flink 三种时间窗口、窗口处理函数使用及案例
Flink 是处理无界数据流的强大工具,提供了丰富的窗口机制。本文介绍了三种时间窗口(滚动窗口、滑动窗口和会话窗口)及其使用方法,包括时间窗口的概念、窗口处理函数的使用和实际案例。通过这些机制,可以灵活地对数据流进行分析和计算,满足不同的业务需求。
208 27
如何在Flink on YARN环境中配置Debezium CDC 3.0,以实现实时捕获数据库变更事件并将其传输到Flink进行处理
本文介绍了如何在Flink on YARN环境中配置Debezium CDC 3.0,以实现实时捕获数据库变更事件并将其传输到Flink进行处理。主要内容包括安装Debezium、配置Kafka Connect、创建Flink任务以及启动任务的具体步骤,为构建实时数据管道提供了详细指导。
183 9
2024Mysql And Redis基础与进阶操作系列(1)作者——LJS[含MySQL的下载、安装、配置详解步骤及报错对应解决方法]
Mysql And Redis基础与进阶操作系列(1)之[MySQL的下载、安装、配置详解步骤及报错对应解决方法]
|
3月前
|
Redis 配置
10月更文挑战第14天
53 1

热门文章

最新文章

AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等