Flink基础之DataStream API

流的合并

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();//connectConnectedStreams<Integer, String> connectDs = ds1.connect(ds3);//处理connectDs.process(new CoProcessFunction<Integer, String, String>() {@Overridepublic void processElement1(Integer value, CoProcessFunction<Integer, String, String>.Context ctx, Collector<String> out) throws Exception {out.collect(value.toString());}@Overridepublic 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的检查点。

FileSink：fileSink = FileSink.<数据流泛型>forRowFormat(输出路径, 数据流编码器)
- 文件滚动策略 .withRollingPolicy().builder()
  - 文件多大滚动.withMaxPartSize(MemorySize.parse(“10m”))
  - 多长时间滚动一次 .withRolloverInterval(Duration.ofSeconds(10))
  - 多久不活跃滚动 .withInactivityInterval(Duration.ofSeconds(5))
- 目录滚动策略：一般设置为按照天或者小时或者其他时间间隔
- 文件输出配置：可以设置输出文件的前缀和后缀

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);//FileSinkFileSink<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);//KafkaSinkKafkaSink<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 firsttry {env.execute();} catch (Exception e) {throw new RuntimeException(e);}}
}

为了在Shell中开启消费者更为便捷，这里写了一个小脚本，用来动态的设置主题并开启相应的Kafka消费者，脚本名称为kc.sh.

#!/bin/bash# 检查参数数量
if [ $# -lt 1 ]; thenecho "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