On Wed, Oct 16, 2019 at 12:32 PM John Muehlhausen <j...@jgm.org> wrote: > > I really need to "get into the zone" on some other development today, but I > want to remind us of something earlier in the thread that gave me the > impression I wasn't stomping on too many paradigms with this proposal: > > Wes: ``So the "length" field in RecordBatch is already the utilized number > of rows. The body buffers can certainly have excess unused space. So > your application can mutate Flatbuffer "length" field in-place as > new records are filled in.''
I'm objecting to RecordBatch.length being inconsistent with the constituent field lengths, that's where the danger lies. If all of the lengths are consistent, no code changes are necessary. > If RecordBatch.length is the utilized number of rows then my PR makes this > actually true. Yes, we need it in a handful of implementations. I'm > willing to provide all of them. To me that is the lowest complexity > solution. > > -John > > On Wed, Oct 16, 2019 at 10:45 AM Wes McKinney <wesmck...@gmail.com> wrote: > > > On Wed, Oct 16, 2019 at 10:17 AM John Muehlhausen <j...@jgm.org> wrote: > > > > > > "pyarrow is intended as a developer-facing library, not a user-facing > > one" > > > > > > Is that really the core issue? I doubt you would want to add this > > proposed > > > logic to pandas even though it is user-facing, because then pandas will > > > either have to re-implement what it means to read a batch (to respect > > > length when it is smaller than array length) or else rely on the single > > > blessed custom metadata for doing this, which doesn't make it custom > > > anymore. > > > > What you have proposed in your PR amounts to an alteration of the IPC > > format to suit this use case. This pushes complexity onto _every_ > > implementation that will need to worry about a "truncated" record > > batch. I'd rather avoid this unless it is truly the only way. > > > > Note that we serialize a significant amount of custom metadata already > > to address pandas-specific issues, and have not had to make any > > changes to the columnar format as a result. > > > > > I think really your concern is that perhaps nobody wants this but me, > > > therefore it should not be in arrow or pandas regardless of whether it is > > > user-facing? But, if that is your thinking, is it true? What is our > > > solution to the locality/latency problem for systems that ingest and > > > process concurrently, if not this solution? I do see it as a general > > > problem that needs at least the beginnings of a general solution... not a > > > "custom" one. > > > > We use the custom_metadata fields to implement a number of built-in > > things in the project, such as extension types. If enough people find > > this useful, then it can be promoted to a formalized concept. As far > > as I can tell, you have developed quite a bit of custom code related > > to this for your application, including manipulating Flatbuffers > > metadata in place to maintain the populated length, so the barrier to > > entry to being able to properly take advantage of this is rather high. > > > > > Also, I wonder whether it is true that pyarrow avoids smart/magical > > > things. The entire concept of a "Table" seems to be in that category? > > The > > > docs specifically mention that it is for convenience. > > > > > > > Table arose out of legitimate developer need. There are a number of > > areas of the project that would be much more difficult if we had to > > worry about regularizing column chunking at any call site that returns > > an in-memory dataset. > > > > > I'd like to focus on two questions: > > > 1- What is the Arrow general solution to the locality/latency tradeoff > > > problem for systems that ingest and process data concurrently? This > > > proposed solution or something else? Or if we propose not to address the > > > problem, why? > > > 2- What will the proposed change negatively impact? It seems that all we > > > are talking about is respecting batch length if arrays happen to be > > longer. > > > > I'm suggesting to help you solve the post-read truncation problem > > without modifying the IPC protocol. If you want to make things work > > for the users without knowledge, I think this can be achieved through > > a plug-in API to define a metadata handler-callback to apply the > > truncation to the record batches. > > > > > Thanks, > > > -John > > > > > > On Wed, Oct 16, 2019 at 8:37 AM Wes McKinney <wesmck...@gmail.com> > > wrote: > > > > > > > hi John, > > > > > > > > > As a practical matter, the reason metadata is not a good solution > > for me > > > > is that it requires awareness on the part of the reader. I want > > (e.g.) a > > > > researcher in Python to be able to map a file of batches in IPC format > > > > without needing to worry about the fact that the file was built in a > > > > streaming fashion and therefore has some unused array elements. > > > > > > > > I don't find this argument to be persuasive. > > > > > > > > pyarrow is intended as a developer-facing library, not a user-facing > > > > one. I don't think you should be having the kinds of users you are > > > > describing using pyarrow directly, instead consuming the library > > > > through a layer above it. Specifically, we are deliberately avoiding > > > > doing anything too "smart" or "magical", instead maintaining tight > > > > developer control over what is going on. > > > > > > > > - Wes > > > > > > > > On Wed, Oct 16, 2019 at 2:18 AM Micah Kornfield <emkornfi...@gmail.com > > > > > > > wrote: > > > > > > > > > > Still thinking through the implications here, but to save others from > > > > > having to go search [1] is the PR. > > > > > > > > > > [1] https://github.com/apache/arrow/pull/5663/files > > > > > > > > > > On Tue, Oct 15, 2019 at 1:42 PM John Muehlhausen <j...@jgm.org> > > wrote: > > > > > > > > > > > A proposal with linked PR now exists in ARROW-5916 and Wes > > commented > > > > that > > > > > > we should kick it around some more. > > > > > > > > > > > > The high-level topic is how Apache Arrow intersects with streaming > > > > > > methodologies: > > > > > > > > > > > > If record batches are strictly immutable, a difficult trade-off is > > > > created > > > > > > for streaming data collection: either I can have low-latency > > > > presentation > > > > > > of new data by appending very small batches (often 1 row) to the > > IPC > > > > stream > > > > > > and lose columnar layout benefits, or I can have high-latency > > > > presentation > > > > > > of new data by waiting to append a batch until it is large enough > > to > > > > gain > > > > > > significant columnar layout benefits. During this waiting period > > the > > > > new > > > > > > data is unavailable to processing. > > > > > > > > > > > > If, on the other hand, [0,length) of a batch is immutable but > > length > > > > may > > > > > > increase, the trade-off is eliminated: I can pre-allocate a batch > > and > > > > > > populate records in it when they occur (without waiting), and also > > gain > > > > > > columnar benefits as each "closed" batch will be large. (A batch > > may > > > > be > > > > > > practically "closed" before the arrays are full when the > > projection of > > > > > > variable-length buffer space is wrong... a space/time tradeoff in > > > > favor of > > > > > > time.) > > > > > > > > > > > > Looking ahead to a day when the reference implementation(s) will be > > > > able to > > > > > > bump RecordBatch.length while populating pre-allocated records > > > > > > in-place, ARROW-5916 reads such batches by ignoring portions of > > arrays > > > > that > > > > > > are beyond RecordBatch.length. > > > > > > > > > > > > If we are not looking ahead to such a day, the discussion is about > > the > > > > > > alternative way that Arrow will avoid the latency/locality tradeoff > > > > > > inherent in streaming data collection. Or, if the answer is > > "streaming > > > > > > apps are and will always be out of scope", that idea needs to be > > > > defended > > > > > > from the observation that practitioners are moving more towards the > > > > fusion > > > > > > of batch and streaming, not away from it. > > > > > > > > > > > > As a practical matter, the reason metadata is not a good solution > > for > > > > me is > > > > > > that it requires awareness on the part of the reader. I want > > (e.g.) a > > > > > > researcher in Python to be able to map a file of batches in IPC > > format > > > > > > without needing to worry about the fact that the file was built in > > a > > > > > > streaming fashion and therefore has some unused array elements. > > > > > > > > > > > > The change itself seems relatively simple. What negative > > consequences > > > > do > > > > > > we anticipate, if any? > > > > > > > > > > > > Thanks, > > > > > > -John > > > > > > > > > > > > On Fri, Jul 5, 2019 at 10:42 AM John Muehlhausen <j...@jgm.org> > > wrote: > > > > > > > > > > > > > This seems to help... still testing it though. > > > > > > > > > > > > > > Status GetFieldMetadata(int field_index, ArrayData* out) { > > > > > > > auto nodes = metadata_->nodes(); > > > > > > > // pop off a field > > > > > > > if (field_index >= static_cast<int>(nodes->size())) { > > > > > > > return Status::Invalid("Ran out of field metadata, likely > > > > > > > malformed"); > > > > > > > } > > > > > > > const flatbuf::FieldNode* node = nodes->Get(field_index); > > > > > > > > > > > > > > * //out->length = node->length();* > > > > > > > * out->length = metadata_->length();* > > > > > > > out->null_count = node->null_count(); > > > > > > > out->offset = 0; > > > > > > > return Status::OK(); > > > > > > > } > > > > > > > > > > > > > > On Fri, Jul 5, 2019 at 10:24 AM John Muehlhausen <j...@jgm.org> > > > > wrote: > > > > > > > > > > > > > >> So far it seems as if pyarrow is completely ignoring the > > > > > > >> RecordBatch.length field. More info to follow... > > > > > > >> > > > > > > >> On Tue, Jul 2, 2019 at 3:02 PM John Muehlhausen <j...@jgm.org> > > > > wrote: > > > > > > >> > > > > > > >>> Crikey! I'll do some testing around that and suggest some test > > > > cases to > > > > > > >>> ensure it continues to work, assuming that it does. > > > > > > >>> > > > > > > >>> -John > > > > > > >>> > > > > > > >>> On Tue, Jul 2, 2019 at 2:41 PM Wes McKinney < > > wesmck...@gmail.com> > > > > > > wrote: > > > > > > >>> > > > > > > >>>> Thanks for the attachment, it's helpful. > > > > > > >>>> > > > > > > >>>> On Tue, Jul 2, 2019 at 1:40 PM John Muehlhausen <j...@jgm.org> > > > > wrote: > > > > > > >>>> > > > > > > > >>>> > Attachments referred to in previous two messages: > > > > > > >>>> > > > > > > > >>>> > > > > > > > > > > > > https://www.dropbox.com/sh/6ycfuivrx70q2jx/AAAt-RDaZWmQ2VqlM-0s6TqWa?dl=0 > > > > > > >>>> > > > > > > > >>>> > On Tue, Jul 2, 2019 at 1:14 PM John Muehlhausen < > > j...@jgm.org> > > > > > > wrote: > > > > > > >>>> > > > > > > > >>>> > > Thanks, Wes, for the thoughtful reply. I really > > appreciate > > > > the > > > > > > >>>> > > engagement. In order to clarify things a bit, I am > > attaching > > > > a > > > > > > >>>> graphic of > > > > > > >>>> > > how our application will take record-wise (row-oriented) > > data > > > > from > > > > > > >>>> an event > > > > > > >>>> > > source and incrementally populate a pre-allocated > > > > Arrow-compatible > > > > > > >>>> buffer, > > > > > > >>>> > > including for variable-length fields. (Obviously at this > > > > stage I > > > > > > >>>> am not > > > > > > >>>> > > using the reference implementation Arrow code, although > > that > > > > would > > > > > > >>>> be a > > > > > > >>>> > > goal.... to contribute that back to the project.) > > > > > > >>>> > > > > > > > > >>>> > > For sake of simplicity these are non-nullable fields. As > > a > > > > > > result a > > > > > > >>>> > > reader of "y" that has no knowledge of the "utilized" > > metadata > > > > > > >>>> would get a > > > > > > >>>> > > long string (zeros, spaces, uninitialized, or whatever we > > > > decide > > > > > > >>>> for the > > > > > > >>>> > > pre-allocation model) for the record just beyond the last > > > > utilized > > > > > > >>>> record. > > > > > > >>>> > > > > > > > > >>>> > > I don't see any "big O"-analysis problems with this > > > > approach. The > > > > > > >>>> > > space/time tradeoff is that we have to guess how much > > room to > > > > > > >>>> allocate for > > > > > > >>>> > > variable-length fields. We will probably almost always be > > > > wrong. > > > > > > >>>> This > > > > > > >>>> > > ends up in "wasted" space. However, we can do > > calculations > > > > based > > > > > > >>>> on these > > > > > > >>>> > > partially filled batches that take full advantage of the > > > > columnar > > > > > > >>>> layout. > > > > > > >>>> > > (Here I've shown the case where we had too little > > > > variable-length > > > > > > >>>> buffer > > > > > > >>>> > > set aside, resulting in "wasted" rows. The flip side is > > that > > > > rows > > > > > > >>>> achieve > > > > > > >>>> > > full [1] utilization but there is wasted variable-length > > > > buffer if > > > > > > >>>> we guess > > > > > > >>>> > > incorrectly in the other direction.) > > > > > > >>>> > > > > > > > > >>>> > > I proposed a few things that are "nice to have" but really > > > > what > > > > > > I'm > > > > > > >>>> eyeing > > > > > > >>>> > > is the ability for a reader-- any reader (e.g. pyarrow)-- > > to > > > > see > > > > > > >>>> that some > > > > > > >>>> > > of the rows in a RecordBatch are not to be read, based on > > the > > > > new > > > > > > >>>> > > "utilized" (or whatever name) metadata. That single > > tweak to > > > > the > > > > > > >>>> > > metadata-- and readers honoring it-- is the core of the > > > > proposal. > > > > > > >>>> > > (Proposal 4.) This would indicate that the attached > > example > > > > (or > > > > > > >>>> something > > > > > > >>>> > > similar) is the blessed approach for those seeking to > > > > accumulate > > > > > > >>>> events and > > > > > > >>>> > > process them while still expecting more data, with the > > > > > > >>>> heavier-weight task > > > > > > >>>> > > of creating a new pre-allocated batch being a rare > > occurrence. > > > > > > >>>> > > > > > > > > >>>> > > > > > > >>>> So the "length" field in RecordBatch is already the utilized > > > > number of > > > > > > >>>> rows. The body buffers can certainly have excess unused > > space. So > > > > your > > > > > > >>>> application can mutate Flatbuffer "length" field in-place as > > new > > > > > > >>>> records are filled in. > > > > > > >>>> > > > > > > >>>> > > Notice that the mutability is only in the sense of > > > > "appending." > > > > > > The > > > > > > >>>> > > current doctrine of total immutability would be revised to > > > > refer > > > > > > to > > > > > > >>>> the > > > > > > >>>> > > immutability of only the already-populated rows. > > > > > > >>>> > > > > > > > > >>>> > > It gives folks an option other than choosing the lesser > > of two > > > > > > >>>> evils: on > > > > > > >>>> > > the one hand, length 1 RecordBatches that don't result in > > a > > > > stream > > > > > > >>>> that is > > > > > > >>>> > > computationally efficient. On the other hand, adding > > > > artificial > > > > > > >>>> latency by > > > > > > >>>> > > accumulating events before "freezing" a larger batch and > > only > > > > then > > > > > > >>>> making > > > > > > >>>> > > it available to computation. > > > > > > >>>> > > > > > > > > >>>> > > -John > > > > > > >>>> > > > > > > > > >>>> > > On Tue, Jul 2, 2019 at 12:21 PM Wes McKinney < > > > > wesmck...@gmail.com > > > > > > > > > > > > > >>>> wrote: > > > > > > >>>> > > > > > > > > >>>> > >> hi John, > > > > > > >>>> > >> > > > > > > >>>> > >> On Tue, Jul 2, 2019 at 11:23 AM John Muehlhausen < > > > > j...@jgm.org> > > > > > > >>>> wrote: > > > > > > >>>> > >> > > > > > > > >>>> > >> > During my time building financial analytics and trading > > > > systems > > > > > > >>>> (23 > > > > > > >>>> > >> years!), both the "batch processing" and "stream > > processing" > > > > > > >>>> paradigms have > > > > > > >>>> > >> been extensively used by myself and by colleagues. > > > > > > >>>> > >> > > > > > > > >>>> > >> > Unfortunately, the tools used in these paradigms have > > not > > > > > > >>>> successfully > > > > > > >>>> > >> overlapped. For example, an analyst might use a Python > > > > notebook > > > > > > >>>> with > > > > > > >>>> > >> pandas to do some batch analysis. Then, for acceptable > > > > latency > > > > > > and > > > > > > >>>> > >> throughput, a C++ programmer must implement the same > > schemas > > > > and > > > > > > >>>> processing > > > > > > >>>> > >> logic in order to analyze real-time data for real-time > > > > decision > > > > > > >>>> support. > > > > > > >>>> > >> (Time horizons often being sub-second or even > > > > sub-millisecond for > > > > > > >>>> an > > > > > > >>>> > >> acceptable reaction to an event. The most aggressive > > > > > > >>>> software-based > > > > > > >>>> > >> systems, leaving custom hardware aside other than things > > like > > > > > > >>>> kernel-bypass > > > > > > >>>> > >> NICs, target 10s of microseconds for a full round trip > > from > > > > data > > > > > > >>>> ingestion > > > > > > >>>> > >> to decision.) > > > > > > >>>> > >> > > > > > > > >>>> > >> > As a result, TCO is more than doubled. A doubling can > > be > > > > > > >>>> accounted for > > > > > > >>>> > >> by two implementations that share little or nothing in > > the > > > > way of > > > > > > >>>> > >> architecture. Then additional effort is required to > > ensure > > > > that > > > > > > >>>> these > > > > > > >>>> > >> implementations continue to behave the same way and are > > > > upgraded > > > > > > in > > > > > > >>>> > >> lock-step. > > > > > > >>>> > >> > > > > > > > >>>> > >> > Arrow purports to be a "bridge" technology that eases > > one > > > > of > > > > > > the > > > > > > >>>> pain > > > > > > >>>> > >> points of working in different ecosystems by providing a > > > > common > > > > > > >>>> event > > > > > > >>>> > >> stream data structure. (Discussion of common processing > > > > > > >>>> techniques is > > > > > > >>>> > >> beyond the scope of this discussion. Suffice it to say > > that > > > > a > > > > > > >>>> streaming > > > > > > >>>> > >> algo can always be run in batch, but not vice versa.) > > > > > > >>>> > >> > > > > > > > >>>> > >> > Arrow seems to be growing up primarily in the batch > > > > processing > > > > > > >>>> world. > > > > > > >>>> > >> One publication notes that "the missing piece is > > streaming, > > > > where > > > > > > >>>> the > > > > > > >>>> > >> velocity of incoming data poses a special challenge. > > There > > > > are > > > > > > >>>> some early > > > > > > >>>> > >> experiments to populate Arrow nodes in microbatches..." > > [1] > > > > Part > > > > > > >>>> our our > > > > > > >>>> > >> discussion could be a response to this observation. In > > what > > > > ways > > > > > > >>>> is it > > > > > > >>>> > >> true or false? What are the plans to remedy this > > > > shortcoming, if > > > > > > >>>> it > > > > > > >>>> > >> exists? What steps can be taken now to ease the > > transition > > > > to > > > > > > >>>> low-latency > > > > > > >>>> > >> streaming support in the future? > > > > > > >>>> > >> > > > > > > > >>>> > >> > > > > > > >>>> > >> Arrow columnar format describes a collection of records > > with > > > > > > values > > > > > > >>>> > >> between records being placed adjacent to each other in > > > > memory. If > > > > > > >>>> you > > > > > > >>>> > >> break that assumption, you don't have a columnar format > > > > anymore. > > > > > > >>>> So I > > > > > > >>>> > >> don't where the "shortcoming" is. We don't have any > > software > > > > in > > > > > > the > > > > > > >>>> > >> project for managing the creation of record batches in a > > > > > > streaming > > > > > > >>>> > >> application, but this seems like an interesting > > development > > > > > > >>>> expansion > > > > > > >>>> > >> area for the project. > > > > > > >>>> > >> > > > > > > >>>> > >> Note that many contributors have already expanded the > > surface > > > > > > area > > > > > > >>>> of > > > > > > >>>> > >> what's in the Arrow libraries in many directions. > > > > > > >>>> > >> > > > > > > >>>> > >> Streaming data collection is yet another area of > > expansion, > > > > but > > > > > > >>>> > >> _personally_ it is not on the short list of projects > > that I > > > > will > > > > > > >>>> > >> personally be working on (or asking my direct or indirect > > > > > > >>>> colleagues > > > > > > >>>> > >> to work on). Since this is a project made up of > > volunteers, > > > > it's > > > > > > >>>> up to > > > > > > >>>> > >> contributors to drive new directions for the project by > > > > writing > > > > > > >>>> design > > > > > > >>>> > >> documents and pull requests. > > > > > > >>>> > >> > > > > > > >>>> > >> > In my own experience, a successful strategy for stream > > > > > > >>>> processing where > > > > > > >>>> > >> context (i.e. recent past events) must be considered by > > > > > > >>>> calculations is to > > > > > > >>>> > >> pre-allocate memory for event collection, to organize > > this > > > > memory > > > > > > >>>> in a > > > > > > >>>> > >> columnar layout, and to run incremental calculations at > > each > > > > > > event > > > > > > >>>> ingress > > > > > > >>>> > >> into the partially populated memory. [Fig 1] When the > > > > > > >>>> pre-allocated > > > > > > >>>> > >> memory has been exhausted, allocate a new batch of > > > > column-wise > > > > > > >>>> memory and > > > > > > >>>> > >> continue. When a batch is no longer pertinent to the > > > > calculation > > > > > > >>>> look-back > > > > > > >>>> > >> window, free the memory back to the heap or pool. > > > > > > >>>> > >> > > > > > > > >>>> > >> > Here we run into the first philosophical barrier with > > > > Arrow, > > > > > > >>>> where > > > > > > >>>> > >> "Arrow data is immutable." [2] There is currently > > little or > > > > no > > > > > > >>>> > >> consideration for reading a partially constructed > > > > RecordBatch, > > > > > > >>>> e.g. one > > > > > > >>>> > >> with only some of the rows containing event data at the > > > > present > > > > > > >>>> moment in > > > > > > >>>> > >> time. > > > > > > >>>> > >> > > > > > > > >>>> > >> > > > > > > >>>> > >> It seems like the use case you have heavily revolves > > around > > > > > > >>>> mutating > > > > > > >>>> > >> pre-allocated, memory-mapped datasets that are being > > > > consumed by > > > > > > >>>> other > > > > > > >>>> > >> processes on the same host. So you want to incrementally > > fill > > > > > > some > > > > > > >>>> > >> memory-mapped data that you've already exposed to another > > > > > > process. > > > > > > >>>> > >> > > > > > > >>>> > >> Because of the memory layout for variable-size and nested > > > > cells, > > > > > > >>>> it is > > > > > > >>>> > >> impossible in general to mutate Arrow record batches. > > This is > > > > > > not a > > > > > > >>>> > >> philosophical position: this was a deliberate technical > > > > decision > > > > > > to > > > > > > >>>> > >> guarantee data locality for scans and predictable O(1) > > random > > > > > > >>>> access > > > > > > >>>> > >> on variable-length and nested data. > > > > > > >>>> > >> > > > > > > >>>> > >> Technically speaking, you can mutate memory in-place for > > > > > > fixed-size > > > > > > >>>> > >> types in-RAM or on-disk, if you want to. It's an > > "off-label" > > > > use > > > > > > >>>> case > > > > > > >>>> > >> but no one is saying you can't do this. > > > > > > >>>> > >> > > > > > > >>>> > >> > Proposal 1: Shift the Arrow "immutability" doctrine to > > > > apply to > > > > > > >>>> > >> populated records of a RecordBatch instead of to all > > records? > > > > > > >>>> > >> > > > > > > > >>>> > >> > > > > > > >>>> > >> Per above, this is impossible in generality. You can't > > alter > > > > > > >>>> > >> variable-length or nested records without rewriting the > > > > record > > > > > > >>>> batch. > > > > > > >>>> > >> > > > > > > >>>> > >> > As an alternative approach, RecordBatch can be used as > > a > > > > single > > > > > > >>>> Record > > > > > > >>>> > >> (batch length of one). [Fig 2] In this approach the > > > > benefit of > > > > > > >>>> the > > > > > > >>>> > >> columnar layout is lost for look-back window processing. > > > > > > >>>> > >> > > > > > > > >>>> > >> > Another alternative approach is to collect an entire > > > > > > RecordBatch > > > > > > >>>> before > > > > > > >>>> > >> stepping through it with the stream processing > > calculation. > > > > [Fig > > > > > > >>>> 3] With > > > > > > >>>> > >> this approach some columnar processing benefit can be > > > > recovered, > > > > > > >>>> however > > > > > > >>>> > >> artificial latency is introduced. As tolerance for > > delays in > > > > > > >>>> decision > > > > > > >>>> > >> support dwindles, this model will be of increasingly > > limited > > > > > > >>>> value. It is > > > > > > >>>> > >> already unworkable in many areas of finance. > > > > > > >>>> > >> > > > > > > > >>>> > >> > When considering the Arrow format and variable length > > > > values > > > > > > >>>> such as > > > > > > >>>> > >> strings, the pre-allocation approach (and subsequent > > > > processing > > > > > > of > > > > > > >>>> a > > > > > > >>>> > >> partially populated batch) encounters a hiccup. How do > > we > > > > know > > > > > > >>>> the amount > > > > > > >>>> > >> of buffer space to pre-allocate? If we allocate too much > > > > buffer > > > > > > >>>> for > > > > > > >>>> > >> variable-length data, some of it will be unused. If we > > > > allocate > > > > > > >>>> too little > > > > > > >>>> > >> buffer for variable-length data, some row entities will > > be > > > > > > >>>> unusable. > > > > > > >>>> > >> (Additional "rows" remain but when populating string > > fields > > > > there > > > > > > >>>> is no > > > > > > >>>> > >> longer string storage space to point them to.) > > > > > > >>>> > >> > > > > > > > >>>> > >> > As with many optimization space/time tradeoff > > problems, the > > > > > > >>>> solution > > > > > > >>>> > >> seems to be to guess. Pre-allocation sets aside variable > > > > length > > > > > > >>>> buffer > > > > > > >>>> > >> storage based on the typical "expected size" of the > > variable > > > > > > >>>> length data. > > > > > > >>>> > >> This can result in some unused rows, as discussed above. > > > > [Fig 4] > > > > > > >>>> In fact > > > > > > >>>> > >> it will necessarily result in one unused row unless the > > last > > > > of > > > > > > >>>> each > > > > > > >>>> > >> variable length field in the last row exactly fits into > > the > > > > > > >>>> remaining space > > > > > > >>>> > >> in the variable length data buffer. Consider the case > > where > > > > > > there > > > > > > >>>> is more > > > > > > >>>> > >> variable length buffer space than data: > > > > > > >>>> > >> > > > > > > > >>>> > >> > Given variable-length field x, last row index of y, > > > > variable > > > > > > >>>> length > > > > > > >>>> > >> buffer v, beginning offset into v of o: > > > > > > >>>> > >> > x[y] begins at o > > > > > > >>>> > >> > x[y] ends at the offset of the next record, there > > is no > > > > > > next > > > > > > >>>> > >> record, so x[y] ends after the total remaining area in > > > > variable > > > > > > >>>> length > > > > > > >>>> > >> buffer... however, this is too much! > > > > > > >>>> > >> > > > > > > > >>>> > >> > > > > > > >>>> > >> It isn't clear to me what you're proposing. It sounds > > like > > > > you > > > > > > >>>> want a > > > > > > >>>> > >> major redesign of the columnar format to permit in-place > > > > mutation > > > > > > >>>> of > > > > > > >>>> > >> strings. I doubt that would be possible at this point. > > > > > > >>>> > >> > > > > > > >>>> > >> > Proposal 2: [low priority] Create an "expected length" > > > > > > statistic > > > > > > >>>> in the > > > > > > >>>> > >> Schema for variable length fields? > > > > > > >>>> > >> > > > > > > > >>>> > >> > Proposal 3: [low priority] Create metadata to store the > > > > index > > > > > > >>>> into > > > > > > >>>> > >> variable-length data that represents the end of the value > > > > for the > > > > > > >>>> last > > > > > > >>>> > >> record? Alternatively: a row is "wasted," however > > > > pre-allocation > > > > > > >>>> is > > > > > > >>>> > >> inexact to begin with. > > > > > > >>>> > >> > > > > > > > >>>> > >> > Proposal 4: Add metadata to indicate to a RecordBatch > > > > reader > > > > > > >>>> that only > > > > > > >>>> > >> some of the rows are to be utilized. [Fig 5] This is > > > > useful not > > > > > > >>>> only when > > > > > > >>>> > >> processing a batch that is still under construction, but > > > > also for > > > > > > >>>> "closed" > > > > > > >>>> > >> batches that were not able to be fully populated due to > > an > > > > > > >>>> imperfect > > > > > > >>>> > >> projection of variable length storage. > > > > > > >>>> > >> > > > > > > > >>>> > >> > On this last proposal, Wes has weighed in: > > > > > > >>>> > >> > > > > > > > >>>> > >> > "I believe your use case can be addressed by > > pre-allocating > > > > > > >>>> record > > > > > > >>>> > >> batches and maintaining application level metadata about > > what > > > > > > >>>> portion of > > > > > > >>>> > >> the record batches has been 'filled' (so the unfilled > > > > records can > > > > > > >>>> be > > > > > > >>>> > >> dropped by slicing). I don't think any change to the > > binary > > > > > > >>>> protocol is > > > > > > >>>> > >> warranted." [3] > > > > > > >>>> > >> > > > > > > > >>>> > >> > > > > > > >>>> > >> My personal opinion is that a solution to the problem you > > > > have > > > > > > can > > > > > > >>>> be > > > > > > >>>> > >> composed from the components (combined with some new > > pieces > > > > of > > > > > > >>>> code) > > > > > > >>>> > >> that we have developed in the project already. > > > > > > >>>> > >> > > > > > > >>>> > >> So the "application level" could be an add-on C++ > > component > > > > in > > > > > > the > > > > > > >>>> > >> Apache Arrow project. Call it a "memory-mapped streaming > > data > > > > > > >>>> > >> collector" that pre-allocates on-disk record batches (of > > only > > > > > > >>>> > >> fixed-size or even possibly dictionary-encoded types) and > > > > then > > > > > > >>>> fills > > > > > > >>>> > >> them incrementally as bits of data come in, updating some > > > > > > auxiliary > > > > > > >>>> > >> metadata that other processes can use to determine what > > > > portion > > > > > > of > > > > > > >>>> the > > > > > > >>>> > >> Arrow IPC messages to "slice off". > > > > > > >>>> > >> > > > > > > >>>> > >> > Concerns with positioning this at the app level: > > > > > > >>>> > >> > > > > > > > >>>> > >> > 1- Do we need to address or begin to address the > > overall > > > > > > concern > > > > > > >>>> of how > > > > > > >>>> > >> Arrow data structures are to be used in "true" > > > > (non-microbatch) > > > > > > >>>> streaming > > > > > > >>>> > >> environments, cf [1] in the last paragraph, as a > > > > *first-class* > > > > > > >>>> usage > > > > > > >>>> > >> pattern? If so, is now the time? > > > > > > >>>> > >> >if you break that design invariant you don't have a > > columnar > > > > > > >>>> format > > > > > > >>>> > >> anymore. > > > > > > >>>> > >> > > > > > > >>>> > >> Arrow provides a binary protocol for describing a payload > > > > data on > > > > > > >>>> the > > > > > > >>>> > >> wire (or on-disk, or in-memory, all the same). I don't > > see > > > > how it > > > > > > >>>> is > > > > > > >>>> > >> in conflict with streaming environments, unless the > > streaming > > > > > > >>>> > >> application has difficulty collecting multiple records > > into > > > > an > > > > > > >>>> Arrow > > > > > > >>>> > >> record batches. In that case, it's a system trade-off. > > > > Currently > > > > > > >>>> > >> people are using Avro with Kafka and sending one record > > at a > > > > > > time, > > > > > > >>>> but > > > > > > >>>> > >> then they're also spending a lot of CPU cycles in > > > > serialization. > > > > > > >>>> > >> > > > > > > >>>> > >> > 2- If we can even make broad-stroke attempts at data > > > > structure > > > > > > >>>> features > > > > > > >>>> > >> that are likely to be useful when streaming becomes a > > first > > > > class > > > > > > >>>> citizen, > > > > > > >>>> > >> it reduces the chances of "breaking" format changes in > > the > > > > > > >>>> future. I do > > > > > > >>>> > >> not believe the proposals place an undue hardship on > > batch > > > > > > >>>> processing > > > > > > >>>> > >> paradigms. We are currently discussing making a breaking > > > > change > > > > > > >>>> to the IPC > > > > > > >>>> > >> format [4], so there is a window of opportunity to > > consider > > > > > > >>>> features useful > > > > > > >>>> > >> for streaming? (Current clients can feel free to ignore > > the > > > > > > >>>> proposed > > > > > > >>>> > >> "utilized" metadata of RecordBatch.) > > > > > > >>>> > >> > > > > > > > >>>> > >> > > > > > > >>>> > >> I think the perception that streaming is not a first > > class > > > > > > citizen > > > > > > >>>> is > > > > > > >>>> > >> an editorialization (e.g. the article you cited was an > > > > editorial > > > > > > >>>> > >> written by an industry analyst based on an interview with > > > > Jacques > > > > > > >>>> and > > > > > > >>>> > >> me). Columnar data formats in general are designed to > > work > > > > with > > > > > > >>>> more > > > > > > >>>> > >> than one value at a time (which we are calling a "batch" > > but > > > > I > > > > > > >>>> think > > > > > > >>>> > >> that's conflating terminology with the "batch processing" > > > > > > paradigm > > > > > > >>>> of > > > > > > >>>> > >> Hadoop, etc.), > > > > > > >>>> > >> > > > > > > >>>> > >> > 3- Part of the promise of Arrow is that applications > > are > > > > not a > > > > > > >>>> world > > > > > > >>>> > >> unto themselves, but interoperate with other > > Arrow-compliant > > > > > > >>>> systems. In > > > > > > >>>> > >> my case, I would like users to be able to examine > > > > RecordBatchs in > > > > > > >>>> tools > > > > > > >>>> > >> such as pyarrow without needing to be aware of any > > streaming > > > > > > >>>> app-specific > > > > > > >>>> > >> metadata. For example, a researcher may pull in an IPC > > > > "File" > > > > > > >>>> containing N > > > > > > >>>> > >> RecordBatch messages corresponding to those in Fig 4. I > > > > would > > > > > > >>>> very much > > > > > > >>>> > >> like for this casual user to not have to apply N slice > > > > operations > > > > > > >>>> based on > > > > > > >>>> > >> out-of-band data to get to the data that is relevant. > > > > > > >>>> > >> > > > > > > > >>>> > >> > > > > > > >>>> > >> Per above, should this become a standard enough use > > case, I > > > > think > > > > > > >>>> that > > > > > > >>>> > >> code can be developed in the Apache project to address > > it. > > > > > > >>>> > >> > > > > > > >>>> > >> > Devil's advocate: > > > > > > >>>> > >> > > > > > > > >>>> > >> > 1- Concurrent access to a mutable (growing) RecordBatch > > > > will > > > > > > >>>> require > > > > > > >>>> > >> synchronization of some sort to get consistent metadata > > > > reads. > > > > > > >>>> Since the > > > > > > >>>> > >> above proposals do not specify how this synchronization > > will > > > > > > occur > > > > > > >>>> for > > > > > > >>>> > >> tools such as pyarrow (we can imagine a Python user > > getting > > > > > > >>>> synchronized > > > > > > >>>> > >> access to File metadata and mapping a read-only area > > before > > > > the > > > > > > >>>> writer is > > > > > > >>>> > >> allowed to continue "appending" to this batch, or > > batches to > > > > this > > > > > > >>>> File), > > > > > > >>>> > >> some "unusual" code will be required anyway, so what is > > the > > > > harm > > > > > > of > > > > > > >>>> > >> consulting side-band data for slicing all the batches as > > > > part of > > > > > > >>>> this > > > > > > >>>> > >> "unusual" code? [Potential response: Yes, but it is > > still > > > > one > > > > > > >>>> less thing > > > > > > >>>> > >> to worry about, and perhaps first-class support for > > common > > > > > > >>>> synchronization > > > > > > >>>> > >> patterns can be forthcoming? These patterns may not > > require > > > > > > >>>> further format > > > > > > >>>> > >> changes?] > > > > > > >>>> > >> > > > > > > > >>>> > >> > My overall concern is that I see a lot of wasted effort > > > > dealing > > > > > > >>>> with > > > > > > >>>> > >> the "impedance mismatch" between batch oriented and > > streaming > > > > > > >>>> systems. I > > > > > > >>>> > >> believe that "best practices" will begin (and continue!) > > to > > > > > > prefer > > > > > > >>>> tools > > > > > > >>>> > >> that help bridge the gap. Certainly this is the case in > > my > > > > own > > > > > > >>>> work. I > > > > > > >>>> > >> agree with the appraisal at the end of the ZDNet article. > > > > If the > > > > > > >>>> above is > > > > > > >>>> > >> not a helpful solution, what other steps can be made? > > Or if > > > > > > Arrow > > > > > > >>>> is > > > > > > >>>> > >> intentionally confined to batch processing for the > > > > foreseeable > > > > > > >>>> future (in > > > > > > >>>> > >> terms of first-class support), I'm interested in the > > > > rationale. > > > > > > >>>> Perhaps > > > > > > >>>> > >> the feeling is that we avoid scope creep now (which I > > > > understand > > > > > > >>>> can be > > > > > > >>>> > >> never-ending) even if it means a certain breaking change > > in > > > > the > > > > > > >>>> future? > > > > > > >>>> > >> > > > > > > > >>>> > >> > > > > > > >>>> > >> There's some semantic issues with what "streaming" and > > > > "batch" > > > > > > >>>> means. > > > > > > >>>> > >> When people see "streaming" nowadays they think "Kafka" > > (or > > > > > > >>>> > >> Kafka-like). Single events flow in and out of streaming > > > > > > computation > > > > > > >>>> > >> nodes (e.g. like > > https://apache.github.io/incubator-heron/ > > > > or > > > > > > >>>> others). > > > > > > >>>> > >> The "streaming" is more about computational semantics > > than > > > > data > > > > > > >>>> > >> representation. > > > > > > >>>> > >> > > > > > > >>>> > >> The Arrow columnar format fundamentally deals with > > multiple > > > > > > >>>> records at > > > > > > >>>> > >> a time (you can have a record batch with size 1, but > > that is > > > > not > > > > > > >>>> going > > > > > > >>>> > >> to be efficient). But I do not think Arrow is > > "intentially > > > > > > >>>> confined" > > > > > > >>>> > >> to batch processing. If it makes sense to use a columnar > > > > format > > > > > > to > > > > > > >>>> > >> represent data in a streaming application, then you can > > > > certainly > > > > > > >>>> use > > > > > > >>>> > >> it for that. I'm aware of people successfully using Arrow > > > > with > > > > > > >>>> Kafka, > > > > > > >>>> > >> for example. > > > > > > >>>> > >> > > > > > > >>>> > >> - Wes > > > > > > >>>> > >> > > > > > > >>>> > >> > Who else encounters the need to mix/match batch and > > > > streaming, > > > > > > >>>> and what > > > > > > >>>> > >> are your experiences? > > > > > > >>>> > >> > > > > > > > >>>> > >> > Thanks for the further consideration and discussion! > > > > > > >>>> > >> > > > > > > > >>>> > >> > [1] https://zd.net/2H0LlBY > > > > > > >>>> > >> > [2] https://arrow.apache.org/docs/python/data.html > > > > > > >>>> > >> > [3] https://bit.ly/2J5sENZ > > > > > > >>>> > >> > [4] https://bit.ly/2Yske8L > > > > > > >>>> > >> > > > > > > >>>> > > > > > > > > >>>> > > > > > > >>> > > > > > > > > > > > > > >
