Flink综合案例(九)

今日目标

• Flink FileSink 落地写入到 HDFS
• FlinkSQL 整合 Hive数据仓库
• 订单自动好评综合案例

Flink FileSink 落地写入到 HDFS

• 常用的文件存储格式

  TextFile

  csv

  rcFile

  parquet

  orc

  sequenceFile

• 支持流批一体的写入到 HDFS

• File Sink 需求

  将流数据写入到 HDFS

  package cn.itcast.flink.filesink;import org.apache.flink.api.common.serialization.SimpleStringEncoder;import org.apache.flink.connector.file.sink.FileSink;import org.apache.flink.core.fs.Path;import org.apache.flink.runtime.state.filesystem.FsStateBackend;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.OutputFileConfig;import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;import java.util.concurrent.TimeUnit;/** * Author itcast * Date 2021/6/24 10:52 * Desc TODO */public class FileSinkDemo {    public static void main(String[] args) throws Exception {        //1．初始化流计算运行环境        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.setParallelism(1);        //2．设置Checkpoint（10s）周期性启动 和 stateBackend 存储路径        // Sink保证仅一次语义使用 checkpoint 和 二段提交        env.enableCheckpointing(10000);        env.setStateBackend(new FsStateBackend("file:///d:/chk/"));        //4．接入socket数据源，获取数据        DataStreamSource<String> source = env.socketTextStream("node1", 9999);        //5．创建Streamingfilesink对象        OutputFileConfig config = OutputFileConfig                .builder()                .withPartPrefix("coo")                .withPartSuffix(".txt")                .build();        //5-1. 创建输出文件配置，指定输出路径 /FlinkStreamFileSink/parquet        FileSink sink = FileSink                .forRowFormat(                new Path("hdfs://node1:8020/FileSink/parquet"),                new SimpleStringEncoder<String>("UTF-8"))                // sink-kafka new FlinkKafkaProducer                //5-2.StreamingFileSink 行格式化 ， withBucketAssigner->DateTimeBucketAssigner                .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd--HH--mm"))                //withRollingPolicy -> 默认滚筒策略                .withRollingPolicy(DefaultRollingPolicy.builder()                        .withMaxPartSize(64 * 1024 * 1024)                        .withRolloverInterval(TimeUnit.SECONDS.toMillis(10))                        .withInactivityInterval(TimeUnit.SECONDS.toMillis(5))                        .build())        //withOutputFileConfig -> 输出文件的配置                .withOutputFileConfig(config)                .build();        //6.设置输出 sink        source.print();        source.sinkTo(sink).setParallelism(1);        //source.addSink(sink).setParallelism(1);        //7．执行任务        env.execute();    }}