[
https://issues.apache.org/jira/browse/FLINK-9609?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16549361#comment-16549361
]
ASF GitHub Bot commented on FLINK-9609:
---------------------------------------
GitHub user zhangminglei opened a pull request:
https://github.com/apache/flink/pull/6375
[FLINK-9609] [connectors] Add bucket ready mechanism for BucketingSin…
## What is the purpose of the change
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.
## Brief change log
Add an interface ```BucketReady``` .
## Verifying this change
This change is a trivial rework / code cleanup without any test coverage.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (not documented)
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/zhangminglei/flink flink-9609-bucketready
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/6375.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #6375
----
commit f95894956ac15d09b51b3a232d6f83227582e641
Author: zhangminglei <zml13856086071@...>
Date: 2018-07-19T14:38:45Z
[FLINK-9609] [connectors] Add bucket ready mechanism for BucketingSink when
checkpoint complete
----
> 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)
