On Wed, Feb 5, 2020 at 3:37 PM David Li <li.david...@gmail.com> wrote: > > Hi Antoine and Wes, > > Thanks for the feedback. Yes, we should definitely consider these as > separate features. > > I agree that it makes sense for the file API (or a derived API) to > expose a generic CacheRanges or PrebufferRanges API. It could then do > coalescing and prefetching as desired based on the actual underlying > data store. (Or, perhaps, the actual prefetch/coalesce steps could be > implemented as wrappers also implementing the file API, with > parameters to tune for concurrency level/resource consumption/backend > storage characteristics; underlying file implementations would just > treat this as a no-op. This is what we prototyped internally in C++ > before writing this up - a wrapper around S3File that naively > prefetched the entire file.) > > As a separate step, prefetching/caching should also make use of a > global (or otherwise shared) IO thread pool, so that parallel reads of > different files implicitly coordinate work with each other as well. > Then, you could queue up reads of several Parquet files, such that a > slow network call for one file doesn't block progress for other files, > without issuing reads for all of these files at once. > > It's unclear to me what readahead at the record batch level would > accomplish - Parquet reads each column chunk in a row group as a > whole, and if the row groups are large, then multiple record batches > would fall in the same row group, so then we wouldn't gain any > parallelism, no? (Admittedly, I'm not familiar with the internals > here.) > > The concurrency manager would apply mostly to networked filesystems, > yes. It's more an observation that some workloads issue fewer reads of > large ranges, and other workloads issue lots of reads on small ranges, > so concurrency should be limited based on estimated bandwidth, and not > purely on number of concurrent tasks. > > In this case, it sounds like there are at least the following concrete tasks: > - Adding a (no-op) CacheRanges method to the RandomAccessFile API.
I'll comment in more detail on some of the other items in due course, but I think this should be handled by an implementation of RandomAccessFile (that wraps a naked RandomAccessFile) with some additional methods, rather than adding this to the abstract RandomAccessFile interface, e.g. class CachingInputFile : public RandomAccessFile { public: CachingInputFile(std::shared_ptr<RandomAccessFile> naked_file); Status CacheRanges(...); }; etc. > - Providing a column range hint to Parquet, and compute column chunk > ranges to buffer via CacheRanges, > - Providing an implementation of CacheRanges (e.g. via a wrapper > RandomAccessFile) that coalesces ranges based on backend storage > characteristics, > - Providing an implementation of CacheRanges that actually prefetches > ranges in the background, > - Providing a shared I/O pool to coordinate such I/O across multiple files, > - Providing some mechanism to limit concurrency in this shared pool, > - Sending down all these hints from the Datasets implementation. > > Thanks, > David > > > On 2/5/20, Wes McKinney <wesmck...@gmail.com> wrote: > > I agree with separating the problem into its constituent concerns to > > make sure that we are developing appropriate abstractions. > > > > Speaking specifically about the Parquet codebase, the way that we > > access a particular ColumnChunk in a row group is fairly simplistic. > > See the ReaderProperties::GetStream method > > > > https://github.com/apache/arrow/blob/master/cpp/src/parquet/properties.cc#L28 > > > > Rather than naively issuing a Read command (either buffered or > > unbuffered, both bad for filesystems like S3), I think we need to > > insert an abstraction at the point where the column reader class > > requests an InputStream for its serialized column chunk data, which is > > right here > > > > https://github.com/apache/arrow/blob/master/cpp/src/parquet/file_reader.cc#L123 > > > > It seems like a stateful object that allows certain byte ranges of the > > file to be cached would do the trick. > > > > Next, an API needs to be provided for applications to indicate which > > column chunks they intend to read so that the contiguous byte ranges > > can be pre-buffered, preventing any column reader from issuing naked > > IO calls. Seems like this should happen at the ReaderProperties level. > > > > The IO scheduling seems like it should be abstracted away from the > > Parquet library itself. So there would be code similar to > > > > std::pair<int64_t, int64_t> read_ranges = ComputePrebufferedRanges(); > > RETURN_NOT_OK(caching_file->CacheRanges(read_ranges)); > > > > The IO scheduling can happen inside the implementation of CacheRanges. > > Then when the column reader is created it will grab a slice of the > > cached data rather than issuing IO calls. > > > > Let me know if this analysis makes sense > > > > - Wes > > > > On Wed, Feb 5, 2020 at 9:24 AM Antoine Pitrou <anto...@python.org> wrote: > >> > >> > >> Hi David, > >> > >> I think we should discuss this as individual features. > >> > >> > Read Coalescing: from Parquet metadata, we know exactly> which byte > >> > ranges of a file will be read, and can “cheatin the S3 IO > >> layer by fetching them in advance > >> > >> It seems there are two things here: coalescing individual reads, and > >> issuing them in advance. It seems those are separate concerns. > >> > >> - coalescing reads: should the IO layer expose a ReadRanges function to > >> issue several reads at once, which the Parquet layer can then exploit? > >> > >> - issuing reads in advance: isn't that solved by readahead *at the > >> record batch level* (not the IO block level)? > >> > >> > Concurrency Manager: rather than limit parallelism by number of > >> > outstanding tasks, we can instead limit the estimated bandwidth > >> > consumption, allowing better performance when read sizes are small. > >> > >> - Concurrency Manager: is this a per-source optimization, applying > >> mainly to networked filesystems? > >> > >> > >> I think we want to make sure that each feature brings progress, instead > >> of trying to lump everything at once in a big PR. > >> > >> Regards > >> > >> Antoine. > >> > >> > >> > >> Le 05/02/2020 à 14:32, David Li a écrit : > >> > Hello all, > >> > > >> > We've been following the Arrow Datasets project with great interest, > >> > especially as we have an in-house library with similar goals built on > >> > top of PyArrow. Recently, we noticed some discussion around optimizing > >> > I/O for such use cases (e.g. PARQUET-1698), which is also where we had > >> > focused our efforts. > >> > > >> > Our long-term goal has been to open-source our library. However, our > >> > code is in Python, but it would be most useful to everyone in the C++ > >> > core, so that R, Python, Ruby, etc. could benefit. Thus, we'd like to > >> > share our high-level design, and offer to work with the community on > >> > the implementation - at the very least, to avoid duplicating work. > >> > We've summarized our approach, and hope this can start a discussion on > >> > how to integrate such optimizations into Datasets: > >> > https://docs.google.com/document/d/1tZsT3dC7UXbLTkqxgVeFGWm9piXScUDujsa0ncvK_Fs/edit# > >> > > >> > At a high level, we have three main optimizations: > >> > - Given a set of columns to read, and potentially a filter on a > >> > partition key, we can use Parquet metadata to compute exact byte > >> > ranges to read from remote storage, and coalesce/split up reads as > >> > necessary based on the characteristics of the storage platform. > >> > - Given byte ranges to read, we can read them in parallel, using a > >> > global thread pool and concurrency manager to limit parallelism and > >> > resource consumption. > >> > - By working at the level of a dataset, we can parallelize these > >> > operations across files, and pipeline steps like reading Parquet > >> > metadata with reading and deserialization. > >> > > >> > We focus on Parquet and S3/object storage here, but these concepts > >> > apply to other file formats and storage systems. > >> > > >> > The main questions here are whether we think the optimizations are > >> > useful for Arrow Datasets, and if so, how the API design and > >> > implementation would proceed - I'd appreciate any feedback on the > >> > approach here and potential API. > >> > > >> > David > >> > > >