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

本文主要是介绍大数据-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> {@Overridepublic 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");// KafkaFlinkKafkaConsumer<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>>() {@Overridepublic 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>() {@Overridepublic Object getKey(Tuple2<String, Integer> value) throws Exception {return value.f0;}}).sum(1);result.print();env.execute("StreamFromKafka");}}

启动Kafka

在这里插入图片描述

启动Redis

在这里插入图片描述

运行代码

在这里插入图片描述

写入数据

在这里插入图片描述

查看结果

在这里插入图片描述

这篇关于大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1:消费Kafka写到Redis的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1133117

相关文章

大模型研发全揭秘:客服工单数据标注的完整攻略

在人工智能(AI)领域,数据标注是模型训练过程中至关重要的一步。无论你是新手还是有经验的从业者,掌握数据标注的技术细节和常见问题的解决方案都能为你的AI项目增添不少价值。在电信运营商的客服系统中,工单数据是客户问题和解决方案的重要记录。通过对这些工单数据进行有效标注,不仅能够帮助提升客服自动化系统的智能化水平,还能优化客户服务流程,提高客户满意度。本文将详细介绍如何在电信运营商客服工单的背景下进行

基于MySQL Binlog的Elasticsearch数据同步实践

一、为什么要做 随着马蜂窝的逐渐发展,我们的业务数据越来越多,单纯使用 MySQL 已经不能满足我们的数据查询需求,例如对于商品、订单等数据的多维度检索。 使用 Elasticsearch 存储业务数据可以很好的解决我们业务中的搜索需求。而数据进行异构存储后,随之而来的就是数据同步的问题。 二、现有方法及问题 对于数据同步,我们目前的解决方案是建立数据中间表。把需要检索的业务数据,统一放到一张M

Zookeeper安装和配置说明

一、Zookeeper的搭建方式 Zookeeper安装方式有三种,单机模式和集群模式以及伪集群模式。 ■ 单机模式:Zookeeper只运行在一台服务器上,适合测试环境; ■ 伪集群模式:就是在一台物理机上运行多个Zookeeper 实例; ■ 集群模式:Zookeeper运行于一个集群上,适合生产环境,这个计算机集群被称为一个“集合体”(ensemble) Zookeeper通过复制来实现

CentOS7安装配置mysql5.7 tar免安装版

一、CentOS7.4系统自带mariadb # 查看系统自带的Mariadb[root@localhost~]# rpm -qa|grep mariadbmariadb-libs-5.5.44-2.el7.centos.x86_64# 卸载系统自带的Mariadb[root@localhost ~]# rpm -e --nodeps mariadb-libs-5.5.44-2.el7

关于数据埋点,你需要了解这些基本知识

产品汪每天都在和数据打交道,你知道数据来自哪里吗? 移动app端内的用户行为数据大多来自埋点,了解一些埋点知识,能和数据分析师、技术侃大山,参与到前期的数据采集,更重要是让最终的埋点数据能为我所用,否则可怜巴巴等上几个月是常有的事。   埋点类型 根据埋点方式,可以区分为: 手动埋点半自动埋点全自动埋点 秉承“任何事物都有两面性”的道理:自动程度高的,能解决通用统计,便于统一化管理,但个性化定

中文分词jieba库的使用与实景应用(一)

知识星球:https://articles.zsxq.com/id_fxvgc803qmr2.html 目录 一.定义: 精确模式(默认模式): 全模式: 搜索引擎模式: paddle 模式(基于深度学习的分词模式): 二 自定义词典 三.文本解析   调整词出现的频率 四. 关键词提取 A. 基于TF-IDF算法的关键词提取 B. 基于TextRank算法的关键词提取

Hadoop企业开发案例调优场景

需求 (1)需求:从1G数据中,统计每个单词出现次数。服务器3台,每台配置4G内存,4核CPU,4线程。 (2)需求分析: 1G / 128m = 8个MapTask;1个ReduceTask;1个mrAppMaster 平均每个节点运行10个 / 3台 ≈ 3个任务(4    3    3) HDFS参数调优 (1)修改:hadoop-env.sh export HDFS_NAMENOD

使用SecondaryNameNode恢复NameNode的数据

1)需求: NameNode进程挂了并且存储的数据也丢失了,如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2)故障模拟 (1)kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 (2)删除NameNode存储的数据(/opt/module/hadoop-3.1.4/data/tmp/dfs/na

异构存储(冷热数据分离)

异构存储主要解决不同的数据,存储在不同类型的硬盘中,达到最佳性能的问题。 异构存储Shell操作 (1)查看当前有哪些存储策略可以用 [lytfly@hadoop102 hadoop-3.1.4]$ hdfs storagepolicies -listPolicies (2)为指定路径(数据存储目录)设置指定的存储策略 hdfs storagepolicies -setStoragePo

Hadoop集群数据均衡之磁盘间数据均衡

生产环境,由于硬盘空间不足,往往需要增加一块硬盘。刚加载的硬盘没有数据时,可以执行磁盘数据均衡命令。(Hadoop3.x新特性) plan后面带的节点的名字必须是已经存在的,并且是需要均衡的节点。 如果节点不存在,会报如下错误: 如果节点只有一个硬盘的话,不会创建均衡计划: (1)生成均衡计划 hdfs diskbalancer -plan hadoop102 (2)执行均衡计划 hd