Hi! Wanted to bump this thread with some updated PRs that reflect these discussions (updated IOs that parameterize FileIO#Sink, and re-use ReadableFile). The base pull requests are:
- https://github.com/apache/beam/pull/8823 (BucketMetadata implementation) - https://github.com/apache/beam/pull/8824/ (FileOperations/IO implementations) And the actual PTransform implements build on top of those PRs: - https://github.com/apache/beam/pull/9250/ (SMB Sink transform) - https://github.com/apache/beam/pull/9251 (SMB Source transform) Finally we have some benchmarks/style changes (using AutoValue/Builder pattern) for those PTransforms: - https://github.com/apache/beam/pull/9253/ (high-level API classes/style fixes) - https://github.com/apache/beam/pull/9279 (benchmarks for SMB sink and source transform) I know it's a lot of pull requests at once -- let us know if there's anything else we can clarify or streamline. Thanks! - Claire/Neville On Fri, Jul 26, 2019 at 12:45 PM Kenneth Knowles <k...@apache.org> wrote: > There is still considerable value in knowing data sources statically so > you can do things like fetch sizes and other metadata and adjust pipeline > shape. I would not expect to delete these, but to implement them on top of > SDF while still giving them a clear URN and payload so runners can know > that it is a statically-specified source. > > Kenn > > On Fri, Jul 26, 2019 at 3:23 AM Robert Bradshaw <rober...@google.com> > wrote: > >> On Thu, Jul 25, 2019 at 11:09 PM Eugene Kirpichov <kirpic...@google.com> >> wrote: >> > >> > Hi Gleb, >> > >> > Regarding the future of io.Read: ideally things would go as follows >> > - All runners support SDF at feature parity with Read (mostly this is >> just the Dataflow runner's liquid sharding and size estimation for bounded >> sources, and backlog for unbounded sources, but I recall that a couple of >> other runners also used size estimation) >> > - Bounded/UnboundedSource APIs are declared "deprecated" - it is >> forbidden to add any new implementations to SDK, and users shouldn't use >> them either (note: I believe it's already effectively forbidden to use them >> for cases where a DoFn/SDF at the current level of support will be >> sufficient) >> > - People one by one rewrite existing Bounded/UnboundedSource based >> PTransforms in the SDK to use SDFs instead >> > - Read.from() is rewritten to use a wrapper SDF over the given Source, >> and explicit support for Read is deleted from runners >> > - In the next major version of Beam - presumably 3.0 - the Read >> transform itself is deleted >> > >> > I don't know what's the current status of SDF/Read feature parity, >> maybe Luke or Cham can comment. An alternative path is offered in >> http://s.apache.org/sdf-via-source. >> >> Python supports initial splitting for SDF of all sources on portable >> runners. Dataflow support for batch SDF is undergoing testing, not yet >> rolled out. Dataflow support for streaming SDF is awaiting portable >> state/timer support. >> >> > On Thu, Jul 25, 2019 at 6:39 AM Gleb Kanterov <g...@spotify.com> wrote: >> >> >> >> What is the long-term plan for org.apache.beam.sdk.io.Read? Is it >> going away in favor of SDF, or we are always going to have both? >> >> >> >> I was looking into AvroIO.read and AvroIO.readAll, both of them use >> AvroSource. AvroIO.readAll is using SDF, and it's implemented with >> ReadAllViaFileBasedSource that takes AvroSource as a parameter. Looking at >> ReadAllViaFileBasedSource I find it not necessary to use Source<?>, it >> should be enough to have something like (KV<ReadableFile, OffsetRange>, >> OutputReceiver<T>), as we have discussed in this thread, and that should be >> fine for SMB as well. It would require duplicating code from AvroSource, >> but in the end, I don't see it as a problem if AvroSource is going away. >> >> >> >> I'm attaching a small diagram I put for myself to better understand >> the code. >> >> >> >> AvroIO.readAll :: PTransform<PBegin, PCollection<T>> -> >> >> >> >> FileIO.matchAll :: PTransform<PCollection<String>, >> PCollection<MatchResult.Metadata>> >> >> FileIO.readMatches :: PTransform<PCollection<MatchResult.Metadata>, >> PCollection<ReadableFile>> >> >> AvroIO.readFiles :: PTransform<PCollection<FileIO.ReadableFile>, >> PCollection<T>> -> >> >> >> >> ReadAllViaFileBasedSource :: PTransform<PCollection<ReadableFile>, >> PCollection<T>> -> >> >> >> >> ParDo.of(SplitIntoRangesFn :: DoFn<ReadableFile, KV<ReadableFile, >> OffsetRange>>) (splittable do fn) >> >> >> >> Reshuffle.viaRandomKey() >> >> >> >> ParDo.of(ReadFileRangesFn(createSource) :: DoFn<KV<ReadableFile, >> OffsetRange>, T>) where >> >> >> >> createSource :: String -> FileBasedSource<T> >> >> >> >> createSource = AvroSource >> >> >> >> >> >> AvroIO.read without getHintMatchedManyFiles() :: PTransform<PBegin, >> PCollection<T>> -> >> >> >> >> Read.Bounded.from(createSource) where >> >> >> >> createSource :: String -> FileBasedSource<T> >> >> >> >> createSource = AvroSource >> >> >> >> >> >> Gleb >> >> >> >> >> >> On Thu, Jul 25, 2019 at 2:41 PM Robert Bradshaw <rober...@google.com> >> wrote: >> >>> >> >>> On Thu, Jul 25, 2019 at 12:35 AM Kenneth Knowles <k...@apache.org> >> wrote: >> >>> > >> >>> > From the peanut gallery, keeping a separate implementation for SMB >> seems fine. Dependencies are serious liabilities for both upstream and >> downstream. It seems like the reuse angle is generating extra work, and >> potentially making already-complex implementations more complex, instead of >> helping things. >> >>> >> >>> +1 >> >>> >> >>> To be clear, what I care about is that WriteFiles(X) and >> >>> WriteSmbFiles(X) can share the same X, for X in {Avro, Parquet, Text, >> >>> TFRecord, ...}. In other words composability of the API (vs. manually >> >>> filling out the matrix). If WriteFiles and WriteSmbFiles find >> >>> opportunities for (easy, clean) implementation sharing, that'd be >> >>> nice, but not the primary goal. >> >>> >> >>> (Similarly for reading, though that's seem less obvious. Certainly >> >>> whatever T is useful for ReadSmb(T) could be useful for a >> >>> (non-liquid-shading) ReadAll(T) however.) >> >>> >> >>> > On Wed, Jul 24, 2019 at 11:59 AM Neville Li <neville....@gmail.com> >> wrote: >> >>> >> >> >>> >> I spoke too soon. Turns out for unsharded writes, numShards can't >> be determined until the last finalize transform, which is again different >> from the current SMB proposal (static number of buckets & shards). >> >>> >> I'll end up with more code specialized for SMB in order to >> generalize existing sink code, which I think we all want to avoid. >> >>> >> >> >>> >> Seems the only option is duplicating some logic like temp file >> handling, which is exactly what we did in the original PR. >> >>> >> I can reuse Compression & Sink<T> for file level writes but that >> seems about the most I can reuse right now. >> >>> >> >> >>> >> On Tue, Jul 23, 2019 at 6:36 PM Neville Li <neville....@gmail.com> >> wrote: >> >>> >>> >> >>> >>> So I spent one afternoon trying some ideas for reusing the last >> few transforms WriteFiles. >> >>> >>> >> >>> >>> WriteShardsIntoTempFilesFn extends DoFn<KV<ShardedKey<Integer>, >> Iterable<UserT>>, FileResult<DestinationT>> >> >>> >>> => GatherResults<ResultT> extends >> PTransform<PCollection<ResultT>, PCollection<List<ResultT>>> >> >>> >>> => FinalizeTempFileBundles extends >> PTransform<PCollection<List<FileResult<DestinationT>>>, >> WriteFilesResult<DestinationT>> >> >>> >>> >> >>> >>> I replaced FileResult<DestinationT> with KV<DestinationT, >> ResourceId> so I can use pre-compute SMB destination file names for the >> transforms. >> >>> >>> I'm also thinking of parameterizing ShardedKey<Integer> for SMB's >> bucket/shard to reuse WriteShardsIntoTempFilesFn. These transforms are >> private and easy to change/pull out. >> >>> >>> >> >>> >>> OTOH they are somewhat coupled with the package private >> {Avro,Text,TFRecord}Sink and their WriteOperation impl (where the bulk of >> temp file handing logic lives). Might be hard to decouple either modifying >> existing code or creating new transforms, unless if we re-write most of >> FileBasedSink from scratch. >> >>> >>> >> >>> >>> Let me know if I'm on the wrong track. >> >>> >>> >> >>> >>> WIP Branch >> https://github.com/spotify/beam/tree/neville/write-files >> >>> >>> >> >>> >>> On Tue, Jul 23, 2019 at 4:22 PM Chamikara Jayalath < >> chamik...@google.com> wrote: >> >>> >>>> >> >>> >>>> >> >>> >>>> >> >>> >>>> On Mon, Jul 22, 2019 at 1:41 PM Robert Bradshaw < >> rober...@google.com> wrote: >> >>> >>>>> >> >>> >>>>> On Mon, Jul 22, 2019 at 7:39 PM Eugene Kirpichov < >> kirpic...@google.com> wrote: >> >>> >>>>> > >> >>> >>>>> > On Mon, Jul 22, 2019 at 7:49 AM Robert Bradshaw < >> rober...@google.com> wrote: >> >>> >>>>> >> >> >>> >>>>> >> On Mon, Jul 22, 2019 at 4:04 PM Neville Li < >> neville....@gmail.com> wrote: >> >>> >>>>> >> > >> >>> >>>>> >> > Thanks Robert. Agree with the FileIO point. I'll look into >> it and see what needs to be done. >> >>> >>>>> >> > >> >>> >>>>> >> > Eugene pointed out that we shouldn't build on >> FileBased{Source,Sink}. So for writes I'll probably build on top of >> WriteFiles. >> >>> >>>>> >> >> >>> >>>>> >> Meaning it could be parameterized by FileIO.Sink, right? >> >>> >>>>> >> >> >>> >>>>> >> >> https://github.com/apache/beam/blob/release-2.13.0/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileIO.java#L779 >> >>> >>>>> > >> >>> >>>>> > Yeah if possible, parameterize FileIO.Sink. >> >>> >>>>> > I would recommend against building on top of WriteFiles >> either. FileIO being implemented on top of WriteFiles was supposed to be a >> temporary measure - the longer-term plan was to rewrite it from scratch >> (albeit with a similar structure) and throw away WriteFiles. >> >>> >>>>> > If possible, I would recommend to pursue this path: if there >> are parts of WriteFiles you want to reuse, I would recommend to implement >> them as new transforms, not at all tied to FileBasedSink (but ok if tied to >> FileIO.Sink), with the goal in mind that FileIO could be rewritten on top >> of these new transforms, or maybe parts of WriteFiles could be swapped out >> for them incrementally. >> >>> >>>>> >> >>> >>>>> Thanks for the feedback. There's a lot that was done, but >> looking at >> >>> >>>>> the code it feels like there's a lot that was not yet done >> either, and >> >>> >>>>> the longer-term plan wasn't clear (though perhaps I'm just not >> finding >> >>> >>>>> the right docs). >> >>> >>>> >> >>> >>>> >> >>> >>>> I'm also a bit unfamiliar with original plans for WriteFiles and >> for updating source interfaces, but I prefer not significantly modifying >> existing IO transforms to suite the SMB use-case. If there are existing >> pieces of code that can be easily re-used that is fine, but existing >> sources/sinks are designed to perform a PCollection -> file transformation >> and vice versa with (usually) runner determined sharding. Things specific >> to SMB such as sharding restrictions, writing metadata to a separate file, >> reading multiple files from the same abstraction, does not sound like >> features that should be included in our usual file read/write transforms. >> >>> >>>> >> >>> >>>>> >> >>> >>>>> >> > Read might be a bigger change w.r.t. collocating ordered >> elements across files within a bucket and TBH I'm not even sure where to >> start. >> >>> >>>>> >> >> >>> >>>>> >> Yeah, here we need an interface that gives us ReadableFile -> >> >>> >>>>> >> Iterable<T>. There are existing >> PTransform<PCollection<ReadableFile>, >> >>> >>>>> >> PCollection<T>> but such an interface is insufficient to >> extract >> >>> >>>>> >> ordered records per shard. It seems the only concrete >> implementations >> >>> >>>>> >> are based on FileBasedSource, which we'd like to avoid, but >> there's no >> >>> >>>>> >> alternative. An SDF, if exposed, would likely be overkill and >> >>> >>>>> >> cumbersome to call (given the reflection machinery involved >> in >> >>> >>>>> >> invoking DoFns). >> >>> >>>>> > >> >>> >>>>> > Seems easiest to just define a new regular Java interface for >> this. >> >>> >>>>> > Could be either, indeed, ReadableFile -> Iterable<T>, or >> something analogous, e.g. (ReadableFile, OutputReceiver<T>) -> void. >> Depends on how much control over iteration you need. >> >>> >>>>> >> >>> >>>>> For this application, one wants to iterate over several files in >> >>> >>>>> parallel. The downside of a new interface is that it shares >> almost >> >>> >>>>> nothing with the "normal" sources (e.g. when features (or >> >>> >>>>> optimizations) get added to one, they won't get added to the >> other). >> >>> >>>> >> >>> >>>> >> >>> >>>>> >> >>> >>>>> >> >>> >>>>> > And yes, DoFn's including SDF's are not designed to be used >> as Java interfaces per se. If you need DoFn machinery in this interface >> (e.g. side inputs), use Contextful - s.apache.org/context-fn. >> >>> >>>>> >> >>> >>>>> Yeah, one of the primary downsides to the NewDoFns is how hard >> it is >> >>> >>>>> to build new DoFns out of others (or, really, use them in any >> context >> >>> >>>>> other than as an argument to ParDo). >> >>> >>>>> >> >>> >>>>> >> > I'll file separate PRs for core changes needed for >> discussion. WDYT? >> >>> >>>>> >> >> >>> >>>>> >> Sounds good. >> >>> >>>> >> >>> >>>> >> >>> >>>> +1 >> >>> >>>> >> >>> >>>>> >> >>> >>>>> >> >> >>> >>>>> >> > On Mon, Jul 22, 2019 at 4:20 AM Robert Bradshaw < >> rober...@google.com> wrote: >> >>> >>>>> >> >> >> >>> >>>>> >> >> On Fri, Jul 19, 2019 at 5:16 PM Neville Li < >> neville....@gmail.com> wrote: >> >>> >>>>> >> >> > >> >>> >>>>> >> >> > Forking this thread to discuss action items regarding >> the change. We can keep technical discussion in the original thread. >> >>> >>>>> >> >> > >> >>> >>>>> >> >> > Background: our SMB POC showed promising performance & >> cost saving improvements and we'd like to adopt it for production soon (by >> EOY). We want to contribute it to Beam so it's better generalized and >> maintained. We also want to avoid divergence between our internal version >> and the PR while it's in progress, specifically any breaking change in the >> produced SMB data. >> >>> >>>>> >> >> >> >>> >>>>> >> >> All good goals. >> >>> >>>>> >> >> >> >>> >>>>> >> >> > To achieve that I'd like to propose a few action items. >> >>> >>>>> >> >> > >> >>> >>>>> >> >> > 1. Reach a consensus about bucket and shard strategy, >> key handling, bucket file and metadata format, etc., anything that affect >> produced SMB data. >> >>> >>>>> >> >> > 2. Revise the existing PR according to #1 >> >>> >>>>> >> >> > 3. Reduce duplicate file IO logic by reusing >> FileIO.Sink, Compression, etc., but keep the existing file level abstraction >> >>> >>>>> >> >> > 4. (Optional) Merge code into extensions::smb but mark >> clearly as @experimental >> >>> >>>>> >> >> > 5. Incorporate ideas from the discussion, e.g. >> ShardingFn, GroupByKeyAndSortValues, FileIO generalization, key URN, etc. >> >>> >>>>> >> >> > >> >>> >>>>> >> >> > #1-4 gives us something usable in the short term, while >> #1 guarantees that production data produced today are usable when #5 lands >> on master. #4 also gives early adopters a chance to give feedback. >> >>> >>>>> >> >> > Due to the scope of #5, it might take much longer and a >> couple of big PRs to achieve, which we can keep iterating on. >> >>> >>>>> >> >> > >> >>> >>>>> >> >> > What are your thoughts on this? >> >>> >>>>> >> >> >> >>> >>>>> >> >> I would like to see some resolution on the FileIO >> abstractions before >> >>> >>>>> >> >> merging into experimental. (We have a FileBasedSink that >> would mostly >> >>> >>>>> >> >> already work, so it's a matter of coming up with an >> analogous Source >> >>> >>>>> >> >> interface.) Specifically I would not want to merge a set >> of per file >> >>> >>>>> >> >> type smb IOs without a path forward to this or the >> determination that >> >>> >>>>> >> >> it's not possible/desirable. >> >> >> >> >> >> >> >> -- >> >> Cheers, >> >> Gleb >> >