[
https://issues.apache.org/jira/browse/BEAM-644?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15516916#comment-15516916
]
Ben Chambers commented on BEAM-644:
-----------------------------------
With SplittableDoFn there are generally two cases:
1. In the presence of an element that is being split, the SplittableDoFn will
specify what the watermark for the data from the data from that element should
be. The output watermark of the ParDo will be less than all the watermarks of
all the elements still in progress (eg., it will hold the output watermark
based on the data being read within the SplittableDoFn).
2. In the absence of data, you hold to some constant related to the input
watermark. So for example, you are receiving file names on an unbounded source
(say Kafka). The timestamp of the messages on Kafka indicate when the file name
was reported. You then have a SplittableDoFn which reads the contents of the
file. You know that the file represents data from the last hour (say a log
file), so you set D=1 hour. Even in the absence of any data the Kafka source
exports a watermark. The output watermark of the SplittableDoFn is one hour
behind that. When the Kafka source receives a file name, the SplittableDoFn is
able to emit records up to an hour behind the timestamp of the file-name
element, because things have been held up.
> Primitive to shift the watermark while assigning timestamps
> -----------------------------------------------------------
>
> Key: BEAM-644
> URL: https://issues.apache.org/jira/browse/BEAM-644
> Project: Beam
> Issue Type: New Feature
> Components: beam-model
> Reporter: Kenneth Knowles
> Assignee: Kenneth Knowles
>
> There is a general need, especially important in the presence of
> SplittableDoFn, to be able to assign new timestamps to elements without
> making them late or droppable.
> - DoFn.withAllowedTimestampSkew is inadequate, because it simply allows one
> to produce late data, but does not allow one to shift the watermark so the
> new data is on-time.
> - For a SplittableDoFn, one may receive an element such as the name of a log
> file that contains elements for the day preceding the log file. The timestamp
> on the filename must currently be the beginning of the log. If such elements
> are constantly flowing, it may be OK, but since we don't know that element is
> coming, in that absence of data, the watermark may advance. We need a way to
> keep it far enough back even in the absence of data holding it back.
> One idea is a new primitive ShiftWatermark / AdjustTimestamps with the
> following pieces:
> - A constant duration (positive or negative) D by which to shift the
> watermark.
> - A function from TimestampedElement<T> to new timestamp that is >= t + D
> So, for example, AdjustTimestamps(<-60 minutes>, f) would allow f to make
> timestamps up to 60 minutes earlier.
> With this primitive added, outputWithTimestamp and withAllowedTimestampSkew
> could be removed, simplifying DoFn.
> Alternatively, all of this functionality could be bolted on to DoFn.
> This ticket is not a proposal, but a record of the issue and ideas that were
> mentioned.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
