接上文：一文说清flink从编码到部署上线
之前写了kafka source，现在补充kafka sink。完善kafka相关操作。
环境说明：MySQL：5.7；flink：1.14.0；hadoop：3.0.0；操作系统：CentOS 7.6；JDK：1.8.0_401；kafka_2.12-2.5.0。
1. kafka 创建 topic
topic：rv-test-sink。
2.添加依赖
<!--flink cdc kafka 相关依赖-->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.11</artifactId>
<version>${flink.version}</version>
</dependency>
3.创建运行环境
package com.zl.utils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import java.time.Duration;
import java.time.ZoneOffset;
import java.util.concurrent.TimeUnit;
/**
* EnvUtil
* @description:
*/
public class EnvUtil {
/**
* 设置flink执行环境
* @param parallelism 并行度
*/
public static StreamExecutionEnvironment setFlinkEnv(int parallelism) {
// System.setProperty("HADOOP_USER_NAME", "用户名") 对应的是 hdfs文件系统目录下的路径：/user/用户名的文件夹名，本文为root
System.setProperty("HADOOP_USER_NAME", "root");
Configuration conf = new Configuration();
conf.setInteger("rest.port", 1000);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
if (parallelism >0 ){
//设置并行度
env.setParallelism(parallelism);
} else {
env.setParallelism(1);// 默认1
}
// 添加重启机制
// env.setRestartStrategy(RestartStrategies.fixedDelayRestart(50, Time.minutes(6)));
// 没有这个配置，会导致“Flink 任务没报错，但是无法同步数据到doris”。
// 启动checkpoint,设置模式为精确一次 (这是默认值)，10*60*1000=60000
env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);
//rocksdb状态后端,启用增量checkpoint
env.setStateBackend(new EmbeddedRocksDBStateBackend(true));
//设置checkpoint路径
CheckpointConfig checkpointConfig = env.getCheckpointConfig();
// 同一时间只允许一个 checkpoint 进行(默认)
checkpointConfig.setMaxConcurrentCheckpoints(1);
//最小间隔，10*60*1000=60000
checkpointConfig.setMinPauseBetweenCheckpoints(60000);
// 取消任务后,checkpoint仍然保存
checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//checkpoint容忍失败的次数
checkpointConfig.setTolerableCheckpointFailureNumber(5);
//checkpoint超时时间 默认10分钟
checkpointConfig.setCheckpointTimeout(TimeUnit.MINUTES.toMillis(10));
//禁用operator chain(方便排查反压)
env.disableOperatorChaining();
return env;
}
public static StreamTableEnvironment getFlinkTenv(StreamExecutionEnvironment env) {
StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
//设置时区 东八
tenv.getConfig().setLocalTimeZone(ZoneOffset.ofHours(8));
Configuration configuration = tenv.getConfig().getConfiguration();
// 开启miniBatch
configuration.setString("table.exec.mini-batch.enabled", "true");
// 批量输出的间隔时间
configuration.setString("table.exec.mini-batch.allow-latency", "5 s");
// 防止OOM设置每个批次最多缓存数据的条数，可以设为2万条
configuration.setString("table.exec.mini-batch.size", "20000");
// 开启LocalGlobal
configuration.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE");
//设置TTL API指定
tenv.getConfig().setIdleStateRetention(Duration.ofHours(25));
return tenv;
}
}
4.核心代码
package com.zl.kafka;
import com.alibaba.fastjson.JSONObject;
import com.zl.utils.EnvUtil;
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 javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;
import java.util.Properties;
public class KafkaExampleSink {
public static void main(String[] args) throws Exception {
// 配置运行环境，并行度1
StreamExecutionEnvironment env = EnvUtil.setFlinkEnv(1);
// 程序间隔离，每个程序单独设置
env.getCheckpointConfig().setCheckpointStorage("hdfs://10.86.97.191:9000/flinktest/KafkaExampleSink");
/// ===== 构造kafka sink =====
// 相关参数配置可以参考下面这两个文档：①https://cloud.tencent.com/developer/article/2089393
// ②https://www.bilibili.com/opus/819228616166473783
// kafka配置
Properties prop = new Properties();
prop.setProperty("bootstrap.servers", "10.86.97.21:9092,10.86.97.21:9093,10.86.97.21:9094");
// 当设置为“true”时，生产者将确保流中只写入每条消息的一个副本。
prop.setProperty("enable.idempotence", "true");
// 指定了生产者在接收到服务器相应之前可以发送多个消息，值越高，占用的内存越大，
// 当然也可以提升吞吐量，发生错误时，可能会造成数据的发送顺序改变，其默认值是5.
prop.setProperty("max.in.flight.requests.per.connection", "5");
prop.setProperty("acks", "all");
// 在kafka中消息发送失败时，指定生产者可以重发消息的次数，默认情况下，
// 生产者在每次重试之间默认等待100ms，可以通过参数retey.backoff.ms参数来改变这个时间间隔。retries的缺省值：0.
prop.setProperty("retries", "5");
// 事务超时时间
prop.setProperty("transaction.timeout.ms", 15 * 60 * 1000 + "");
String topic = "rv-test-sink";
FlinkKafkaProducer<String> flinkKafkaProducer = new FlinkKafkaProducer<String>(
topic,// topic
new KafkaSerializationSchema<String>() {
@Override
public ProducerRecord<byte[], byte[]> serialize(String s, @Nullable Long aLong) {
return new ProducerRecord<>(topic, s.getBytes(StandardCharsets.UTF_8));
}
},
prop,
FlinkKafkaProducer.Semantic.EXACTLY_ONCE
);
/// ===== 构造模拟数据 =====
JSONObject rvJsonObject = new JSONObject();
rvJsonObject.put("dt","2024-12-20");// 日期取当天
rvJsonObject.put("uuid","data-stream-1");
rvJsonObject.put("report_time",1733881971621L);
String mockJson = JSONObject.toJSONString(rvJsonObject);
/// ===== sink kafka =====
env.fromElements(mockJson).addSink(flinkKafkaProducer).setParallelism(3).name("kafka-sink").uid("kafka-sink");
env.execute("kafka-sink-job");
}// main
}
5.运行
由于不是持续输入流，运行完会结束。
sink到kafka的数据如下：
6.完整代码
完整代码见：完整代码
