[
https://issues.apache.org/jira/browse/SPARK-25331?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16608933#comment-16608933
]
Mihaly Toth commented on SPARK-25331:
-------------------------------------
I was thinking about how to make FileStreamSink idempotent even in failure
cases using the first approach (deterministic file names).
As a starting point we need to put the partition id (and bucket id if it
exists) into the file name and remove the UUID from it. If the same batch is
rewritten and the file already exists we can simply skip writing it again
assuming the same partition of the same batch will generate the same data
again. There are a few special cases though:
If the file is half written when the writing executor stops there will be
missing records from the end of the file. We can eliminate this by first
writing the data into a temp file and then moving it to its intended location.
The other problematic are is around {{maxRecordsPerFile}} limit. This way the
same batch+partition pair may generate multiple files. It may happen that some
of these files were created but some of them are missing. Generating those that
are missing is only working if the ordering of the items is exactly the same in
each run. This may or may not be true. If the order differs between the two
runs and we simply skip generating those that already exists there may be
missing or duplicated items in the resulting files.
We could subtract the records in the already existing files from the input RDD.
I feel that would make the writing logic quite complex, and that would imply
unexpected computational load onto the executor. But that would be working in
all cases.
Another solution for partially generated file sets would be to start reading
and generating the file at the same time and compare the records one by one. If
the already existing file is the same as the file to be generated we can skip
creating the file. If it is different we can create a file with some mark in
its name like "-v2". With this the receiver can achieve exactly once semantics
in the following ways:
# Do not limit the maximum records per file
# Limit the number of records but apply strict ordering on the resulting rdd
# Limit the number of records without applying strict ordering but compensate
those files that have newer versions appearing in the output directory
[~rxin], [~Gengliang.Wang] what is your opinion on this?
> Structured Streaming File Sink duplicates records in case of driver failure
> ---------------------------------------------------------------------------
>
> Key: SPARK-25331
> URL: https://issues.apache.org/jira/browse/SPARK-25331
> Project: Spark
> Issue Type: Bug
> Components: Structured Streaming
> Affects Versions: 2.3.1
> Reporter: Mihaly Toth
> Priority: Major
>
> Lets assume {{FileStreamSink.addBtach}} is called and an appropriate job has
> been started by {{FileFormatWrite.write}} and then the resulting task sets
> are completed but in the meantime the driver dies. In such a case repeating
> {{FileStreamSink.addBtach}} will result in duplicate writing of the data
> In the event the driver fails after the executors start processing the job
> the processed batch will be written twice.
> Steps needed:
> # call {{FileStreamSink.addBtach}}
> # make the {{ManifestFileCommitProtocol}} fail to finish its {{commitJob}}
> # call {{FileStreamSink.addBtach}} with the same data
> # make the {{ManifestFileCommitProtocol}} finish its {{commitJob}}
> successfully
> # Verify file output - according to {{Sink.addBatch}} documentation the rdd
> should be written only once
> I have created a wip PR with a unit test:
> https://github.com/apache/spark/pull/22331
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
