spark-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Thakrar, Jayesh" <>
Subject Re: Datasource API V2 and checkpointing
Date Tue, 01 May 2018 19:58:12 GMT
Just wondering-

Given that currently V2 is less performant because of use of Row vs InternalRow (and other
things?), is still evolving, and is missing some of the other features of V1, it might help
to focus on remediating those features and then look at porting the filesources over.

As for the escape hatch (or additional capabilities), can that be implemented as traits?

And imho, i think filesources and other core sources should have the same citizenship level
as us granted to the other sources in V2. This is so that others can use then as good references
for emulation.


From: Joseph Torres <>
Sent: Tuesday, May 1, 2018 1:58:54 PM
To: Ryan Blue
Cc: Thakrar, Jayesh;
Subject: Re: Datasource API V2 and checkpointing

I agree that Spark should fully handle state serialization and recovery for most sources.
This is how it works in V1, and we definitely wouldn't want or need to change that in V2.*
The question is just whether we should have an escape hatch for the sources that don't want
Spark to do that, and if so what the escape hatch should look like.

I don't think a watermark checkpoint would work, because there's no guarantee (especially
considering the "maxFilesPerTrigger" option) that all files with the same timestamp will be
in the same batch. But in general, hanging the fundamental mechanics of how file sources take
checkpoints seems like it would impose a serious risk of performance regressions, which I
don't think are a desirable risk when performing an API migration that's going to swap out
users' queries from under them. I would be very uncomfortable merging a V2 file source which
we can't confidently assert has the same performance characteristics as the existing one.

* Technically, most current sources do write their initial offset to the checkpoint directory,
but this is just a workaround to the fact that the V1 API has no handle to give Spark the
initial offset. So if you e.g. start a Kafka stream from latest offsets, and it fails in the
first batch, Spark won't know to restart the stream from the initial offset which was originally
generated. That's easily fixable in V2, and then no source will have to even look at the checkpoint
directory if it doesn't want to.

On Tue, May 1, 2018 at 10:26 AM, Ryan Blue <<>>
I think there's a difference. You're right that we wanted to clean up the API in V2 to avoid
file sources using side channels. But there's a big difference between adding, for example,
a way to report partitioning and designing for sources that need unbounded state. It's a judgment
call, but I think unbounded state is definitely not something that we should design around.
Another way to think about it: yes, we want to design a better API using existing sources
as guides, but we don't need to assume that everything those sources do should to be supported.
It is reasonable to say that this is a case we don't want to design for and the source needs
to change. Why can't we use a high watermark of files' modified timestamps?

For most sources, I think Spark should handle state serialization and recovery. Maybe we can
find a good way to make the file source with unbounded state work, but this shouldn't be one
of the driving cases for the design and consequently a reason for every source to need to
manage its own state in a checkpoint directory.


On Mon, Apr 30, 2018 at 12:37 PM, Joseph Torres <<>>
I'd argue that letting bad cases influence the design is an explicit goal of DataSourceV2.
One of the primary motivations for the project was that file sources hook into a series of
weird internal side channels, with favorable performance characteristics that are difficult
to match in the API we actually declare to Spark users. So a design that we can't migrate
file sources to without a side channel would be worrying; won't we end up regressing to the
same situation?

On Mon, Apr 30, 2018 at 11:59 AM, Ryan Blue <<>>
Should we really plan the API for a source with state that grows indefinitely? It sounds like
we're letting a bad case influence the design, when we probably shouldn't.

On Mon, Apr 30, 2018 at 11:05 AM, Joseph Torres <<>>
Offset is just a type alias for arbitrary JSON-serializable state. Most implementations should
(and do) just toss the blob at Spark and let Spark handle recovery on its own.

In the case of file streams, the obstacle is that the conceptual offset is very large: a list
of every file which the stream has ever read. In order to parse this efficiently, the stream
connector needs detailed control over how it's stored; the current implementation even has
complex compactification and retention logic.

On Mon, Apr 30, 2018 at 10:48 AM, Ryan Blue <<>>
Why don't we just have the source return a Serializable of state when it reports offsets?
Then Spark could handle storing the source's state and the source wouldn't need to worry about
file system paths. I think that would be easier for implementations and better for recovery
because it wouldn't leave unknown state on a single machine's file system.


On Fri, Apr 27, 2018 at 9:23 AM, Joseph Torres <<>>
The precise interactions with the DataSourceV2 API haven't yet been hammered out in design.
But much of this comes down to the core of Structured Streaming rather than the API details.

The execution engine handles checkpointing and recovery. It asks the streaming data source
for offsets, and then determines that batch N contains the data between offset A and offset
B. On recovery, if batch N needs to be re-run, the execution engine just asks the source for
the same offset range again. Sources also get a handle to their own subfolder of the checkpoint,
which they can use as scratch space if they need. For example, Spark's FileStreamReader keeps
a log of all the files it's seen, so its offsets can be simply indices into the log rather
than huge strings containing all the paths.

SPARK-23323 is orthogonal. That commit coordinator is responsible for ensuring that, within
a single Spark job, two different tasks can't commit the same partition.

On Fri, Apr 27, 2018 at 8:53 AM, Thakrar, Jayesh <<>>
Wondering if this issue is related to SPARK-23323?

Any pointers will be greatly appreciated….


From: "Thakrar, Jayesh" <<>>
Date: Monday, April 23, 2018 at 9:49 PM
To: "<>" <<>>
Subject: Datasource API V2 and checkpointing

I was wondering when checkpointing is enabled, who does the actual work?
The streaming datasource or the execution engine/driver?

I have written a small/trivial datasource that just generates strings.
After enabling checkpointing, I do see a folder being created under the checkpoint folder,
but there's nothing else in there.

Same question for write-ahead and recovery?
And on a restart from a failed streaming session - who should set the offsets?
The driver/Spark or the datasource?

Any pointers to design docs would also be greatly appreciated.


Ryan Blue
Software Engineer

Ryan Blue
Software Engineer

Ryan Blue
Software Engineer

View raw message