目录
扩展文件索引进行阅读
介绍
案例演示
http://www.Sina.com/http://www.Sina.com/http://www.Sina.com/https://ci.Apache.org/projects/flink/flink-donk
新的datasinkAPI (贝塔)。
在以前发布的Flink版本[1]中，由于source connector工作受流批处理两种模式的支持，因此在Flink 1.12中，社区支持统一的数据链路API (flip 新的抽象引入了write/commit协议和更模块化的接口。 Sink的实现者只需定义what和how:SinkWriter来写入数据并输出需要commit的内容即可。 (例如，committables )。 Committer和GlobalCommitter封装了committables的处理方法。 框架负责when和where。 也就是说，什么时候，在什么机器或进程上进行commit？
通过这种模块化抽象，您可以在BATCH和STREAMING两种运行时模式下实现不同的运行时策略，以便只在一个sink上实现，也可以高效地运行这两种模式。 Flink 1.12提供了一个统一的filesinkconnector来替换现有的流文件连接器(flink-19758 )。 其他连接器也将逐步转移到新接口。
Flink 1.12中的FileSink为批处理和流处理提供了集成的接收器，用于将分区文件写入Flink文件系统抽象所支持的文件系统。 此文件系统连接器为批处理和流处理提供了相同的保证，是现有流文件接收器的改进。
33558 www.Sina.com/package cn.lanson.extend； importorg.Apache.flink.API.com mon.serialization.simplestringencoder； importorg.Apache.flink.connector.file.sink.filesink； importorg.Apache.flink.core.fs.path； importorg.Apache.flink.runtime.state.file system.fsstatebackend； importorg.Apache.flink.streaming.API.datastream.datastream source； importorg.Apache.flink.streaming.API.environment.streamexecutionenvironment； importorg.Apache.flink.streaming.API.functions.sink.file system.output file config； importorg.Apache.flink.streaming.API.functions.sink.file system.bucketassigners.datetimebucketasigner； importorg.Apache.flink.streaming.API.functions.sink.file system.rolling policies.defaultrolling policy； import Java.util.concurrent.time unit；/* * * author lanson * desc */publicclassfilesinkdemo { publicstaticvoidmain (string [ ] args ) throwsexception(/) ) ) ) ) ) ) ) )1. envstreamexecutionenvironmentenv=streamexecutionenvironment.getexecutionenvironment (； env.enable checkpointing (time unit.seconds.tomillis (10 ) )； env.setstate back end (newfsstatebackend (' file ://d :/ckp ' ) )； //2.sourcedatastreamsourcestringlines=env.socket text stream (' node1'，9999 )； //3.sink//sink前缀和后缀//文件标题和文件扩展名//prefix-XXX--.txtoutputfileconfigconfig=output file config.buildix finalfilesinkstringsink=filesink.forrowformat (new path (output path )， newsimplestringencoderstring (utf-8 ) ) ) withbucketassigner (newdatetimebucketassigner ) ).withrollingpolicy ) desting 时间单元. minutes.tomillis (15 ) ).withinactivityinterval (时间单元. minutes.tomillis ) (5) )。 lines.Sinkto(sink ).setParallelism(1) ) 1； env.execute (； }