2021年大数据Flink（四十八）：扩展阅读  Streaming File Sink

package cn.lanson.extend;

import org.apache.flink.api.common.serialization.SimpleStringEncoder;
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;

public class StreamFileSinkDemo {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(TimeUnit.SECONDS.toMillis(10));
        env.setStateBackend(new FsStateBackend("file:///D:/ckp"));

        //2.source
        DataStreamSource<String> lines = env.socketTextStream("node1", 9999);

        //3.sink
        //设置sink的前缀和后缀
        //文件的头和文件扩展名
        //prefix-xxx-.txt
        OutputFileConfig config = OutputFileConfig
                .builder()
                .withPartPrefix("prefix")
                .withPartSuffix(".txt")
                .build();

        //设置sink的路径
        String outputPath = "hdfs://node1:8020/FlinkStreamFileSink/parquet";

        //创建StreamingFileSink
        final StreamingFileSink<String> sink = StreamingFileSink
                .forRowFormat(
                        new Path(outputPath),
                        new SimpleStringEncoder<String>("UTF-8"))
                /**
                 * 设置桶分配政策
                 * DateTimeBucketAssigner --默认的桶分配政策，默认基于时间的分配器，每小时产生一个桶，格式如下yyyy-MM-dd--HH
                 * BasePathBucketAssigner ：将所有部分文件（part file）存储在基本路径中的分配器（单个全局桶）
                 */
                .withBucketAssigner(new DateTimeBucketAssigner<>())
                /**
                 * 有三种滚动政策
                 *  CheckpointRollingPolicy
                 *  DefaultRollingPolicy
                 *  OnCheckpointRollingPolicy
                 */
                .withRollingPolicy(
                        /**
                         * 滚动策略决定了写出文件的状态变化过程
                         * 1. In-progress ：当前文件正在写入中
                         * 2. Pending ：当处于 In-progress 状态的文件关闭（closed）了，就变为 Pending 状态
                         * 3. Finished ：在成功的 Checkpoint 后，Pending 状态将变为 Finished 状态
                         *
                         * 观察到的现象
                         * 1.会根据本地时间和时区，先创建桶目录
                         * 2.文件名称规则：part-<subtaskIndex>-<partFileIndex>
                         * 3.在macos中默认不显示隐藏文件，需要显示隐藏文件才能看到处于In-progress和Pending状态的文件，因为文件是按照.开头命名的
                         *
                         */
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(TimeUnit.SECONDS.toMillis(2)) //设置滚动间隔
                                .withInactivityInterval(TimeUnit.SECONDS.toMillis(1)) //设置不活动时间间隔
                                .withMaxPartSize(1024 * 1024 * 1024) // 最大尺寸
                                .build())
                .withOutputFileConfig(config)
                .build();

        lines.addSink(sink).setParallelism(1);

        env.execute();
    }
}

return new Path(bucketPath, outputFileConfig.getPartPrefix() + '-' + subtaskIndex + '-' + partCounter + outputFileConfig.getPartSuffix());

OutputFileConfig config = OutputFileConfig

 .builder()

 .withPartPrefix("prefix")

 .withPartSuffix(".ext")

 .build()

            

StreamingFileSink sink = StreamingFileSink

 .forRowFormat(new Path(outputPath), new SimpleStringEncoder<String>("UTF-8"))

 .withBucketAssigner(new KeyBucketAssigner())

 .withRollingPolicy(OnCheckpointRollingPolicy.build())

 .withOutputFileConfig(config)

 .build()

//行
StreamingFileSink.forRowFormat(new Path(path), new SimpleStringEncoder<T>())
        .withBucketAssigner(new PaulAssigner<>()) //分桶策略
        .withRollingPolicy(new PaulRollingPolicy<>()) //滚动策略
        .withBucketCheckInterval(CHECK_INTERVAL) //检查周期
        .build();

//列 parquet
StreamingFileSink.forBulkFormat(new Path(path), ParquetAvroWriters.forReflectRecord(clazz))
        .withBucketAssigner(new PaulBucketAssigner<>())
        .withBucketCheckInterval(CHECK_INTERVAL)
        .build();

import org.apache.flink.api.common.serialization.SimpleStringEncoder

import org.apache.flink.core.fs.Path

import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink



// 1. 构建DataStream

DataStream input  = ...

// 2. 构建StreamingFileSink，指定BasePath、Encoder、RollingPolicy

StreamingFileSink sink  = StreamingFileSink

    .forRowFormat(new Path(outputPath), new SimpleStringEncoder[String]("UTF-8"))

    .withRollingPolicy(

        DefaultRollingPolicy.builder()

            .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))

            .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))

            .withMaxPartSize(1024 * 1024 * 1024)

            .build())

    .build()

// 3. 添加Sink到InputDataSteam即可

input.addSink(sink)

<!-- streaming File Sink所需要的jar包-->

<dependency>

    <groupId>org.apache.flink</groupId>

    <artifactId>flink-parquet_2.12</artifactId>

    <version>1.12.0</version>

</dependency>



<!-- https://mvnrepository.com/artifact/org.apache.avro/avro -->

<dependency>

    <groupId>org.apache.avro</groupId>

    <artifactId>avro</artifactId>

    <version>1.12.0</version>

</dependency>



<dependency>

    <groupId>org.apache.parquet</groupId>

    <artifactId>parquet-avro</artifactId>

    <version>1.12.0</version>

</dependency>

public class DateTimeBucketAssigner<IN> implements BucketAssigner<IN, String> {

private static final long serialVersionUID = 1L;



// 默认的时间格式字符串

private static final String DEFAULT_FORMAT_STRING = "yyyy-MM-dd--HH";



// 时间格式字符串

private final String formatString;



// 时区

private final ZoneId zoneId;



// DateTimeFormatter被用来通过当前系统时间和DateTimeFormat来生成时间字符串

private transient DateTimeFormatter dateTimeFormatter;



/**

 * 使用默认的`yyyy-MM-dd--HH`和系统时区构建DateTimeBucketAssigner

 */

public DateTimeBucketAssigner() {

    this(DEFAULT_FORMAT_STRING);

}



/**

 * 通过能被SimpleDateFormat解析的时间字符串和系统时区

 * 来构建DateTimeBucketAssigner

 */

public DateTimeBucketAssigner(String formatString) {

    this(formatString, ZoneId.systemDefault());

}



/**

 * 通过默认的`yyyy-MM-dd--HH`和指定的时区

 * 来构建DateTimeBucketAssigner

 */

public DateTimeBucketAssigner(ZoneId zoneId) {

    this(DEFAULT_FORMAT_STRING, zoneId);

}



/**

 * 通过能被SimpleDateFormat解析的时间字符串和指定的时区

 * 来构建DateTimeBucketAssigner

 */

public DateTimeBucketAssigner(String formatString, ZoneId zoneId) {

    this.formatString = Preconditions.checkNotNull(formatString);

    this.zoneId = Preconditions.checkNotNull(zoneId);

}



/**

 * 使用指定的时间格式和时区来格式化当前ProcessingTime，以获取BucketId

 */

@Override

public String getBucketId(IN element, BucketAssigner.Context context) {

    if (dateTimeFormatter == null) {

        dateTimeFormatter = DateTimeFormatter.ofPattern(formatString).withZone(zoneId);

    }

  return dateTimeFormatter.format(Instant.ofEpochMilli(context.currentProcessingTime()));

}



@Override

public SimpleVersionedSerializer<String> getSerializer() {

    return SimpleVersionedStringSerializer.INSTANCE;

}



@Override

public String toString() {

    return "DateTimeBucketAssigner{" +

    "formatString='" + formatString + '\'' +

    ", zoneId=" + zoneId +

    '}';

    }

}

@PublicEvolving
public class BasePathBucketAssigner<T> implements BucketAssigner<T, String> {
	private static final long serialVersionUID = -6033643155550226022L;
	/**
	 * BucketId永远为""，即Bucket全路径为用户指定的BasePath
	 */
	@Override
	public String getBucketId(T element, BucketAssigner.Context context) {
		return "";
	}
	/**
	 * 用SimpleVersionedStringSerializer来序列化BucketId
	 */
	@Override
	public SimpleVersionedSerializer<String> getSerializer() {
		// in the future this could be optimized as it is the empty string.
		return SimpleVersionedStringSerializer.INSTANCE;
	}

	@Override
	public String toString() {
		return "BasePathBucketAssigner";
	}
}