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. Jayesh ________________________________ From: Joseph Torres <joseph.tor...@databricks.com> Sent: Tuesday, May 1, 2018 1:58:54 PM To: Ryan Blue Cc: Thakrar, Jayesh; dev@spark.apache.org 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 <rb...@netflix.com<mailto:rb...@netflix.com>> wrote: 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. rb On Mon, Apr 30, 2018 at 12:37 PM, Joseph Torres <joseph.tor...@databricks.com<mailto:joseph.tor...@databricks.com>> wrote: 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 <rb...@netflix.com<mailto:rb...@netflix.com>> wrote: 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 <joseph.tor...@databricks.com<mailto:joseph.tor...@databricks.com>> wrote: 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 <rb...@netflix.com<mailto:rb...@netflix.com>> wrote: 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. rb On Fri, Apr 27, 2018 at 9:23 AM, Joseph Torres <joseph.tor...@databricks.com<mailto:joseph.tor...@databricks.com>> wrote: 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 <jthak...@conversantmedia.com<mailto:jthak...@conversantmedia.com>> wrote: Wondering if this issue is related to SPARK-23323? Any pointers will be greatly appreciated…. Thanks, Jayesh From: "Thakrar, Jayesh" <jthak...@conversantmedia.com<mailto:jthak...@conversantmedia.com>> Date: Monday, April 23, 2018 at 9:49 PM To: "dev@spark.apache.org<mailto:dev@spark.apache.org>" <dev@spark.apache.org<mailto:dev@spark.apache.org>> 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. Thanks, Jayesh -- Ryan Blue Software Engineer Netflix -- Ryan Blue Software Engineer Netflix -- Ryan Blue Software Engineer Netflix
