[
https://issues.apache.org/jira/browse/FLINK-9138?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16427254#comment-16427254
]
Lakshmi Rao commented on FLINK-9138:
------------------------------------
[~narayaruna] [~fhueske] do you have thoughts on how to implement this? I have
a working version of it, that I've been running on a stream for the last couple
of days and has been writing to my filesystem of choice every X minutes. The
basic idea is to track a bucketCreationTime and add this snippet to the
[shouldRoll|https://github.com/apache/flink/blob/release-1.4.0/flink-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/bucketing/BucketingSink.java#L452]
method:
{code:java}
else {
long createTime = bucketState.bucketCreationTime;
long currentProcessingTime = processingTimeService.getCurrentProcessingTime();
if(currentProcessingTime - createTime > rollIntervalThreshold) {
shouldRoll = true;
LOG.debug( "BucketingSinkWithSuffix {} starting new bucket because bucket
is older than rollover interval {} ",
subtaskIndex,
rollIntervalThreshold
);
}
{code}
I can create a PR out of it, if this makes sense.
> Enhance BucketingSink to also flush data by time interval
> ---------------------------------------------------------
>
> Key: FLINK-9138
> URL: https://issues.apache.org/jira/browse/FLINK-9138
> Project: Flink
> Issue Type: Improvement
> Components: filesystem-connector
> Affects Versions: 1.4.2
> Reporter: Narayanan Arunachalam
> Priority: Major
>
> BucketingSink now supports flushing data to the file system by size limit and
> by period of inactivity. It will be useful to also flush data by a specified
> time period. This way, the data will be written out when write throughput is
> low but there is no significant time period gaps between the writes. This
> reduces ETA for the data in the file system and should help move the
> checkpoints faster as well.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
