Hi Eric, I think it will depend on how you implement the sink and when the data in the sink partitions are committed.
I think the batch can be repeated during task retries as well as if the driver fails before the batch id is committed in sparks checkpoint. In the first case may be the sink had not committed the batch (rolled back) so the batch id is not present and it can go ahead and write the data. In the second case the sink needs to check for duplicates and ignore that batch. I guess irrespective of the case the Sink just needs to ignore the batch if a batch Id is present in the datastore. Also the discussion in the mail thread you posted assumes that each intermediate operation is idempotent otherwise the data generated when the batch replays can be different. On Thu, 6 Dec 2018 at 11:55, Eric Wohlstadter <wohls...@cs.ubc.ca> wrote: > Hi Arun, Gabor, > Thanks for the feedback. We are using the "Exactly-once using > transactional writes" approach, so we don't rely on message keys for > idempotent writes. > So I should clarify that my question is specific to the "Exactly-once > using transactional writes" approach. > > We are following the approach mentioned from this previous list posting: > http://apache-spark-developers-list.1001551.n3.nabble.com/Easy-way-to-get-offset-metatada-with-Spark-Streaming-API-td22406.html > > Quoted from that posting: "I think the right way to look at this is the > batchId is just a proxy for offsets that is agnostic to what type of source > you are reading from (or how many sources their are). We might call into a > custom sink with the same batchId more than once, but it will always > contain the same data (there is no race condition, since this is stored in > a write-ahead log). As long as you check/commit the batch id in the same > transaction as the data you will get exactly once." > > We coordinate each micro-batch transaction from the driver, and atomically > commit the batchId with the data to a datastore. > > So my question is really specific to this part of the referenced posting "We > might call into a custom sink with the same batchId more than once". > > Before committing any micro-batch, we can check whether the batchId has > already been committed. > I know that this is required in the case of driver recovery, because the > batchId in the checkpoint file might be out-of-synch with the batchId that > was committed in a datastore transaction. > > Are there other cases where Spark might call into a custom sink with the > same batchId more than once? > > i.e. do we need to check Spark's current batchId against the datastore for > each micro-batch, or can we get away with only doing this check at the time > of driver recovery? > > > > > > On Thu, Dec 6, 2018 at 1:28 AM Gabor Somogyi <gabor.g.somo...@gmail.com> > wrote: > >> Hi Eric, >> >> In order to have exactly-once one need re-playable source and idempotent >> sink. >> The cases what you've mentioned are covering the 2 main group of issues. >> Practically any kind of programming problem can end-up in duplicated data >> (even in the code which feeds kafka). >> Don't know why have you asked this because if the sink see an already >> processed key then it should be just skipped and doesn't matter why it is >> duplicated. >> Cody has a really good writing about semantics: >> https://github.com/koeninger/kafka-exactly-once/blob/master/blogpost.md#delivery-semantics >> >> I think if you reach Continuous Processing this it worth to consider: >> "There are currently no automatic retries of failed tasks. Any failure >> will lead to the query being stopped and it needs to be manually restarted >> from the checkpoint." >> >> BR, >> G >> >> >> On Wed, Dec 5, 2018 at 8:36 PM Eric Wohlstadter <wohls...@gmail.com> >> wrote: >> >>> Hi all, >>> We are working on implementing a streaming sink on 2.3.1 with the >>> DataSourceV2 APIs. >>> >>> Can anyone help check if my understanding is correct, with respect to >>> the failure modes which need to be covered? >>> >>> We are assuming that a Reliable Receiver (such as Kafka) is used as the >>> stream source. And we only want to support micro-batch execution at this >>> time (not yet Continuous Processing). >>> >>> I believe the possible failures that need to be covered are: >>> >>> 1. Task failure: If a task fails, it may have written data to the sink >>> output before failure. Subsequent attempts for a failed task must be >>> idempotent, so that no data is duplicated in the output. >>> 2. Driver failure: If the driver fails, upon recovery, it might replay a >>> micro-batch that was already seen by the sink (if a failure occurs after >>> the sink has committed output but before the driver has updated the >>> checkpoint). In this case, the sink must be idempotent when a micro-batch >>> is replayed so that no data is duplicated in the output. >>> >>> Are there any other cases where data might be duplicated in the stream? >>> i.e. if neither of these 2 failures occur, is there still a case where >>> data can be duplicated? >>> >>> Thanks for any help to check if my understanding is correct. >>> >>> >>> >>> >>>
