Flink comprehensive case (IX)

Flink Comprehensive case ( Nine )

Today's goal

• Flink FileSink Landing write to HDFS
• FlinkSQL Integrate Hive Data warehouse
• Comprehensive cases of automatic praise of orders

Flink FileSink Landing write to HDFS

• Common file storage formats

  TextFile

  csv

  rcFile

  parquet

  orc

  sequenceFile

• Support the integrated writing of stream and batch to HDFS

• File Sink demand

  Write stream data to 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． Initialize the stream computing runtime environment         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.setParallelism(1);        //2． Set up Checkpoint（10s） Start periodically   and  stateBackend  Storage path         // Sink Ensure only one semantic use  checkpoint  and   Second paragraph submission         env.enableCheckpointing(10000);        env.setStateBackend(new FsStateBackend("file:///d:/chk/"));        //4． Access socket data source , get data         DataStreamSource<String> source = env.socketTextStream("node1", 9999);        //5． establish Streamingfilesink object         OutputFileConfig config = OutputFileConfig                .builder()                .withPartPrefix("coo")                .withPartSuffix(".txt")                .build();        //5-1.  Create output file configuration , Specify the output path  /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  Line formatting  , withBucketAssigner->DateTimeBucketAssigner                .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd--HH--mm"))                //withRollingPolicy ->  Default drum strategy                 .withRollingPolicy(DefaultRollingPolicy.builder()                        .withMaxPartSize(64 * 1024 * 1024)                        .withRolloverInterval(TimeUnit.SECONDS.toMillis(10))                        .withInactivityInterval(TimeUnit.SECONDS.toMillis(5))                        .build())        //withOutputFileConfig ->  Configuration of output file                 .withOutputFileConfig(config)                .build();        //6. Set output  sink        source.print();        source.sinkTo(sink).setParallelism(1);        //source.addSink(sink).setParallelism(1);        //7． Perform tasks         env.execute();    }}