您现在的位置是:首页 >其他 >Flink系列-11、Flink DataStream的Sink网站首页其他
Flink系列-11、Flink DataStream的Sink
版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
官方网址:https://flink.apache.org/
学习资料:https://flink-learning.org.cn/
目录
Flink在批处理中常见的sink
- 基于本地集合的sink(Collection-based-sink)
print
printToErr - 基于文件的sink(File-based-sink)
- 自定义的sink(Custom-based-sink)
- 基于kafka的sink操作
sink到本地集合
package batch.sink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/**
* @author lwh
* @date 2023/5/5
* @description 到本地集合
**/
public class SinkToLocalCollectionDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Integer> source = env.fromElements(1, 2, 3, 4, 5, 6);
// 打印
source.print();
// 带前缀打印
source.print("我是前缀>>>");
// 打印到stderr中
source.printToErr();
// 打印到stderr并带前缀
source.printToErr("我是stderr前缀>>>");
env.execute();
}
}
基于文件的sink
通过writeAsText将数据写出。
支持本地文件和HDFS
package batch.sink;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/**
* @author lwh
* @date 2023/5/5
* @description 基于文件的sink
**/
public class SinkToLocalFileAndHDFSDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Tuple3<Integer, String, Double>> source = env.fromElements(
Tuple3.of(19, "潇潇", 170.50),
Tuple3.of(11, "甜甜", 168.8),
Tuple3.of(16, "刚刚", 178.8),
Tuple3.of(19, "蛋蛋", 179.99)
);
/*
写文件可以设置为并行度为1, 避免产生出来多个文件
*/
// 写出到本地文件
source.writeAsText("data/output/SinkToLocalFileAndHDFSDemo.txt", FileSystem.WriteMode.OVERWRITE)
.setParallelism(1);
// 写出到hdfs
source.writeAsText("hdfs://node1:8020/output/SinkToLocalFileAndHDFSDemo.txt", FileSystem.WriteMode.OVERWRITE)
.setParallelism(1);
// 写出为csv
source.writeAsCsv("data/output/SinkToLocalFileAndHDFSDemo.csv",
FileSystem.WriteMode.OVERWRITE,
"
",
",").setParallelism(1);
env.execute();
}
}
StreamingFileSink
通过上面的代码可以看出,writeAsText已经废弃,推荐使用StreamingFileSink
参见:https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/connectors/streamfile_sink.html
这个连接器提供了一个 Sink 来将分区文件写入到支持 Flink FileSystem 接口的文件系统中。
Streaming File Sink 会将数据写入到桶中。由于输入流可能是无界的,因此每个桶中的数据被划分为多个有限大小的文件。如何分桶是可以配置的,默认使用基于时间的分桶策略,这种策略每个小时创建一个新的桶,桶中包含的文件将记录所有该小时内从流中接收到的数据。
桶目录中的实际输出数据会被划分为多个部分文件(part file),每一个接收桶数据的 Sink Subtask ,至少包含一个部分文件(part file)。额外的部分文件(part file)将根据滚动策略创建,滚动策略是可以配置的。默认的策略是根据文件大小和超时时间来滚动文件。超时时间指打开文件的最长持续时间,以及文件关闭前的最长非活动时间。
使用 StreamingFileSink 时需要启用 Checkpoint ,每次做 Checkpoint 时写入完成。如果 Checkpoint 被禁用,部分文件(part file)将永远处于 ‘in-progress’ 或 ‘pending’ 状态,下游系统无法安全地读取。
由于我们用的是流任务,那么任务会一直持续进行,数据也会持续不断的写出,由于数据是源源不断的产生,那么就需要给数据设立边界,让其完成某个文件数据的写出。不然某个文件会一直处于写入状态中。
那么StreamingFileSink就是一个写出流数据的类
它会将数据分桶(分part)写出到文件中,按照指定规则(时间、文件大小等),完成某一part的写入过程。
比如:每隔1小时或者每当文件大小达到比如1GB的时候,就完成当前文件的写入,将状态标记为Finished,然后开启一个新文件继续写流数据。
数据在写出之前,在Flink内部会按照各个子任务(并行)划分数据桶,每个桶可以包含多个part文件
文件在写的过程中有3个状态:
- In-progress :当前文件正在写入中
- Pending :当处于 In-progress 状态的文件关闭(closed)了,就变为 Pending 状态
- Finished :在成功的 Checkpoint 后,Pending 状态将变为 Finished 状态
文件格式
StreamingFileSink 支持行编码格式和批量编码格式,比如 Apache Parquet 。这两种变体随附了各自的构建器,可以使用以下静态方法创建:
• Row-encoded sink: StreamingFileSink.forRowFormat(basePath, rowEncoder)
一次写入一行数据
• Bulk-encoded sink: StreamingFileSink.forBulkFormat(basePath, bulkWriterFactory)
一次写入一批数据, 如parquet、avro
桶分配逻辑
简单理解:如何划分桶
桶分配逻辑定义了如何将数据结构化为基本输出目录中的子目录
行格式和批量格式都使用 DateTimeBucketAssigner 作为默认的分配器。 默认情况下,DateTimeBucketAssigner 基于系统默认时区每小时创建一个桶,格式如下: yyyy-MM-dd–HH 。日期格式(即桶的大小)和时区都可以手动配置。
我们可以在格式构建器上调用 .withBucketAssigner(assigner) 来自定义 BucketAssigner 。
Flink 有两个内置的 BucketAssigners :
• DateTimeBucketAssigner :默认基于时间的分配器
• BasePathBucketAssigner :将所有部分文件(part file)存储在基本路径中的分配器(单个全局桶)
内置的不满足需求可以自定义实现BucketAssigner
滚动策略
简单理解:啥时候(按时间、按大小等)算完成1个文件的写入。
滚动策略 RollingPolicy 定义了指定的文件在何时关闭(closed)并将其变为 Pending 状态,随后变为 Finished 状态。处于 Pending 状态的文件会在下一次 Checkpoint 时变为 Finished 状态,通过设置 Checkpoint 间隔时间,可以控制部分文件(part file)对下游读取者可用的速度、大小和数量。
Flink 有两个内置的滚动策略:
• DefaultRollingPolicy
核心策略:
- 当没有正在写入的part文件的时候,不工作,
- 当文件达到最大桶大小的时候关闭文件完成写入 (by default 128MB), 可设置
- 当前写入文件写入时长超过默认间隔 (by default 60 sec), 或者 可设置
- 当前文件一定时间内没有写入(by default 60 sec). 可设置
• OnCheckpointRollingPolicy
核心策略:
当进行一次CheckPoint活动的时候,完成当前文件的写入(跟随检查点的节奏走)
内置的不满足需求可以自定义实现RollingPolicy
代码
package batch.sink;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.BasePathBucketAssigner;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy;
/**
* @author lwh
* @date 2023/5/5
* @description StreamingFileSink
**/
public class StreamingFileSinkDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
env.setStateBackend(new FsStateBackend(
"file:///D:\checkpoint"));
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
// Socket Source
DataStreamSource<String> socketTextStream = env.socketTextStream("node1", 9999);
// 泛型指的是处理的数据类型是什么
StreamingFileSink<String> sink = StreamingFileSink.forRowFormat(
new Path("data/output/sink3"), // 文件写出的路径
// 文件写出的序列化器和编码
new SimpleStringEncoder<String>("UTF-8"))
// 桶分配策略
.withBucketAssigner(new BasePathBucketAssigner<String>())
// 文件滚动(完成一次文件写出)的策略
.withRollingPolicy(
OnCheckpointRollingPolicy.build()
).build();
socketTextStream.addSink(sink);
env.execute();
}
}
sink到kafka
示例
数据写出到Kafka中
使用:FlinkKafkaProducer<>(brokerList, topic, new SimpleStringSchema()); 来定义一个kafka的sink对象
开发步骤
- 创建流处理环境
- 设置并行度
- 添加自定义MySql数据源
- 转换元组数据为字符串
- 构建FlinkKafkaProducer
- 添加sink
- 执行任务
package batch.sink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.codehaus.commons.nullanalysis.Nullable;
import java.util.Properties;
/**
* @author lwh
* @date 2023/5/5
* @description sink kafka
**/
public class KafkaSinkDemo {
public static void main(String[] args) throws Exception {
// Env
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> source = env.fromElements(
"Sink to Kafka Test 1",
"Sink to Kafka Test 2",
"Sink to Kafka Test 3",
"Sink to Kafka Test 4",
"Sink to Kafka Test 5",
"Sink to Kafka Test 6"
);
// 使用FlinkKafkaProducer来构建一个Kafka的生产者
String brokerList = "node1:9092,node2:9092,node3:9092";
String topic = "kafkatopic";
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", brokerList);
// 废弃
// FlinkKafkaProducer<String> kafkaSink = new FlinkKafkaProducer<>(brokerList, topic, new SimpleStringSchema());
FlinkKafkaProducer<String> kafkaSink = new FlinkKafkaProducer<>(
topic, // topic
new MyKafkaSerializationSchema(), // 自定义实现kafka的序列化器
properties, // producer的config
FlinkKafkaProducer.Semantic.EXACTLY_ONCE // kafka的一致性选择
);
// 构建kafka的sink
source.addSink(kafkaSink);
env.execute();
}
// 自定义构建kafka序列化
public static class MyKafkaSerializationSchema implements KafkaSerializationSchema<String> {
private String topic = "kafkatopic";
@Override
public ProducerRecord<byte[], byte[]> serialize(String element, @Nullable Long timestamp) {
return new ProducerRecord<byte[], byte[]>(
topic,
element.getBytes()
);
}
/*
ProducerRecord有很多构造, 我们使用的是最基础的, 给定topic并且将数据byte化即可, 即ProducerRecord(String topic, V value)
别的构造还会要求传入如:
- partition号,int类型, 传入啥写哪个分区, 如果不给定的话, 按照key的hash来计算, 如果没有key的话就按照轮询的方式写入kafka各个分区
- key, 数据的key, 用以计算key的hash来计算数据落入哪个分区
- timestamp, 给数据一个指定的时间戳, 如果不设置, 默认以当前系统时间
上面3个都有默认值, 所以我们不需要设置, 有需要设置可以用其它的重载的构造函数, 如:
ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value)
*/
}
}
sink到mysql
示例
加载下列本地集合,导入MySql中
UserInfo(9, "xiaoxiao", "123456", "潇潇")
开发步骤
- 创建流执行环境
- 准备数据
- 添加sink
- 构建自定义Sink,继承自RichSinkFunction
- 重写open方法,获取Connection和PreparedStatement
- 重写invoke方法,执行插入操作
- 重写close方法,关闭连接操作
- 执行任务
package batch.sink;
import entity.UserInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
/**
* @author lwh
* @date 2023/5/5
* @description sink mysql
**/
public class MysqlSinkDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<UserInfo> source = env.fromElements(new UserInfo(9, "xiaoxiao", "123456", "潇潇"));
source.addSink(new MyMySQLSink());
env.execute();
}
public static class MyMySQLSink extends RichSinkFunction<UserInfo> {
private Connection connection = null; // 数据库连接对象
private PreparedStatement ps = null; // ps对象
/*
invoke就是执行的方法, 类似自定义Source中的run
*/
@Override
public void invoke(UserInfo value, Context context) throws Exception {
this.ps.setInt(1, value.getId());
this.ps.setString(2, value.getUsername());
this.ps.setString(3, value.getPassword());
this.ps.setString(4, value.getName());
this.ps.execute();
}
// 实例化的时候执行一次, 适合用来做连接的创建
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
String url = "jdbc:mysql://localhost:3306/flink?useUnicode=true&characterEncoding=utf-8&useSSL=false";
Class.forName("com.mysql.jdbc.Driver");
this.connection = DriverManager.getConnection(url, "root", "123456");
this.ps = connection.prepareStatement("INSERT INTO user VALUES(?,?,?,?);");
}
// 销毁实例的时候执行一次, 适合用来释放使用的资源
@Override
public void close() throws Exception {
super.close();
// 关闭资源
if (this.ps != null) this.ps.close();
if (this.connection != null) this.connection.close();
}
}
}
sink到redis
通过flink操作redis其实我们可以通过传统的redis连接池Jpoools进行redis的相关操作,但是flink提供了专门操作redis的RedisSink,使用起来更方便,而且不用我们考虑性能的问题,接下来将主要介绍RedisSink如何使用。
RedisSink简介
Redis Sink 提供用于向Redis发送数据的接口的类。接收器可以使用三种不同的方法与不同类型的Redis环境进行通信:
- 单Redis服务器
- Redis集群
- Redis Sentinel
注意:本文主要介绍如何创建与单个redis服务器通信的接收器,其他模式请参考flink官网。https://bahir.apache.org/docs/flink/current/flink-streaming-redis/
如何使用Redis Sink?
Redis Sink 核心类是RedisMapper的一个接口,使用时我们要编写自己的redis操作类实现这个接口中的三个方法,如下所示:
- getCommandDescription():设置使用的redis数据结构类型,和key的名词,通过RedisCommand设置数据结构类型
- String getKeyFromData(T data):设置value中的键值对 key的值
- String getValueFromData(T data);设置value中的键值对 value的值
使用RedisCommand设置数据结构类型时和redis结构对应关系。
DataType | Redis Command[Sink] |
---|---|
HASH | HSET |
LIST | RPUSH, LPUSH |
SET | SADD |
PUBSUB | PUBLISH |
STRING | SET |
HYPER_LOG_LOG | PFADD |
SORTED_SET | ZADD |
SORTED_SET | ZREM |
代码
package batch.sink;
import org.apache.flink.api.common.functions.MapFunction;
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.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
/**
* @author lwh
* @date 2023/5/5
* @description
**/
public class RedisSinkDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// Source
DataStreamSource<String> socketTextStream = env.socketTextStream("node1", 9999);
// 将数据转换为Tuple2, 注意, socket 传入的数据需要是kv字符串以空格分隔
SingleOutputStreamOperator<Tuple2<String, String>> map = socketTextStream.map(new MapFunction<String, Tuple2<String, String>>() {
@Override
public Tuple2<String, String> map(String value) throws Exception {
String[] strings = value.split(" ");
return Tuple2.of(strings[0], strings[1]);
}
});
// 构建Redis conf
FlinkJedisPoolConfig redisConf = new FlinkJedisPoolConfig.Builder().setHost("node1").setPort(6379).build();
// 基于自定义的RedisMapper实现来构建RedisSink
map.addSink(new RedisSink<Tuple2<String, String>>(redisConf, new MyRedisMapper()));
env.execute("Redis Sink Demo");
}
/**
* 自定义实现RedisMapper接口
*/
public static class MyRedisMapper implements RedisMapper<Tuple2<String, String>> {
/*
描述要写出的数据类型
*/
@Override
public RedisCommandDescription getCommandDescription() {
return new RedisCommandDescription(RedisCommand.LPUSH);
}
// 设置key
@Override
public String getKeyFromData(Tuple2<String, String> data) {
return data.f0;
}
// 设置value
@Override
public String getValueFromData(Tuple2<String, String> data) {
return data.f1;
}
}
}