Flink 按时间分文件写入HDFS案例

朱雀 2022-01-22 11:09 442阅读 0赞

参考官网API：

[https://flink.sojb.cn/dev/connectors/filesystem\_sink.html][https_flink.sojb.cn_dev_connectors_filesystem_sink.html]

其余案例：

[https://my.oschina.net/u/2000675/blog/1787545][https_my.oschina.net_u_2000675_blog_1787545]

[https://blog.csdn.net/baifanwudi/article/details/88247469][https_blog.csdn.net_baifanwudi_article_details_88247469]

正常操作：

1）导入依赖：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

2）

DataStream<String> input = env.fromElements("a","b")

BucketingSink<String> sink = new BucketingSink<String>("/base/path"); sink.setBucketer(new DateTimeBucketer<String>("yyyy-MM-dd--HHmm", ZoneId.of("America/Los_Angeles"))); sink.setWriter(new SequenceFileWriter<IntWritable, Text>()); sink.setBatchSize(1024 * 1024 * 400); // this is 400 MB, sink.setBatchRolloverInterval(20 * 60 * 1000); // this is 20 mins input.addSink(sink);

主要设置三个属性Bucketer，Writer，BatchSize。

Bucketer：数据写入hdfs目录划分，DateTimeBucketer是根据当前系统时间划分，具体粒度根据传入的参数确定。当然我们也可以设置自己的划分规则，利用数据里的字段确定划分目录；

例如我根据Json数据里的Timestamp字段确定划分目录：

class DateHourBucketer implements Bucketer<JSONObject>{
    			private static final long serialVersionUID = 1L;
    			private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd--HH");
    			@Override
    			public Path getBucketPath(Clock clock, Path basePath, JSONObject element) {
    				// TODO Auto-generated method stub
    				Long timetamp = element.getLong("Timestamp");
    				String newDateTimeString = format.format(new Date(timetamp));
    				return new Path(basePath + "/" + newDateTimeString);
    			}
    			
    		}

Writer：数据写入格式，默认转化为字符串写入。如果数据格式为SequenceFiles，我们可以用SequenceFileWriter；

BatchSize：默认每一个线程一个part文件，batchsize指定part文件多大的时候生成新的文件

当然我们还是可以设置路径前缀、后缀，多长时间关闭文件句柄等等属性。

默认生成的路径格式如下：

/base/path/{date-time}/part-{parallel-task}-{count}

count是由于BatchSize而设定的part文件编号

注意要开**启checkpoint**，不然文件一直处于pending状态，句柄无法关闭，不能读取

[https_flink.sojb.cn_dev_connectors_filesystem_sink.html]: https://flink.sojb.cn/dev/connectors/filesystem_sink.html
[https_my.oschina.net_u_2000675_blog_1787545]: https://my.oschina.net/u/2000675/blog/1787545
[https_blog.csdn.net_baifanwudi_article_details_88247469]: https://blog.csdn.net/baifanwudi/article/details/88247469
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20220122/541cf71482904f92b7b71fdc4e18c524.png