说明
可通过实现 BucketAssigner 类自定义桶名
实现类
class BucketAssignerImpl implements BucketAssigner<Tuple2<String, Integer>, String> {
// 文件夹命名规则：当前时间的 yyyy-MM-dd_HH 格式作为文件名
@Override
public String getBucketId(Tuple2<String, Integer> element, Context context) {
Date date = new Date();
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd_HH");
String name = sdf.format(date);
return name;
}
// 序列化类 参考默认的
@Override
public SimpleVersionedSerializer<String> getSerializer() {
return SimpleVersionedStringSerializer.INSTANCE;
}
}
其他代码
package std;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
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.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
public class StdFileSink_fold {
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2);
checkpoint(env);
// 转换操作
SingleOutputStreamOperator<Tuple2<String, Integer>> soso = env.addSource(buildKafka())
.flatMap(new FlatMapFunImpl());
SingleOutputStreamOperator<Tuple2<String, Integer>> soso2 = soso
.keyBy(value -> value.f0)
.sum(1);
FileSink<Tuple2<String, Integer>> sink = FileSink
.<Tuple2<String, Integer>> forRowFormat(
new Path("./result/filesink") // 保存到本地。保存到hdfs的写法：hdfs://hadoop1:8020/flink_filesink
, new SimpleStringEncoder<>("UTF-8"))
.withBucketAssigner(new BucketAssignerImpl())
.withRollingPolicy( // 指定滚动策略
DefaultRollingPolicy.builder()
// 如果接收数据，每 15 分钟就滚动生成一个新文件
.withRolloverInterval(TimeUnit.SECONDS.toMillis(20))
// 最近 5 分钟没有收到数据，就滚动生成一个新文件
.withInactivityInterval(TimeUnit.SECONDS.toMillis(10))
// 文件大小达到 120KB （写入最后一条记录后），就滚动生成一个新文件
.withMaxPartSize(120 )
.build())
.build();
soso2.sinkTo(sink);
soso.print();
env.execute("myjob");
}
private static FlinkKafkaConsumer buildKafka(){
Properties pro = new Properties();
pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop1:9092,hadoop2:9092,hadoop3:9092");
pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "consumer-group");
pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
// 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>("kafka2flink", new SimpleStringSchema(), pro);
return fkc;
}
static class FlatMapFunImpl implements FlatMapFunction<String, Tuple2<String,Integer>> {
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
String[] arr = value.split(",");
for(String item:arr){
out.collect(Tuple2.of(item,1));
}
}
}
private static void checkpoint(StreamExecutionEnvironment env){
// 必设置，每隔 5000 ms 执行一次checkpoint。
env.enableCheckpointing(5000);//一般秒级、分钟级
//设置保存文件路径，生产环境中应该是 HDFS
// env.setStateBackend(new FsStateBackend("file:///D:/flink_cp"));
env.setStateBackend(new FsStateBackend("file:///D:/flink_cp"));
//env.setStateBackend(new RocksDBStateBackend("file:///D:/bigdata/flink_java/state/RocksDB",true));
// 设置2个checkpoint之间最少等待时间，这里至少要等到500ms
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
//checkpoint容忍的错误数量，达到此数量，就让该任务失败，默认值是：0。这里容忍3个错误
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
//取消任务时，是否清理checkpoint
// true：取消任务时，会删除外部checkpoint，默认值
// false：取消任务时，保留外部checkpoint
//这里设置不删除
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//设置checkpoint的执行模式：EXACTLY_ONCE，默认
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 设置checkpoint的超时时间，如果在60s内尚未完成说明本次checkpoint失败，则丢弃
env.getCheckpointConfig().setCheckpointTimeout(60000);//这里设置10分钟
// 设置同一时间有多少个checkpoint可以同时执行。与 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500); 可能冲突
//env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);//默认为1
// 重启策略：程序出现异常时，重启2次，每次延迟3秒中重启，超过2次退出程序
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, 3000));
}
}
参考：
https://www.jianshu.com/p/22bf54da7f5d