流的合并
- union联合:被unioin的流中的数据类型必须一致
- connect连接:合并的两条流的数据类型可以不一致
- connec后,得到的是ConnectedStreams
- 合并后需要根据数据流是否经过keyby分区
- coConnect: 将两条数据流合并为同一数据类型
- keyedConnect
public class Flink09_UnionConnectStream {
public static void main(String[] args) {
//1.创建运行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//默认是最大并行度
env.setParallelism(1);
DataStreamSource<Integer> ds1 = env.fromElements(1, 2, 3, 4, 5, 6, 7);
DataStreamSource<Integer> ds2 = env.fromElements(8, 9);
DataStreamSource<String> ds3 = env.fromElements("a", "b", "c");
DataStream<Integer> unionDs = ds1.union(ds2);
unionDs.print();
//connect
ConnectedStreams<Integer, String> connectDs = ds1.connect(ds3);
//处理
connectDs.process(
new CoProcessFunction<Integer, String, String>() {
@Override
public void processElement1(Integer value, CoProcessFunction<Integer, String, String>.Context ctx, Collector<String> out) throws Exception {
out.collect(value.toString());
}
@Override
public void processElement2(String value, CoProcessFunction<Integer, String, String>.Context ctx, Collector<String> out) throws Exception {
out.collect(value.toUpperCase());
}
}
).print("connect");
try {
env.execute();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
Sink输出算子
目前所使用的大多数Sink, 都是基于2PC的方式来保证状态精确一次性。2PC 即 two face commit, 两阶段提交,该机制的实现必须要开启Flink的检查点。文章来源:https://www.toymoban.com/news/detail-753746.html
- FileSink:fileSink = FileSink.<数据流泛型>forRowFormat(输出路径, 数据流编码器)
- 文件滚动策略 .withRollingPolicy().builder()
- 文件多大滚动.withMaxPartSize(MemorySize.parse(“10m”))
- 多长时间滚动一次 .withRolloverInterval(Duration.ofSeconds(10))
- 多久不活跃滚动 .withInactivityInterval(Duration.ofSeconds(5))
- 目录滚动策略:一般设置为按照天或者小时或者其他时间间隔
- 文件输出配置:可以设置输出文件的前缀和后缀
- 文件滚动策略 .withRollingPolicy().builder()
public class Flink01_FileSink {
public static void main(String[] args) {
//1.创建运行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(2000);
//默认是最大并行度
env.setParallelism(1);
DataStreamSource<Event> ds = Flink06_EventSource.getEventSource(env);
//FileSink
FileSink<String> stringFileSink = FileSink.<String>forRowFormat(new Path("output"),
new SimpleStringEncoder<>())
.withRollingPolicy(//文件滚动策略
DefaultRollingPolicy.builder()
.withMaxPartSize(MemorySize.parse("10m"))//文件多大滚动
.withRolloverInterval(Duration.ofSeconds(10))//多久滚动
.withInactivityInterval(Duration.ofSeconds(5))//多久不活跃滚动
.build()
)
.withBucketAssigner(//目录滚动策略
new DateTimeBucketAssigner<>("yyyy-MM-dd HH-mm")
)
.withBucketCheckInterval(1000L)//检查的间隔
.withOutputFileConfig(OutputFileConfig.builder()
.withPartPrefix("atguigu")
.withPartSuffix(".log")
.build())
.build();
ds.map(JSON::toJSONString).sinkTo(stringFileSink);
try {
env.execute();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
- Kafka Sink(重点)
- 生产者对象:KafkaProducer
- Kafka生产者分区策略:
- 如果明确指定分区号,直接用
- 如果没有指定分区号,但是Record中带了key,就按照key的hash值对分区数取余得到分区号
- 如果没有指定相关分区号,使用粘性分区策略
- 生产者相关配置
- key.serializer : key的序列化器
- value.serializer: value的序列化器
- bootstrap.servers: 集群位置
- retries: 重试次数
- batch.size 批次大小
- linger.ms 批次超时时间
- acks 应答级别
- transaction.id 事务ID
- Shell中开启Kafka消费者的命令:
kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
public class Flink02_KafkaSink {
public static void main(String[] args) {
//1.创建运行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//默认是最大并行度
env.setParallelism(1);
//开启检查点
env.enableCheckpointing(5000);
DataStreamSource<Event> ds = Flink06_EventSource.getEventSource(env);
//KafkaSink
KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
.setBootstrapServers("hadoop102:9092,hadoop103:9092")
.setRecordSerializer(
KafkaRecordSerializationSchema.<String>builder()
.setTopic("first")
.setValueSerializationSchema(new SimpleStringSchema())
.build()
)
//语义
//AT_LEAST_ONCE:至少一次,表示数据可能重复,需要考虑去重操作
//EXACTLY_ONCE:精确一次
//kafka transaction timeout is larger than broker
//kafka超时时间:1H
//broker超时时间:15分钟
// .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)//数据传输的保障
.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)//数据传输的保障
.setTransactionalIdPrefix("flink"+ RandomUtils.nextInt(0,100000))
// .setProperty(ProducerConfig.RETRIES_CONFIG,"10")
.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,"600000")
.build();
ds.map(
JSON::toJSONString
).sinkTo(kafkaSink);//写入到kafka 生产者
//shell 消费者:kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
try {
env.execute();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
为了在Shell中开启消费者更为便捷,这里写了一个小脚本,用来动态的设置主题并开启相应的Kafka消费者,脚本名称为kc.sh.文章来源地址https://www.toymoban.com/news/detail-753746.html
#!/bin/bash
# 检查参数数量
if [ $# -lt 1 ]; then
echo "Usage: $0 <topic>"
exit 1
fi
# 从命令行参数获取主题
topic=$1
# Kafka配置
bootstrap_server="hadoop102:9092"
# 构建kafka-console-consumer命令
consumer_command="kafka-console-consumer.sh --bootstrap-server $bootstrap_server --topic $topic"
# 打印消费命令
echo "Running Kafka Consumer for topic: $topic"
echo "Command: $consumer_command"
# 执行消费命令
$consumer_command
到了这里,关于Flink基础之DataStream API的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!