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();
      }
  }