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 > > > > >>>> > >> > > > > >>>> > > > > > > >>>> > > > > >>> > > > > > >
