You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@beam.apache.org by Joey Tran <jo...@schrodinger.com> on 2025/01/31 00:18:51 UTC
FlumeJava - what happened to PObjects
I read the FlumeJava paper and I was just curious what happened to
PObjects. They seem like a useful construct. Do they exist in the java SDK
in some version still? Or were they done away with because they made
pipeline optimization more difficult?
Re: FlumeJava - what happened to PObjects
Posted by Robert Bradshaw via dev <de...@beam.apache.org>.
PObjects are just PCollections with a single element in them. They were
most useful for side inputs (and the BeamJava equivalent is the various
Views) but weren't really worth the extra baggage in the API otherwise, as
there's not much novel one can do with a PObject vs. a singleton
PCollection.
On Thu, Jan 30, 2025 at 4:20 PM Joey Tran <jo...@schrodinger.com> wrote:
> I read the FlumeJava paper and I was just curious what happened to
> PObjects. They seem like a useful construct. Do they exist in the java SDK
> in some version still? Or were they done away with because they made
> pipeline optimization more difficult?
>
Re: FlumeJava - what happened to PObjects
Posted by Kellen Dye via dev <de...@beam.apache.org>.
Yes, scio has a `materialize` which under the hood encodes each element
using its coder and serializes the sequence of byte arrays in a thin avro
wrapper. Some spotify-internal teams use this to pass data between a
sequence of jobs but it ends up just being a shortcut when you don't want
to permanently register the job results in some external system, so overall
we've seen this have limited utility and mostly recommend people use other
strategies.
Scio also has some pseudo-side-input implementations backed by spotify's
disk-backed-hashmap-implementation (sparkey) that write out a cache to GCS
(also by default using the element coders) that can be loaded on worker
nodes, but this doesn't return control to the main/driver class.
Kellen
On Mon, Feb 3, 2025 at 9:50 AM Kenneth Knowles <ke...@apache.org> wrote:
> We did actually get to the point with the SQL shell that you can issue
> windowed streaming queries and watch the results come in (on the
> DirectRunner with a hack probably similar to interactive runner). But it
> turns out that interactively waiting for the end of the window is a bit
> boring :-)
>
> Kenn
>
> On Fri, Jan 31, 2025 at 1:20 PM Robert Burke <ro...@frantil.com> wrote:
>
>> Really the main difference that makes it a little more complicated (but
>> not terribly so) is Flume isn't windowed (along with the metadata). Beam
>> would need to make that up front and available for use.
>>
>> On Fri, Jan 31, 2025, 9:50 AM Robert Bradshaw via dev <
>> dev@beam.apache.org> wrote:
>>
>>> On Fri, Jan 31, 2025 at 8:19 AM Joey Tran <jo...@schrodinger.com>
>>> wrote:
>>>
>>>> Is there an equivalent to `PCollectionView` in python?
>>>>
>>>> > ... as there's not much novel one can do with a PObject vs. a
>>>> singleton PCollection.
>>>>
>>>> Ah maybe I misunderstood how PObjects work. From the FlumeJava paper:
>>>>
>>>>
>>>>> These features can be used to express a computation that needs
>>>>> to iterate until the computed data converges:
>>>>>
>>>>> PCollection<Data> results = computeInitialApproximation();
>>>>> for (;;) {
>>>>> results = computeNextApproximation(results);
>>>>> PCollection<Boolean> haveConverged =
>>>>> results.parallelDo(checkIfConvergedFn(),
>>>>> collectionOf(booleans()));
>>>>> PObject<Boolean> allHaveConverged =
>>>>> haveConverged.combine(AND_BOOLS);
>>>>> FlumeJava.run();
>>>>> if (allHaveConverged.getValue()) break;
>>>>> }
>>>>> ... continue working with converged results ...
>>>>
>>>>
>>>> I had understood this to mean that the `PObject` will materialize the
>>>> pcollection into in-memory values, which is maybe a little novel? At least,
>>>> in the python SDK, I've always been writing elements to disk by transform
>>>> and then reading it manually outside of the pipeline back into the original
>>>> python objects.
>>>>
>>>
>>> FWIW, this wasn't unique to PObjects in FlumeJava, one could do the same
>>> for PCollections. While this is useful on the one hand, this notion of
>>> "materialization" (especially combined with further execution) has
>>> complexities in that keeping the main program alive now is essential to
>>> pipeline completion (when this feature is used) as opposed to the "fire and
>>> forget" model of Dataflow. (There were thoughts about doing runner-side
>>> lazy graph expansion, but they never got fleshed out.)
>>>
>>> Note that one can easily write a PTransform that internally executes a
>>> write and exposes an API to return the set of elements as in-memory objects
>>> (using the coder that was used for write) post pipeline completion. One
>>> would probably need to supply a distributed filestore to use, as there's
>>> not really a good "default." This doesn't have to be provided as part of
>>> Beam (though arguably it's a common enough usecase that maybe it should
>>> be). There's also some exploration in this area with Python's interactive
>>> utilities. One could conceivably even support this in streaming (though not
>>> without providing manual details of a backing store, e.g. a cofka instance
>>> or pubsub topic to use).
>>>
>>>
>>>> On Fri, Jan 31, 2025 at 9:50 AM Kenneth Knowles <ke...@apache.org>
>>>> wrote:
>>>>
>>>>> Fun fact, this was one of my onboarding projects :-)
>>>>>
>>>>> It is the way it is because the invariant we need in order to apply
>>>>> windowing independent of core business logic is: if you apply a transform
>>>>> to windowed input, each window should contain the same output it would if
>>>>> it were the entirety of the data. (composites are permitted to break
>>>>> this rule, but the core compute primitives never do)
>>>>>
>>>>> And I guess it seemed wrong to call something an "Object" when it was
>>>>> really an object per window. TBH the "P" was probably always a misnomer...
>>>>>
>>>>> Kenn
>>>>>
>>>>> On Thu, Jan 30, 2025 at 7:26 PM Robert Bradshaw via dev <
>>>>> dev@beam.apache.org> wrote:
>>>>>
>>>>>> On Thu, Jan 30, 2025 at 4:25 PM Reuven Lax via dev <
>>>>>> dev@beam.apache.org> wrote:
>>>>>>
>>>>>>> PCollectionView is the equivalent of PObject. Given that the Beam
>>>>>>> API needed to work with the windowing model, we needed something like a
>>>>>>> PObject that could be windowed. This is what PCollectionView provides.
>>>>>>>
>>>>>>
>>>>>> +1, I'd forgotten about the windowing complexities as well.
>>>>>>
>>>>>>
>>>>>>> On Thu, Jan 30, 2025 at 4:20 PM Joey Tran <jo...@schrodinger.com>
>>>>>>> wrote:
>>>>>>>
>>>>>>>> I read the FlumeJava paper and I was just curious what happened to
>>>>>>>> PObjects. They seem like a useful construct. Do they exist in the java SDK
>>>>>>>> in some version still? Or were they done away with because they made
>>>>>>>> pipeline optimization more difficult?
>>>>>>>>
>>>>>>>
Re: FlumeJava - what happened to PObjects
Posted by Kenneth Knowles <ke...@apache.org>.
We did actually get to the point with the SQL shell that you can issue
windowed streaming queries and watch the results come in (on the
DirectRunner with a hack probably similar to interactive runner). But it
turns out that interactively waiting for the end of the window is a bit
boring :-)
Kenn
On Fri, Jan 31, 2025 at 1:20 PM Robert Burke <ro...@frantil.com> wrote:
> Really the main difference that makes it a little more complicated (but
> not terribly so) is Flume isn't windowed (along with the metadata). Beam
> would need to make that up front and available for use.
>
> On Fri, Jan 31, 2025, 9:50 AM Robert Bradshaw via dev <de...@beam.apache.org>
> wrote:
>
>> On Fri, Jan 31, 2025 at 8:19 AM Joey Tran <jo...@schrodinger.com>
>> wrote:
>>
>>> Is there an equivalent to `PCollectionView` in python?
>>>
>>> > ... as there's not much novel one can do with a PObject vs. a
>>> singleton PCollection.
>>>
>>> Ah maybe I misunderstood how PObjects work. From the FlumeJava paper:
>>>
>>>
>>>> These features can be used to express a computation that needs
>>>> to iterate until the computed data converges:
>>>>
>>>> PCollection<Data> results = computeInitialApproximation();
>>>> for (;;) {
>>>> results = computeNextApproximation(results);
>>>> PCollection<Boolean> haveConverged =
>>>> results.parallelDo(checkIfConvergedFn(),
>>>> collectionOf(booleans()));
>>>> PObject<Boolean> allHaveConverged =
>>>> haveConverged.combine(AND_BOOLS);
>>>> FlumeJava.run();
>>>> if (allHaveConverged.getValue()) break;
>>>> }
>>>> ... continue working with converged results ...
>>>
>>>
>>> I had understood this to mean that the `PObject` will materialize the
>>> pcollection into in-memory values, which is maybe a little novel? At least,
>>> in the python SDK, I've always been writing elements to disk by transform
>>> and then reading it manually outside of the pipeline back into the original
>>> python objects.
>>>
>>
>> FWIW, this wasn't unique to PObjects in FlumeJava, one could do the same
>> for PCollections. While this is useful on the one hand, this notion of
>> "materialization" (especially combined with further execution) has
>> complexities in that keeping the main program alive now is essential to
>> pipeline completion (when this feature is used) as opposed to the "fire and
>> forget" model of Dataflow. (There were thoughts about doing runner-side
>> lazy graph expansion, but they never got fleshed out.)
>>
>> Note that one can easily write a PTransform that internally executes a
>> write and exposes an API to return the set of elements as in-memory objects
>> (using the coder that was used for write) post pipeline completion. One
>> would probably need to supply a distributed filestore to use, as there's
>> not really a good "default." This doesn't have to be provided as part of
>> Beam (though arguably it's a common enough usecase that maybe it should
>> be). There's also some exploration in this area with Python's interactive
>> utilities. One could conceivably even support this in streaming (though not
>> without providing manual details of a backing store, e.g. a cofka instance
>> or pubsub topic to use).
>>
>>
>>> On Fri, Jan 31, 2025 at 9:50 AM Kenneth Knowles <ke...@apache.org> wrote:
>>>
>>>> Fun fact, this was one of my onboarding projects :-)
>>>>
>>>> It is the way it is because the invariant we need in order to apply
>>>> windowing independent of core business logic is: if you apply a transform
>>>> to windowed input, each window should contain the same output it would if
>>>> it were the entirety of the data. (composites are permitted to break
>>>> this rule, but the core compute primitives never do)
>>>>
>>>> And I guess it seemed wrong to call something an "Object" when it was
>>>> really an object per window. TBH the "P" was probably always a misnomer...
>>>>
>>>> Kenn
>>>>
>>>> On Thu, Jan 30, 2025 at 7:26 PM Robert Bradshaw via dev <
>>>> dev@beam.apache.org> wrote:
>>>>
>>>>> On Thu, Jan 30, 2025 at 4:25 PM Reuven Lax via dev <
>>>>> dev@beam.apache.org> wrote:
>>>>>
>>>>>> PCollectionView is the equivalent of PObject. Given that the Beam API
>>>>>> needed to work with the windowing model, we needed something like a PObject
>>>>>> that could be windowed. This is what PCollectionView provides.
>>>>>>
>>>>>
>>>>> +1, I'd forgotten about the windowing complexities as well.
>>>>>
>>>>>
>>>>>> On Thu, Jan 30, 2025 at 4:20 PM Joey Tran <jo...@schrodinger.com>
>>>>>> wrote:
>>>>>>
>>>>>>> I read the FlumeJava paper and I was just curious what happened to
>>>>>>> PObjects. They seem like a useful construct. Do they exist in the java SDK
>>>>>>> in some version still? Or were they done away with because they made
>>>>>>> pipeline optimization more difficult?
>>>>>>>
>>>>>>
Re: FlumeJava - what happened to PObjects
Posted by Robert Burke <ro...@frantil.com>.
Really the main difference that makes it a little more complicated (but not
terribly so) is Flume isn't windowed (along with the metadata). Beam would
need to make that up front and available for use.
On Fri, Jan 31, 2025, 9:50 AM Robert Bradshaw via dev <de...@beam.apache.org>
wrote:
> On Fri, Jan 31, 2025 at 8:19 AM Joey Tran <jo...@schrodinger.com>
> wrote:
>
>> Is there an equivalent to `PCollectionView` in python?
>>
>> > ... as there's not much novel one can do with a PObject vs. a singleton
>> PCollection.
>>
>> Ah maybe I misunderstood how PObjects work. From the FlumeJava paper:
>>
>>
>>> These features can be used to express a computation that needs
>>> to iterate until the computed data converges:
>>>
>>> PCollection<Data> results = computeInitialApproximation();
>>> for (;;) {
>>> results = computeNextApproximation(results);
>>> PCollection<Boolean> haveConverged =
>>> results.parallelDo(checkIfConvergedFn(),
>>> collectionOf(booleans()));
>>> PObject<Boolean> allHaveConverged =
>>> haveConverged.combine(AND_BOOLS);
>>> FlumeJava.run();
>>> if (allHaveConverged.getValue()) break;
>>> }
>>> ... continue working with converged results ...
>>
>>
>> I had understood this to mean that the `PObject` will materialize the
>> pcollection into in-memory values, which is maybe a little novel? At least,
>> in the python SDK, I've always been writing elements to disk by transform
>> and then reading it manually outside of the pipeline back into the original
>> python objects.
>>
>
> FWIW, this wasn't unique to PObjects in FlumeJava, one could do the same
> for PCollections. While this is useful on the one hand, this notion of
> "materialization" (especially combined with further execution) has
> complexities in that keeping the main program alive now is essential to
> pipeline completion (when this feature is used) as opposed to the "fire and
> forget" model of Dataflow. (There were thoughts about doing runner-side
> lazy graph expansion, but they never got fleshed out.)
>
> Note that one can easily write a PTransform that internally executes a
> write and exposes an API to return the set of elements as in-memory objects
> (using the coder that was used for write) post pipeline completion. One
> would probably need to supply a distributed filestore to use, as there's
> not really a good "default." This doesn't have to be provided as part of
> Beam (though arguably it's a common enough usecase that maybe it should
> be). There's also some exploration in this area with Python's interactive
> utilities. One could conceivably even support this in streaming (though not
> without providing manual details of a backing store, e.g. a cofka instance
> or pubsub topic to use).
>
>
>> On Fri, Jan 31, 2025 at 9:50 AM Kenneth Knowles <ke...@apache.org> wrote:
>>
>>> Fun fact, this was one of my onboarding projects :-)
>>>
>>> It is the way it is because the invariant we need in order to apply
>>> windowing independent of core business logic is: if you apply a transform
>>> to windowed input, each window should contain the same output it would if
>>> it were the entirety of the data. (composites are permitted to break
>>> this rule, but the core compute primitives never do)
>>>
>>> And I guess it seemed wrong to call something an "Object" when it was
>>> really an object per window. TBH the "P" was probably always a misnomer...
>>>
>>> Kenn
>>>
>>> On Thu, Jan 30, 2025 at 7:26 PM Robert Bradshaw via dev <
>>> dev@beam.apache.org> wrote:
>>>
>>>> On Thu, Jan 30, 2025 at 4:25 PM Reuven Lax via dev <de...@beam.apache.org>
>>>> wrote:
>>>>
>>>>> PCollectionView is the equivalent of PObject. Given that the Beam API
>>>>> needed to work with the windowing model, we needed something like a PObject
>>>>> that could be windowed. This is what PCollectionView provides.
>>>>>
>>>>
>>>> +1, I'd forgotten about the windowing complexities as well.
>>>>
>>>>
>>>>> On Thu, Jan 30, 2025 at 4:20 PM Joey Tran <jo...@schrodinger.com>
>>>>> wrote:
>>>>>
>>>>>> I read the FlumeJava paper and I was just curious what happened to
>>>>>> PObjects. They seem like a useful construct. Do they exist in the java SDK
>>>>>> in some version still? Or were they done away with because they made
>>>>>> pipeline optimization more difficult?
>>>>>>
>>>>>
Re: FlumeJava - what happened to PObjects
Posted by Robert Bradshaw via dev <de...@beam.apache.org>.
On Fri, Jan 31, 2025 at 8:19 AM Joey Tran <jo...@schrodinger.com> wrote:
> Is there an equivalent to `PCollectionView` in python?
>
> > ... as there's not much novel one can do with a PObject vs. a singleton
> PCollection.
>
> Ah maybe I misunderstood how PObjects work. From the FlumeJava paper:
>
>
>> These features can be used to express a computation that needs
>> to iterate until the computed data converges:
>>
>> PCollection<Data> results = computeInitialApproximation();
>> for (;;) {
>> results = computeNextApproximation(results);
>> PCollection<Boolean> haveConverged =
>> results.parallelDo(checkIfConvergedFn(),
>> collectionOf(booleans()));
>> PObject<Boolean> allHaveConverged =
>> haveConverged.combine(AND_BOOLS);
>> FlumeJava.run();
>> if (allHaveConverged.getValue()) break;
>> }
>> ... continue working with converged results ...
>
>
> I had understood this to mean that the `PObject` will materialize the
> pcollection into in-memory values, which is maybe a little novel? At least,
> in the python SDK, I've always been writing elements to disk by transform
> and then reading it manually outside of the pipeline back into the original
> python objects.
>
FWIW, this wasn't unique to PObjects in FlumeJava, one could do the same
for PCollections. While this is useful on the one hand, this notion of
"materialization" (especially combined with further execution) has
complexities in that keeping the main program alive now is essential to
pipeline completion (when this feature is used) as opposed to the "fire and
forget" model of Dataflow. (There were thoughts about doing runner-side
lazy graph expansion, but they never got fleshed out.)
Note that one can easily write a PTransform that internally executes a
write and exposes an API to return the set of elements as in-memory objects
(using the coder that was used for write) post pipeline completion. One
would probably need to supply a distributed filestore to use, as there's
not really a good "default." This doesn't have to be provided as part of
Beam (though arguably it's a common enough usecase that maybe it should
be). There's also some exploration in this area with Python's interactive
utilities. One could conceivably even support this in streaming (though not
without providing manual details of a backing store, e.g. a cofka instance
or pubsub topic to use).
> On Fri, Jan 31, 2025 at 9:50 AM Kenneth Knowles <ke...@apache.org> wrote:
>
>> Fun fact, this was one of my onboarding projects :-)
>>
>> It is the way it is because the invariant we need in order to apply
>> windowing independent of core business logic is: if you apply a transform
>> to windowed input, each window should contain the same output it would if
>> it were the entirety of the data. (composites are permitted to break
>> this rule, but the core compute primitives never do)
>>
>> And I guess it seemed wrong to call something an "Object" when it was
>> really an object per window. TBH the "P" was probably always a misnomer...
>>
>> Kenn
>>
>> On Thu, Jan 30, 2025 at 7:26 PM Robert Bradshaw via dev <
>> dev@beam.apache.org> wrote:
>>
>>> On Thu, Jan 30, 2025 at 4:25 PM Reuven Lax via dev <de...@beam.apache.org>
>>> wrote:
>>>
>>>> PCollectionView is the equivalent of PObject. Given that the Beam API
>>>> needed to work with the windowing model, we needed something like a PObject
>>>> that could be windowed. This is what PCollectionView provides.
>>>>
>>>
>>> +1, I'd forgotten about the windowing complexities as well.
>>>
>>>
>>>> On Thu, Jan 30, 2025 at 4:20 PM Joey Tran <jo...@schrodinger.com>
>>>> wrote:
>>>>
>>>>> I read the FlumeJava paper and I was just curious what happened to
>>>>> PObjects. They seem like a useful construct. Do they exist in the java SDK
>>>>> in some version still? Or were they done away with because they made
>>>>> pipeline optimization more difficult?
>>>>>
>>>>
Re: FlumeJava - what happened to PObjects
Posted by Kenneth Knowles <ke...@apache.org>.
On Fri, Jan 31, 2025 at 12:06 PM Joey Tran <jo...@schrodinger.com>
wrote:
> Yeah I briefly considered reimplementing PObject in python to make unit
> testing asserts a little more ergonomic (`results_pobject.getvalue() ==
> set([1, 2, 3])` feels more natural than `assert_that(results, equal_to([1,
> 2, 3]))`) but when I proposed it to some beam users at my company someone
> excitedly asked if they could start adding conditional forks in their
> pipeline and that made me a lot more hesitant.
>
> I suppose it wouldn't be that hard to have our internal distributed runner
> just check to see if there are any PObject transforms and reject a pipeline
> if it does.
>
> Anyways, I asked here because I was wondering if the same line of thought
> might've lead to the removal of PObject (just out of curiosity)
>
Well, I'd recommend this takeaway from the answers here: having an API so
that the main() driver program can read in the contents of a PCollection is
very doable and largely orthogonal to the rename/revamp of PObject to
PCollectionView.
One big win with newer ideas about portable side inputs is that they can
have more than one API. The API for ye olde PObject, or equivalently a
PCollectionView in the global window, is to just read the whole contents of
the PCollection. So this only really works for very small data, while we've
now got implementations for Map and Set views with more limited APIs that
work for big data. If you built an API on top of Beam that really focused
on having persistent queryable intermediate materializations, you might
want to use those. But more likely you would just limit it to small data to
control a loop, say. Or just materialize it to an appropriate storage
system that query natively like you already do.
Kenn
>
> On Fri, Jan 31, 2025 at 11:57 AM Kenneth Knowles <ke...@apache.org> wrote:
>
>>
>>
>> On Fri, Jan 31, 2025 at 11:20 AM Joey Tran <jo...@schrodinger.com>
>> wrote:
>>
>>> Is there an equivalent to `PCollectionView` in python?
>>>
>>> > ... as there's not much novel one can do with a PObject vs. a
>>> singleton PCollection.
>>>
>>> Ah maybe I misunderstood how PObjects work. From the FlumeJava paper:
>>>
>>>
>>>> These features can be used to express a computation that needs
>>>> to iterate until the computed data converges:
>>>>
>>>> PCollection<Data> results = computeInitialApproximation();
>>>> for (;;) {
>>>> results = computeNextApproximation(results);
>>>> PCollection<Boolean> haveConverged =
>>>> results.parallelDo(checkIfConvergedFn(),
>>>> collectionOf(booleans()));
>>>> PObject<Boolean> allHaveConverged =
>>>> haveConverged.combine(AND_BOOLS);
>>>> FlumeJava.run();
>>>> if (allHaveConverged.getValue()) break;
>>>> }
>>>> ... continue working with converged results ...
>>>
>>>
>>> I had understood this to mean that the `PObject` will materialize the
>>> pcollection into in-memory values, which is maybe a little novel?
>>>
>>
>>>
>> At least, in the python SDK, I've always been writing elements to disk by
>>> transform and then reading it manually outside of the pipeline back into
>>> the original python objects.
>>>
>>
>> Ah, yes, the magic is more in the `FlumeJava.run()` which is equivalent
>> to doing a whole pipeline.run() and the PObject written to a place where it
>> can be then read by the main driver program. Our interactive runner and SQL
>> shell do this for in-memory pipelines, but we don't have anything like that
>> for our distributed runners. If I understand correctly, scio does something
>> along these lines. I think having some sort of "standard for storage
>> between driver main program and distributed pipeline execution" is
>> compatible with Beam, in theory. I also suspect it is very easy to
>> accidentally write a program that is weirdly slow in confusing ways,
>> because of the blurring of the different execution environments/phases.
>>
>> Kenn
>>
>>
>>> On Fri, Jan 31, 2025 at 9:50 AM Kenneth Knowles <ke...@apache.org> wrote:
>>>
>>>> Fun fact, this was one of my onboarding projects :-)
>>>>
>>>> It is the way it is because the invariant we need in order to apply
>>>> windowing independent of core business logic is: if you apply a transform
>>>> to windowed input, each window should contain the same output it would if
>>>> it were the entirety of the data. (composites are permitted to break
>>>> this rule, but the core compute primitives never do)
>>>>
>>>> And I guess it seemed wrong to call something an "Object" when it was
>>>> really an object per window. TBH the "P" was probably always a misnomer...
>>>>
>>>> Kenn
>>>>
>>>> On Thu, Jan 30, 2025 at 7:26 PM Robert Bradshaw via dev <
>>>> dev@beam.apache.org> wrote:
>>>>
>>>>> On Thu, Jan 30, 2025 at 4:25 PM Reuven Lax via dev <
>>>>> dev@beam.apache.org> wrote:
>>>>>
>>>>>> PCollectionView is the equivalent of PObject. Given that the Beam API
>>>>>> needed to work with the windowing model, we needed something like a PObject
>>>>>> that could be windowed. This is what PCollectionView provides.
>>>>>>
>>>>>
>>>>> +1, I'd forgotten about the windowing complexities as well.
>>>>>
>>>>>
>>>>>> On Thu, Jan 30, 2025 at 4:20 PM Joey Tran <jo...@schrodinger.com>
>>>>>> wrote:
>>>>>>
>>>>>>> I read the FlumeJava paper and I was just curious what happened to
>>>>>>> PObjects. They seem like a useful construct. Do they exist in the java SDK
>>>>>>> in some version still? Or were they done away with because they made
>>>>>>> pipeline optimization more difficult?
>>>>>>>
>>>>>>
Re: FlumeJava - what happened to PObjects
Posted by Robert Burke <ro...@frantil.com>.
The main thing stopping a low impedance version of collection
materialization is the lack of a robust notion of common storage across the
submission program and different runners.
We almost have this with runner specific staging directories, but we don't
make them a canonical facet of Beam execution, or part of the existing
contracts.
Ultimately the moment you get to wantings something in memory, you'd just
need to write it out somewhere known and read it back in. Wether that's
expanding the Job Management with a way to request such objects, or hiding
the read & write mechanisms, and tagging a collection on job submission
that we'd like to get the data after the job completes.
On Fri, Jan 31, 2025, 9:06 AM Joey Tran <jo...@schrodinger.com> wrote:
> Yeah I briefly considered reimplementing PObject in python to make unit
> testing asserts a little more ergonomic (`results_pobject.getvalue() ==
> set([1, 2, 3])` feels more natural than `assert_that(results, equal_to([1,
> 2, 3]))`) but when I proposed it to some beam users at my company someone
> excitedly asked if they could start adding conditional forks in their
> pipeline and that made me a lot more hesitant.
>
> I suppose it wouldn't be that hard to have our internal distributed runner
> just check to see if there are any PObject transforms and reject a pipeline
> if it does.
>
> Anyways, I asked here because I was wondering if the same line of thought
> might've lead to the removal of PObject (just out of curiosity)
>
> On Fri, Jan 31, 2025 at 11:57 AM Kenneth Knowles <ke...@apache.org> wrote:
>
>>
>>
>> On Fri, Jan 31, 2025 at 11:20 AM Joey Tran <jo...@schrodinger.com>
>> wrote:
>>
>>> Is there an equivalent to `PCollectionView` in python?
>>>
>>> > ... as there's not much novel one can do with a PObject vs. a
>>> singleton PCollection.
>>>
>>> Ah maybe I misunderstood how PObjects work. From the FlumeJava paper:
>>>
>>>
>>>> These features can be used to express a computation that needs
>>>> to iterate until the computed data converges:
>>>>
>>>> PCollection<Data> results = computeInitialApproximation();
>>>> for (;;) {
>>>> results = computeNextApproximation(results);
>>>> PCollection<Boolean> haveConverged =
>>>> results.parallelDo(checkIfConvergedFn(),
>>>> collectionOf(booleans()));
>>>> PObject<Boolean> allHaveConverged =
>>>> haveConverged.combine(AND_BOOLS);
>>>> FlumeJava.run();
>>>> if (allHaveConverged.getValue()) break;
>>>> }
>>>> ... continue working with converged results ...
>>>
>>>
>>> I had understood this to mean that the `PObject` will materialize the
>>> pcollection into in-memory values, which is maybe a little novel?
>>>
>>
>>>
>> At least, in the python SDK, I've always been writing elements to disk by
>>> transform and then reading it manually outside of the pipeline back into
>>> the original python objects.
>>>
>>
>> Ah, yes, the magic is more in the `FlumeJava.run()` which is equivalent
>> to doing a whole pipeline.run() and the PObject written to a place where it
>> can be then read by the main driver program. Our interactive runner and SQL
>> shell do this for in-memory pipelines, but we don't have anything like that
>> for our distributed runners. If I understand correctly, scio does something
>> along these lines. I think having some sort of "standard for storage
>> between driver main program and distributed pipeline execution" is
>> compatible with Beam, in theory. I also suspect it is very easy to
>> accidentally write a program that is weirdly slow in confusing ways,
>> because of the blurring of the different execution environments/phases.
>>
>> Kenn
>>
>>
>>> On Fri, Jan 31, 2025 at 9:50 AM Kenneth Knowles <ke...@apache.org> wrote:
>>>
>>>> Fun fact, this was one of my onboarding projects :-)
>>>>
>>>> It is the way it is because the invariant we need in order to apply
>>>> windowing independent of core business logic is: if you apply a transform
>>>> to windowed input, each window should contain the same output it would if
>>>> it were the entirety of the data. (composites are permitted to break
>>>> this rule, but the core compute primitives never do)
>>>>
>>>> And I guess it seemed wrong to call something an "Object" when it was
>>>> really an object per window. TBH the "P" was probably always a misnomer...
>>>>
>>>> Kenn
>>>>
>>>> On Thu, Jan 30, 2025 at 7:26 PM Robert Bradshaw via dev <
>>>> dev@beam.apache.org> wrote:
>>>>
>>>>> On Thu, Jan 30, 2025 at 4:25 PM Reuven Lax via dev <
>>>>> dev@beam.apache.org> wrote:
>>>>>
>>>>>> PCollectionView is the equivalent of PObject. Given that the Beam API
>>>>>> needed to work with the windowing model, we needed something like a PObject
>>>>>> that could be windowed. This is what PCollectionView provides.
>>>>>>
>>>>>
>>>>> +1, I'd forgotten about the windowing complexities as well.
>>>>>
>>>>>
>>>>>> On Thu, Jan 30, 2025 at 4:20 PM Joey Tran <jo...@schrodinger.com>
>>>>>> wrote:
>>>>>>
>>>>>>> I read the FlumeJava paper and I was just curious what happened to
>>>>>>> PObjects. They seem like a useful construct. Do they exist in the java SDK
>>>>>>> in some version still? Or were they done away with because they made
>>>>>>> pipeline optimization more difficult?
>>>>>>>
>>>>>>
Re: FlumeJava - what happened to PObjects
Posted by Joey Tran <jo...@schrodinger.com>.
Yeah I briefly considered reimplementing PObject in python to make unit
testing asserts a little more ergonomic (`results_pobject.getvalue() ==
set([1, 2, 3])` feels more natural than `assert_that(results, equal_to([1,
2, 3]))`) but when I proposed it to some beam users at my company someone
excitedly asked if they could start adding conditional forks in their
pipeline and that made me a lot more hesitant.
I suppose it wouldn't be that hard to have our internal distributed runner
just check to see if there are any PObject transforms and reject a pipeline
if it does.
Anyways, I asked here because I was wondering if the same line of thought
might've lead to the removal of PObject (just out of curiosity)
On Fri, Jan 31, 2025 at 11:57 AM Kenneth Knowles <ke...@apache.org> wrote:
>
>
> On Fri, Jan 31, 2025 at 11:20 AM Joey Tran <jo...@schrodinger.com>
> wrote:
>
>> Is there an equivalent to `PCollectionView` in python?
>>
>> > ... as there's not much novel one can do with a PObject vs. a singleton
>> PCollection.
>>
>> Ah maybe I misunderstood how PObjects work. From the FlumeJava paper:
>>
>>
>>> These features can be used to express a computation that needs
>>> to iterate until the computed data converges:
>>>
>>> PCollection<Data> results = computeInitialApproximation();
>>> for (;;) {
>>> results = computeNextApproximation(results);
>>> PCollection<Boolean> haveConverged =
>>> results.parallelDo(checkIfConvergedFn(),
>>> collectionOf(booleans()));
>>> PObject<Boolean> allHaveConverged =
>>> haveConverged.combine(AND_BOOLS);
>>> FlumeJava.run();
>>> if (allHaveConverged.getValue()) break;
>>> }
>>> ... continue working with converged results ...
>>
>>
>> I had understood this to mean that the `PObject` will materialize the
>> pcollection into in-memory values, which is maybe a little novel?
>>
>
>>
> At least, in the python SDK, I've always been writing elements to disk by
>> transform and then reading it manually outside of the pipeline back into
>> the original python objects.
>>
>
> Ah, yes, the magic is more in the `FlumeJava.run()` which is equivalent to
> doing a whole pipeline.run() and the PObject written to a place where it
> can be then read by the main driver program. Our interactive runner and SQL
> shell do this for in-memory pipelines, but we don't have anything like that
> for our distributed runners. If I understand correctly, scio does something
> along these lines. I think having some sort of "standard for storage
> between driver main program and distributed pipeline execution" is
> compatible with Beam, in theory. I also suspect it is very easy to
> accidentally write a program that is weirdly slow in confusing ways,
> because of the blurring of the different execution environments/phases.
>
> Kenn
>
>
>> On Fri, Jan 31, 2025 at 9:50 AM Kenneth Knowles <ke...@apache.org> wrote:
>>
>>> Fun fact, this was one of my onboarding projects :-)
>>>
>>> It is the way it is because the invariant we need in order to apply
>>> windowing independent of core business logic is: if you apply a transform
>>> to windowed input, each window should contain the same output it would if
>>> it were the entirety of the data. (composites are permitted to break
>>> this rule, but the core compute primitives never do)
>>>
>>> And I guess it seemed wrong to call something an "Object" when it was
>>> really an object per window. TBH the "P" was probably always a misnomer...
>>>
>>> Kenn
>>>
>>> On Thu, Jan 30, 2025 at 7:26 PM Robert Bradshaw via dev <
>>> dev@beam.apache.org> wrote:
>>>
>>>> On Thu, Jan 30, 2025 at 4:25 PM Reuven Lax via dev <de...@beam.apache.org>
>>>> wrote:
>>>>
>>>>> PCollectionView is the equivalent of PObject. Given that the Beam API
>>>>> needed to work with the windowing model, we needed something like a PObject
>>>>> that could be windowed. This is what PCollectionView provides.
>>>>>
>>>>
>>>> +1, I'd forgotten about the windowing complexities as well.
>>>>
>>>>
>>>>> On Thu, Jan 30, 2025 at 4:20 PM Joey Tran <jo...@schrodinger.com>
>>>>> wrote:
>>>>>
>>>>>> I read the FlumeJava paper and I was just curious what happened to
>>>>>> PObjects. They seem like a useful construct. Do they exist in the java SDK
>>>>>> in some version still? Or were they done away with because they made
>>>>>> pipeline optimization more difficult?
>>>>>>
>>>>>
Re: FlumeJava - what happened to PObjects
Posted by Kenneth Knowles <ke...@apache.org>.
On Fri, Jan 31, 2025 at 11:20 AM Joey Tran <jo...@schrodinger.com>
wrote:
> Is there an equivalent to `PCollectionView` in python?
>
> > ... as there's not much novel one can do with a PObject vs. a singleton
> PCollection.
>
> Ah maybe I misunderstood how PObjects work. From the FlumeJava paper:
>
>
>> These features can be used to express a computation that needs
>> to iterate until the computed data converges:
>>
>> PCollection<Data> results = computeInitialApproximation();
>> for (;;) {
>> results = computeNextApproximation(results);
>> PCollection<Boolean> haveConverged =
>> results.parallelDo(checkIfConvergedFn(),
>> collectionOf(booleans()));
>> PObject<Boolean> allHaveConverged =
>> haveConverged.combine(AND_BOOLS);
>> FlumeJava.run();
>> if (allHaveConverged.getValue()) break;
>> }
>> ... continue working with converged results ...
>
>
> I had understood this to mean that the `PObject` will materialize the
> pcollection into in-memory values, which is maybe a little novel?
>
>
At least, in the python SDK, I've always been writing elements to disk by
> transform and then reading it manually outside of the pipeline back into
> the original python objects.
>
Ah, yes, the magic is more in the `FlumeJava.run()` which is equivalent to
doing a whole pipeline.run() and the PObject written to a place where it
can be then read by the main driver program. Our interactive runner and SQL
shell do this for in-memory pipelines, but we don't have anything like that
for our distributed runners. If I understand correctly, scio does something
along these lines. I think having some sort of "standard for storage
between driver main program and distributed pipeline execution" is
compatible with Beam, in theory. I also suspect it is very easy to
accidentally write a program that is weirdly slow in confusing ways,
because of the blurring of the different execution environments/phases.
Kenn
> On Fri, Jan 31, 2025 at 9:50 AM Kenneth Knowles <ke...@apache.org> wrote:
>
>> Fun fact, this was one of my onboarding projects :-)
>>
>> It is the way it is because the invariant we need in order to apply
>> windowing independent of core business logic is: if you apply a transform
>> to windowed input, each window should contain the same output it would if
>> it were the entirety of the data. (composites are permitted to break
>> this rule, but the core compute primitives never do)
>>
>> And I guess it seemed wrong to call something an "Object" when it was
>> really an object per window. TBH the "P" was probably always a misnomer...
>>
>> Kenn
>>
>> On Thu, Jan 30, 2025 at 7:26 PM Robert Bradshaw via dev <
>> dev@beam.apache.org> wrote:
>>
>>> On Thu, Jan 30, 2025 at 4:25 PM Reuven Lax via dev <de...@beam.apache.org>
>>> wrote:
>>>
>>>> PCollectionView is the equivalent of PObject. Given that the Beam API
>>>> needed to work with the windowing model, we needed something like a PObject
>>>> that could be windowed. This is what PCollectionView provides.
>>>>
>>>
>>> +1, I'd forgotten about the windowing complexities as well.
>>>
>>>
>>>> On Thu, Jan 30, 2025 at 4:20 PM Joey Tran <jo...@schrodinger.com>
>>>> wrote:
>>>>
>>>>> I read the FlumeJava paper and I was just curious what happened to
>>>>> PObjects. They seem like a useful construct. Do they exist in the java SDK
>>>>> in some version still? Or were they done away with because they made
>>>>> pipeline optimization more difficult?
>>>>>
>>>>
Re: FlumeJava - what happened to PObjects
Posted by Joey Tran <jo...@schrodinger.com>.
Is there an equivalent to `PCollectionView` in python?
> ... as there's not much novel one can do with a PObject vs. a singleton
PCollection.
Ah maybe I misunderstood how PObjects work. From the FlumeJava paper:
> These features can be used to express a computation that needs
> to iterate until the computed data converges:
>
> PCollection<Data> results = computeInitialApproximation();
> for (;;) {
> results = computeNextApproximation(results);
> PCollection<Boolean> haveConverged =
> results.parallelDo(checkIfConvergedFn(),
> collectionOf(booleans()));
> PObject<Boolean> allHaveConverged =
> haveConverged.combine(AND_BOOLS);
> FlumeJava.run();
> if (allHaveConverged.getValue()) break;
> }
> ... continue working with converged results ...
I had understood this to mean that the `PObject` will materialize the
pcollection into in-memory values, which is maybe a little novel? At least,
in the python SDK, I've always been writing elements to disk by transform
and then reading it manually outside of the pipeline back into the original
python objects.
On Fri, Jan 31, 2025 at 9:50 AM Kenneth Knowles <ke...@apache.org> wrote:
> Fun fact, this was one of my onboarding projects :-)
>
> It is the way it is because the invariant we need in order to apply
> windowing independent of core business logic is: if you apply a transform
> to windowed input, each window should contain the same output it would if
> it were the entirety of the data. (composites are permitted to break this
> rule, but the core compute primitives never do)
>
> And I guess it seemed wrong to call something an "Object" when it was
> really an object per window. TBH the "P" was probably always a misnomer...
>
> Kenn
>
> On Thu, Jan 30, 2025 at 7:26 PM Robert Bradshaw via dev <
> dev@beam.apache.org> wrote:
>
>> On Thu, Jan 30, 2025 at 4:25 PM Reuven Lax via dev <de...@beam.apache.org>
>> wrote:
>>
>>> PCollectionView is the equivalent of PObject. Given that the Beam API
>>> needed to work with the windowing model, we needed something like a PObject
>>> that could be windowed. This is what PCollectionView provides.
>>>
>>
>> +1, I'd forgotten about the windowing complexities as well.
>>
>>
>>> On Thu, Jan 30, 2025 at 4:20 PM Joey Tran <jo...@schrodinger.com>
>>> wrote:
>>>
>>>> I read the FlumeJava paper and I was just curious what happened to
>>>> PObjects. They seem like a useful construct. Do they exist in the java SDK
>>>> in some version still? Or were they done away with because they made
>>>> pipeline optimization more difficult?
>>>>
>>>
Re: FlumeJava - what happened to PObjects
Posted by Kenneth Knowles <ke...@apache.org>.
Fun fact, this was one of my onboarding projects :-)
It is the way it is because the invariant we need in order to apply
windowing independent of core business logic is: if you apply a transform
to windowed input, each window should contain the same output it would if
it were the entirety of the data. (composites are permitted to break this
rule, but the core compute primitives never do)
And I guess it seemed wrong to call something an "Object" when it was
really an object per window. TBH the "P" was probably always a misnomer...
Kenn
On Thu, Jan 30, 2025 at 7:26 PM Robert Bradshaw via dev <de...@beam.apache.org>
wrote:
> On Thu, Jan 30, 2025 at 4:25 PM Reuven Lax via dev <de...@beam.apache.org>
> wrote:
>
>> PCollectionView is the equivalent of PObject. Given that the Beam API
>> needed to work with the windowing model, we needed something like a PObject
>> that could be windowed. This is what PCollectionView provides.
>>
>
> +1, I'd forgotten about the windowing complexities as well.
>
>
>> On Thu, Jan 30, 2025 at 4:20 PM Joey Tran <jo...@schrodinger.com>
>> wrote:
>>
>>> I read the FlumeJava paper and I was just curious what happened to
>>> PObjects. They seem like a useful construct. Do they exist in the java SDK
>>> in some version still? Or were they done away with because they made
>>> pipeline optimization more difficult?
>>>
>>
Re: FlumeJava - what happened to PObjects
Posted by Robert Bradshaw via dev <de...@beam.apache.org>.
On Thu, Jan 30, 2025 at 4:25 PM Reuven Lax via dev <de...@beam.apache.org>
wrote:
> PCollectionView is the equivalent of PObject. Given that the Beam API
> needed to work with the windowing model, we needed something like a PObject
> that could be windowed. This is what PCollectionView provides.
>
+1, I'd forgotten about the windowing complexities as well.
> On Thu, Jan 30, 2025 at 4:20 PM Joey Tran <jo...@schrodinger.com>
> wrote:
>
>> I read the FlumeJava paper and I was just curious what happened to
>> PObjects. They seem like a useful construct. Do they exist in the java SDK
>> in some version still? Or were they done away with because they made
>> pipeline optimization more difficult?
>>
>
Re: FlumeJava - what happened to PObjects
Posted by Reuven Lax via dev <de...@beam.apache.org>.
PCollectionView is the equivalent of PObject. Given that the Beam API
needed to work with the windowing model, we needed something like a PObject
that could be windowed. This is what PCollectionView provides.
On Thu, Jan 30, 2025 at 4:20 PM Joey Tran <jo...@schrodinger.com> wrote:
> I read the FlumeJava paper and I was just curious what happened to
> PObjects. They seem like a useful construct. Do they exist in the java SDK
> in some version still? Or were they done away with because they made
> pipeline optimization more difficult?
>