flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-9609) Add bucket ready mechanism for BucketingSink when checkpoint complete
Date Fri, 20 Jul 2018 10:02:00 GMT

    [ https://issues.apache.org/jira/browse/FLINK-9609?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16550599#comment-16550599
] 

ASF GitHub Bot commented on FLINK-9609:
---------------------------------------

Github user zhangminglei commented on a diff in the pull request:

    https://github.com/apache/flink/pull/6375#discussion_r203995868
  
    --- Diff: flink-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/bucketing/BucketingSink.java
---
    @@ -722,9 +727,31 @@ public void notifyCheckpointComplete(long checkpointId) throws Exception
{
     					}
     				}
     			}
    +			isBucketReady(partitionPaths);
     		}
     	}
     
    +	@Override
    +	public boolean isBucketReady(Set<Path> bucketPathes) {
    --- End diff --
    
    If we finish the for loop and haven't returned true yet, then return false. I think this
is reasonable. If a file end with pending or inProgress, This means that the file is not ready.
and we can not use it.
    



> Add bucket ready mechanism for BucketingSink when checkpoint complete
> ---------------------------------------------------------------------
>
>                 Key: FLINK-9609
>                 URL: https://issues.apache.org/jira/browse/FLINK-9609
>             Project: Flink
>          Issue Type: New Feature
>          Components: filesystem-connector, Streaming Connectors
>    Affects Versions: 1.5.0, 1.4.2
>            Reporter: zhangminglei
>            Assignee: zhangminglei
>            Priority: Major
>              Labels: pull-request-available
>
> Currently, BucketingSink only support {{notifyCheckpointComplete}}. However, users want
to do some extra work when a bucket is ready. It would be nice if we can support {{BucketReady}}
mechanism for users or we can tell users when a bucket is ready for use. For example, One
bucket is created for every 5 minutes, at the end of 5 minutes before creating the next bucket,
the user might need to do something as the previous bucket ready, like sending the timestamp
of the bucket ready time to a server or do some other stuff.
> Here, Bucket ready means all the part files suffix name under a bucket neither {{.pending}}
nor {{.in-progress}}. Then we can think this bucket is ready for user use. Like a watermark
means no elements with a timestamp older or equal to the watermark timestamp should arrive
at the window. We can also refer to the concept of watermark here, or we can call this *BucketWatermark*
if we could.
> Recently, I found a user who wants this functionality which I would think.
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Let-BucketingSink-roll-file-on-each-checkpoint-td19034.html
> Below is what he said:
> My user case is we read data from message queue, write to HDFS, and our ETL team will
use the data in HDFS. *In the case, ETL need to know if all data is ready to be read accurately*,
so we use a counter to count how many data has been wrote, if the counter is equal to the
number we received, we think HDFS file is ready. We send the counter message in a custom sink
so ETL can know how many data has been wrote, but if use current BucketingSink, even through
HDFS file is flushed, ETL may still cannot read the data. If we can close file during checkpoint,
then the result is accurately. And for the HDFS small file problem, it can be controller by
use bigger checkpoint interval. 



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Mime
View raw message