You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@beam.apache.org by Yushu Yao <ya...@gmail.com> on 2022/05/23 16:38:08 UTC
RDD (Spark dataframe) into a PCollection?
Hi Folks,
I know this is not the optimal way to use beam :-) But assume I only use
the spark runner.
I have a spark library (very complex) that emits a spark dataframe (or
RDD).
I also have an existing complex beam pipeline that can do post processing
on the data inside the dataframe.
However, the beam part needs a pcollection to start with. The question is,
how can I convert a spark RDD into a pcollection?
Thanks
-Yushu
Re: RDD (Spark dataframe) into a PCollection?
Posted by Yushu Yao <ya...@gmail.com>.
Thanks Robert and Brian.
As for "writing the RDD somewhere", I can totally write a bunch of files on
disk/s3. Any other options?
-Yushu
On Mon, May 23, 2022 at 11:40 AM Brian Hulette <bh...@google.com> wrote:
> Yeah I'm not sure of any simple way to do this. I wonder if it's worth
> considering building some Spark runner-specific feature around this, or at
> least packaging up Robert's proposed solution?
>
> There could be other interesting integrations in this space too, e.g.
> using Spark RDDs as a cache for Interactive Beam.
>
> Brian
>
> On Mon, May 23, 2022 at 11:35 AM Robert Bradshaw <ro...@google.com>
> wrote:
>
>> The easiest way to do this would be to write the RDD somewhere then
>> read it from Beam.
>>
>> On Mon, May 23, 2022 at 9:39 AM Yushu Yao <ya...@gmail.com> wrote:
>> >
>> > Hi Folks,
>> >
>> > I know this is not the optimal way to use beam :-) But assume I only
>> use the spark runner.
>> >
>> > I have a spark library (very complex) that emits a spark dataframe (or
>> RDD).
>> > I also have an existing complex beam pipeline that can do post
>> processing on the data inside the dataframe.
>> >
>> > However, the beam part needs a pcollection to start with. The question
>> is, how can I convert a spark RDD into a pcollection?
>> >
>> > Thanks
>> > -Yushu
>> >
>>
>
Re: RDD (Spark dataframe) into a PCollection?
Posted by Moritz Mack <mm...@talend.com>.
Hi Yushu,
Have a look at org.apache.beam.runners.spark.translation.EvaluationContext in the Spark runner. It maintains that mapping between PCollections and RDDs (wrapped in the BoundedDataset helper). As Reuven just pointed out, values are timestamped (and windowed) in Beam, therefore BoundedDataset expects a JavaRDD<WindowedValue<T>>.
The idea is to map your external RDD to a new PCollection (PCollection.createPrimitiveOutputInternal) in the EvaluationContext (and vice versa). You can then apply Beam transforms to that PCollection (and with that effectively to the mapped RDD) as you are used to. Obviously, there’s a few steps necessary as EvaluationContext isn’t easily accessible from the outside.
Just having a quick look, creating a RddSource for Spark RDDs seems also not too bad. That would allow you to do something like this:
pipeline.apply(Read.from(new RddSource<>(javaRdd.rdd(), coder)));
Though I haven’t done much testing beyond a quick experiment. One notable disadvantage of that approach Is that all RDD partition data must be broadcasted to all workers to then pick the right partition. This should mostly be fine, but some types of partitions carry data as well …
https://gist.github.com/mosche/5c1ef8ba281a9a08df1ec67fac700d03
/Moritz
On 24.05.22, 16:46, "Yushu Yao" <ya...@gmail.com> wrote:
Looks like it's a valid use case. Wondering anyone can give some high level guidelines on how to implement this? I can give it a try. -Yushu On Tue, May 24, 2022 at 2:42 AM Jan Lukavský <je...@seznam.cz> wrote:
ZjQcmQRYFpfptBannerStart
This Message Is From an External Sender
This message came from outside your organization.
Exercise caution when opening attachments or clicking any links.
ZjQcmQRYFpfptBannerEnd
Looks like it's a valid use case.
Wondering anyone can give some high level guidelines on how to implement this?
I can give it a try.
-Yushu
On Tue, May 24, 2022 at 2:42 AM Jan Lukavský <je...@seznam.cz>> wrote:
+dev@beam<ma...@beam.apache.org>
On 5/24/22 11:40, Jan Lukavský wrote:
Hi,
I think this feature is valid. Every runner for which Beam is not a 'native' SDK uses some form of translation context, which maps PCollection to internal representation of the particular SDK of the runner (RDD in this case). It should be possible to "import" an RDD into the specific runner via something like
SparkRunner runner = ....;
PCollection<...> pCollection = runner.importRDD(rdd);
and similarly
RDD<...> rdd = runner.exportRDD(pCollection);
Yes, apparently this would be runner specific, but that is the point, actually. This would enable using features and libraries, that Beam does not have, or micro-optimize some particular step using runner-specific features, that we don't have in Beam. We actually had this feature (at least in a prototype) many years ago when Euphoria was a separate project.
Jan
On 5/23/22 20:58, Alexey Romanenko wrote:
On 23 May 2022, at 20:40, Brian Hulette <bh...@google.com>> wrote:
Yeah I'm not sure of any simple way to do this. I wonder if it's worth considering building some Spark runner-specific feature around this, or at least packaging up Robert's proposed solution?
I’m not sure that a runner specific feature is a good way to do this since the other runners won’t be able to support it or I’m missing something?
There could be other interesting integrations in this space too, e.g. using Spark RDDs as a cache for Interactive Beam.
Another option could be to add something like SparkIO (or FlinkIO/whatever) to read/write data from/to Spark data structures for such cases (Spark schema to Beam schema convention also could be supported). And dreaming a bit more, for those who need to have a mixed pipeline (e.g. Spark + Beam) such connectors could support the push-downs of pure Spark pipelines and then use the result downstream in Beam.
—
Alexey
Brian
On Mon, May 23, 2022 at 11:35 AM Robert Bradshaw <ro...@google.com>> wrote:
The easiest way to do this would be to write the RDD somewhere then
read it from Beam.
On Mon, May 23, 2022 at 9:39 AM Yushu Yao <ya...@gmail.com>> wrote:
>
> Hi Folks,
>
> I know this is not the optimal way to use beam :-) But assume I only use the spark runner.
>
> I have a spark library (very complex) that emits a spark dataframe (or RDD).
> I also have an existing complex beam pipeline that can do post processing on the data inside the dataframe.
>
> However, the beam part needs a pcollection to start with. The question is, how can I convert a spark RDD into a pcollection?
>
> Thanks
> -Yushu
>
As a recipient of an email from Talend, your contact personal data will be on our systems. Please see our privacy notice. <https://www.talend.com/privacy/>
Re: RDD (Spark dataframe) into a PCollection?
Posted by Moritz Mack <mm...@talend.com>.
Hi Yushu,
Have a look at org.apache.beam.runners.spark.translation.EvaluationContext in the Spark runner. It maintains that mapping between PCollections and RDDs (wrapped in the BoundedDataset helper). As Reuven just pointed out, values are timestamped (and windowed) in Beam, therefore BoundedDataset expects a JavaRDD<WindowedValue<T>>.
The idea is to map your external RDD to a new PCollection (PCollection.createPrimitiveOutputInternal) in the EvaluationContext (and vice versa). You can then apply Beam transforms to that PCollection (and with that effectively to the mapped RDD) as you are used to. Obviously, there’s a few steps necessary as EvaluationContext isn’t easily accessible from the outside.
Just having a quick look, creating a RddSource for Spark RDDs seems also not too bad. That would allow you to do something like this:
pipeline.apply(Read.from(new RddSource<>(javaRdd.rdd(), coder)));
Though I haven’t done much testing beyond a quick experiment. One notable disadvantage of that approach Is that all RDD partition data must be broadcasted to all workers to then pick the right partition. This should mostly be fine, but some types of partitions carry data as well …
https://gist.github.com/mosche/5c1ef8ba281a9a08df1ec67fac700d03
/Moritz
On 24.05.22, 16:46, "Yushu Yao" <ya...@gmail.com> wrote:
Looks like it's a valid use case. Wondering anyone can give some high level guidelines on how to implement this? I can give it a try. -Yushu On Tue, May 24, 2022 at 2:42 AM Jan Lukavský <je...@seznam.cz> wrote:
ZjQcmQRYFpfptBannerStart
This Message Is From an External Sender
This message came from outside your organization.
Exercise caution when opening attachments or clicking any links.
ZjQcmQRYFpfptBannerEnd
Looks like it's a valid use case.
Wondering anyone can give some high level guidelines on how to implement this?
I can give it a try.
-Yushu
On Tue, May 24, 2022 at 2:42 AM Jan Lukavský <je...@seznam.cz>> wrote:
+dev@beam<ma...@beam.apache.org>
On 5/24/22 11:40, Jan Lukavský wrote:
Hi,
I think this feature is valid. Every runner for which Beam is not a 'native' SDK uses some form of translation context, which maps PCollection to internal representation of the particular SDK of the runner (RDD in this case). It should be possible to "import" an RDD into the specific runner via something like
SparkRunner runner = ....;
PCollection<...> pCollection = runner.importRDD(rdd);
and similarly
RDD<...> rdd = runner.exportRDD(pCollection);
Yes, apparently this would be runner specific, but that is the point, actually. This would enable using features and libraries, that Beam does not have, or micro-optimize some particular step using runner-specific features, that we don't have in Beam. We actually had this feature (at least in a prototype) many years ago when Euphoria was a separate project.
Jan
On 5/23/22 20:58, Alexey Romanenko wrote:
On 23 May 2022, at 20:40, Brian Hulette <bh...@google.com>> wrote:
Yeah I'm not sure of any simple way to do this. I wonder if it's worth considering building some Spark runner-specific feature around this, or at least packaging up Robert's proposed solution?
I’m not sure that a runner specific feature is a good way to do this since the other runners won’t be able to support it or I’m missing something?
There could be other interesting integrations in this space too, e.g. using Spark RDDs as a cache for Interactive Beam.
Another option could be to add something like SparkIO (or FlinkIO/whatever) to read/write data from/to Spark data structures for such cases (Spark schema to Beam schema convention also could be supported). And dreaming a bit more, for those who need to have a mixed pipeline (e.g. Spark + Beam) such connectors could support the push-downs of pure Spark pipelines and then use the result downstream in Beam.
—
Alexey
Brian
On Mon, May 23, 2022 at 11:35 AM Robert Bradshaw <ro...@google.com>> wrote:
The easiest way to do this would be to write the RDD somewhere then
read it from Beam.
On Mon, May 23, 2022 at 9:39 AM Yushu Yao <ya...@gmail.com>> wrote:
>
> Hi Folks,
>
> I know this is not the optimal way to use beam :-) But assume I only use the spark runner.
>
> I have a spark library (very complex) that emits a spark dataframe (or RDD).
> I also have an existing complex beam pipeline that can do post processing on the data inside the dataframe.
>
> However, the beam part needs a pcollection to start with. The question is, how can I convert a spark RDD into a pcollection?
>
> Thanks
> -Yushu
>
As a recipient of an email from Talend, your contact personal data will be on our systems. Please see our privacy notice. <https://www.talend.com/privacy/>
Re: RDD (Spark dataframe) into a PCollection?
Posted by Yushu Yao <ya...@gmail.com>.
Looks like it's a valid use case.
Wondering anyone can give some high level guidelines on how to implement
this?
I can give it a try.
-Yushu
On Tue, May 24, 2022 at 2:42 AM Jan Lukavský <je...@seznam.cz> wrote:
> +dev@beam <de...@beam.apache.org>
> On 5/24/22 11:40, Jan Lukavský wrote:
>
> Hi,
> I think this feature is valid. Every runner for which Beam is not a
> 'native' SDK uses some form of translation context, which maps PCollection
> to internal representation of the particular SDK of the runner (RDD in this
> case). It should be possible to "import" an RDD into the specific runner
> via something like
>
> SparkRunner runner = ....;
> PCollection<...> pCollection = runner.importRDD(rdd);
>
> and similarly
>
> RDD<...> rdd = runner.exportRDD(pCollection);
>
> Yes, apparently this would be runner specific, but that is the point,
> actually. This would enable using features and libraries, that Beam does
> not have, or micro-optimize some particular step using runner-specific
> features, that we don't have in Beam. We actually had this feature (at
> least in a prototype) many years ago when Euphoria was a separate project.
>
> Jan
> On 5/23/22 20:58, Alexey Romanenko wrote:
>
>
>
> On 23 May 2022, at 20:40, Brian Hulette <bh...@google.com> wrote:
>
> Yeah I'm not sure of any simple way to do this. I wonder if it's worth
> considering building some Spark runner-specific feature around this, or at
> least packaging up Robert's proposed solution?
>
>
> I’m not sure that a runner specific feature is a good way to do this since
> the other runners won’t be able to support it or I’m missing something?
>
> There could be other interesting integrations in this space too, e.g.
> using Spark RDDs as a cache for Interactive Beam.
>
>
> Another option could be to add something like SparkIO (or
> FlinkIO/whatever) to read/write data from/to Spark data structures for such
> cases (Spark schema to Beam schema convention also could be supported). And
> dreaming a bit more, for those who need to have a mixed pipeline (e.g.
> Spark + Beam) such connectors could support the push-downs of pure Spark
> pipelines and then use the result downstream in Beam.
>
> —
> Alexey
>
>
>
> Brian
>
> On Mon, May 23, 2022 at 11:35 AM Robert Bradshaw <ro...@google.com>
> wrote:
>
>> The easiest way to do this would be to write the RDD somewhere then
>> read it from Beam.
>>
>> On Mon, May 23, 2022 at 9:39 AM Yushu Yao <ya...@gmail.com> wrote:
>> >
>> > Hi Folks,
>> >
>> > I know this is not the optimal way to use beam :-) But assume I only
>> use the spark runner.
>> >
>> > I have a spark library (very complex) that emits a spark dataframe (or
>> RDD).
>> > I also have an existing complex beam pipeline that can do post
>> processing on the data inside the dataframe.
>> >
>> > However, the beam part needs a pcollection to start with. The question
>> is, how can I convert a spark RDD into a pcollection?
>> >
>> > Thanks
>> > -Yushu
>> >
>>
>
>
Re: RDD (Spark dataframe) into a PCollection?
Posted by Yushu Yao <ya...@gmail.com>.
Looks like it's a valid use case.
Wondering anyone can give some high level guidelines on how to implement
this?
I can give it a try.
-Yushu
On Tue, May 24, 2022 at 2:42 AM Jan Lukavský <je...@seznam.cz> wrote:
> +dev@beam <de...@beam.apache.org>
> On 5/24/22 11:40, Jan Lukavský wrote:
>
> Hi,
> I think this feature is valid. Every runner for which Beam is not a
> 'native' SDK uses some form of translation context, which maps PCollection
> to internal representation of the particular SDK of the runner (RDD in this
> case). It should be possible to "import" an RDD into the specific runner
> via something like
>
> SparkRunner runner = ....;
> PCollection<...> pCollection = runner.importRDD(rdd);
>
> and similarly
>
> RDD<...> rdd = runner.exportRDD(pCollection);
>
> Yes, apparently this would be runner specific, but that is the point,
> actually. This would enable using features and libraries, that Beam does
> not have, or micro-optimize some particular step using runner-specific
> features, that we don't have in Beam. We actually had this feature (at
> least in a prototype) many years ago when Euphoria was a separate project.
>
> Jan
> On 5/23/22 20:58, Alexey Romanenko wrote:
>
>
>
> On 23 May 2022, at 20:40, Brian Hulette <bh...@google.com> wrote:
>
> Yeah I'm not sure of any simple way to do this. I wonder if it's worth
> considering building some Spark runner-specific feature around this, or at
> least packaging up Robert's proposed solution?
>
>
> I’m not sure that a runner specific feature is a good way to do this since
> the other runners won’t be able to support it or I’m missing something?
>
> There could be other interesting integrations in this space too, e.g.
> using Spark RDDs as a cache for Interactive Beam.
>
>
> Another option could be to add something like SparkIO (or
> FlinkIO/whatever) to read/write data from/to Spark data structures for such
> cases (Spark schema to Beam schema convention also could be supported). And
> dreaming a bit more, for those who need to have a mixed pipeline (e.g.
> Spark + Beam) such connectors could support the push-downs of pure Spark
> pipelines and then use the result downstream in Beam.
>
> —
> Alexey
>
>
>
> Brian
>
> On Mon, May 23, 2022 at 11:35 AM Robert Bradshaw <ro...@google.com>
> wrote:
>
>> The easiest way to do this would be to write the RDD somewhere then
>> read it from Beam.
>>
>> On Mon, May 23, 2022 at 9:39 AM Yushu Yao <ya...@gmail.com> wrote:
>> >
>> > Hi Folks,
>> >
>> > I know this is not the optimal way to use beam :-) But assume I only
>> use the spark runner.
>> >
>> > I have a spark library (very complex) that emits a spark dataframe (or
>> RDD).
>> > I also have an existing complex beam pipeline that can do post
>> processing on the data inside the dataframe.
>> >
>> > However, the beam part needs a pcollection to start with. The question
>> is, how can I convert a spark RDD into a pcollection?
>> >
>> > Thanks
>> > -Yushu
>> >
>>
>
>
Re: RDD (Spark dataframe) into a PCollection?
Posted by Jan Lukavský <je...@seznam.cz>.
+dev@beam <ma...@beam.apache.org>
On 5/24/22 11:40, Jan Lukavský wrote:
>
> Hi,
> I think this feature is valid. Every runner for which Beam is not a
> 'native' SDK uses some form of translation context, which maps
> PCollection to internal representation of the particular SDK of the
> runner (RDD in this case). It should be possible to "import" an RDD
> into the specific runner via something like
>
> SparkRunner runner = ....;
> PCollection<...> pCollection = runner.importRDD(rdd);
>
> and similarly
>
> RDD<...> rdd = runner.exportRDD(pCollection);
>
> Yes, apparently this would be runner specific, but that is the point,
> actually. This would enable using features and libraries, that Beam
> does not have, or micro-optimize some particular step using
> runner-specific features, that we don't have in Beam. We actually had
> this feature (at least in a prototype) many years ago when Euphoria
> was a separate project.
>
> Jan
>
> On 5/23/22 20:58, Alexey Romanenko wrote:
>>
>>
>>> On 23 May 2022, at 20:40, Brian Hulette <bh...@google.com> wrote:
>>>
>>> Yeah I'm not sure of any simple way to do this. I wonder if it's
>>> worth considering building some Spark runner-specific feature around
>>> this, or at least packaging up Robert's proposed solution?
>>
>> I’m not sure that a runner specific feature is a good way to do this
>> since the other runners won’t be able to support it or I’m missing
>> something?
>>
>>> There could be other interesting integrations in this space too,
>>> e.g. using Spark RDDs as a cache for Interactive Beam.
>>
>> Another option could be to add something like SparkIO (or
>> FlinkIO/whatever) to read/write data from/to Spark data structures
>> for such cases (Spark schema to Beam schema convention also could be
>> supported). And dreaming a bit more, for those who need to have a
>> mixed pipeline (e.g. Spark + Beam) such connectors could support the
>> push-downs of pure Spark pipelines and then use the result downstream
>> in Beam.
>>
>> —
>> Alexey
>>
>>
>>>
>>> Brian
>>>
>>> On Mon, May 23, 2022 at 11:35 AM Robert Bradshaw
>>> <ro...@google.com> wrote:
>>>
>>> The easiest way to do this would be to write the RDD somewhere then
>>> read it from Beam.
>>>
>>> On Mon, May 23, 2022 at 9:39 AM Yushu Yao <ya...@gmail.com>
>>> wrote:
>>> >
>>> > Hi Folks,
>>> >
>>> > I know this is not the optimal way to use beam :-) But assume
>>> I only use the spark runner.
>>> >
>>> > I have a spark library (very complex) that emits a spark
>>> dataframe (or RDD).
>>> > I also have an existing complex beam pipeline that can do post
>>> processing on the data inside the dataframe.
>>> >
>>> > However, the beam part needs a pcollection to start with. The
>>> question is, how can I convert a spark RDD into a pcollection?
>>> >
>>> > Thanks
>>> > -Yushu
>>> >
>>>
>>
Re: RDD (Spark dataframe) into a PCollection?
Posted by Jan Lukavský <je...@seznam.cz>.
+dev@beam <ma...@beam.apache.org>
On 5/24/22 11:40, Jan Lukavský wrote:
>
> Hi,
> I think this feature is valid. Every runner for which Beam is not a
> 'native' SDK uses some form of translation context, which maps
> PCollection to internal representation of the particular SDK of the
> runner (RDD in this case). It should be possible to "import" an RDD
> into the specific runner via something like
>
> SparkRunner runner = ....;
> PCollection<...> pCollection = runner.importRDD(rdd);
>
> and similarly
>
> RDD<...> rdd = runner.exportRDD(pCollection);
>
> Yes, apparently this would be runner specific, but that is the point,
> actually. This would enable using features and libraries, that Beam
> does not have, or micro-optimize some particular step using
> runner-specific features, that we don't have in Beam. We actually had
> this feature (at least in a prototype) many years ago when Euphoria
> was a separate project.
>
> Jan
>
> On 5/23/22 20:58, Alexey Romanenko wrote:
>>
>>
>>> On 23 May 2022, at 20:40, Brian Hulette <bh...@google.com> wrote:
>>>
>>> Yeah I'm not sure of any simple way to do this. I wonder if it's
>>> worth considering building some Spark runner-specific feature around
>>> this, or at least packaging up Robert's proposed solution?
>>
>> I’m not sure that a runner specific feature is a good way to do this
>> since the other runners won’t be able to support it or I’m missing
>> something?
>>
>>> There could be other interesting integrations in this space too,
>>> e.g. using Spark RDDs as a cache for Interactive Beam.
>>
>> Another option could be to add something like SparkIO (or
>> FlinkIO/whatever) to read/write data from/to Spark data structures
>> for such cases (Spark schema to Beam schema convention also could be
>> supported). And dreaming a bit more, for those who need to have a
>> mixed pipeline (e.g. Spark + Beam) such connectors could support the
>> push-downs of pure Spark pipelines and then use the result downstream
>> in Beam.
>>
>> —
>> Alexey
>>
>>
>>>
>>> Brian
>>>
>>> On Mon, May 23, 2022 at 11:35 AM Robert Bradshaw
>>> <ro...@google.com> wrote:
>>>
>>> The easiest way to do this would be to write the RDD somewhere then
>>> read it from Beam.
>>>
>>> On Mon, May 23, 2022 at 9:39 AM Yushu Yao <ya...@gmail.com>
>>> wrote:
>>> >
>>> > Hi Folks,
>>> >
>>> > I know this is not the optimal way to use beam :-) But assume
>>> I only use the spark runner.
>>> >
>>> > I have a spark library (very complex) that emits a spark
>>> dataframe (or RDD).
>>> > I also have an existing complex beam pipeline that can do post
>>> processing on the data inside the dataframe.
>>> >
>>> > However, the beam part needs a pcollection to start with. The
>>> question is, how can I convert a spark RDD into a pcollection?
>>> >
>>> > Thanks
>>> > -Yushu
>>> >
>>>
>>
Re: RDD (Spark dataframe) into a PCollection?
Posted by Jan Lukavský <je...@seznam.cz>.
Yes, I suppose it might be more complex than the code snippet, that was
just to demonstrate the idea. Also the "exportRDD" would probably return
WindowedValue<T> instead of plain T.
On 5/24/22 17:23, Reuven Lax wrote:
> Something like this seems reasonable. Beam PCollections also have a
> timestamp associated with every element, so the importRDD function
> probably needs a way to specify the timestamp (could be an attribute
> name for dataframes or a timestamp extraction function for regular RDDs).
>
> On Tue, May 24, 2022 at 2:40 AM Jan Lukavský <je...@seznam.cz> wrote:
>
> Hi,
> I think this feature is valid. Every runner for which Beam is not
> a 'native' SDK uses some form of translation context, which maps
> PCollection to internal representation of the particular SDK of
> the runner (RDD in this case). It should be possible to "import"
> an RDD into the specific runner via something like
>
> SparkRunner runner = ....;
> PCollection<...> pCollection = runner.importRDD(rdd);
>
> and similarly
>
> RDD<...> rdd = runner.exportRDD(pCollection);
>
> Yes, apparently this would be runner specific, but that is the
> point, actually. This would enable using features and libraries,
> that Beam does not have, or micro-optimize some particular step
> using runner-specific features, that we don't have in Beam. We
> actually had this feature (at least in a prototype) many years ago
> when Euphoria was a separate project.
>
> Jan
>
> On 5/23/22 20:58, Alexey Romanenko wrote:
>>
>>
>>> On 23 May 2022, at 20:40, Brian Hulette <bh...@google.com> wrote:
>>>
>>> Yeah I'm not sure of any simple way to do this. I wonder if it's
>>> worth considering building some Spark runner-specific feature
>>> around this, or at least packaging up Robert's proposed solution?
>>
>> I’m not sure that a runner specific feature is a good way to do
>> this since the other runners won’t be able to support it or I’m
>> missing something?
>>
>>> There could be other interesting integrations in this space too,
>>> e.g. using Spark RDDs as a cache for Interactive Beam.
>>
>> Another option could be to add something like SparkIO (or
>> FlinkIO/whatever) to read/write data from/to Spark data
>> structures for such cases (Spark schema to Beam schema convention
>> also could be supported). And dreaming a bit more, for those who
>> need to have a mixed pipeline (e.g. Spark + Beam) such connectors
>> could support the push-downs of pure Spark pipelines and then use
>> the result downstream in Beam.
>>
>> —
>> Alexey
>>
>>
>>>
>>> Brian
>>>
>>> On Mon, May 23, 2022 at 11:35 AM Robert Bradshaw
>>> <ro...@google.com> wrote:
>>>
>>> The easiest way to do this would be to write the RDD
>>> somewhere then
>>> read it from Beam.
>>>
>>> On Mon, May 23, 2022 at 9:39 AM Yushu Yao
>>> <ya...@gmail.com> wrote:
>>> >
>>> > Hi Folks,
>>> >
>>> > I know this is not the optimal way to use beam :-) But
>>> assume I only use the spark runner.
>>> >
>>> > I have a spark library (very complex) that emits a spark
>>> dataframe (or RDD).
>>> > I also have an existing complex beam pipeline that can do
>>> post processing on the data inside the dataframe.
>>> >
>>> > However, the beam part needs a pcollection to start with.
>>> The question is, how can I convert a spark RDD into a
>>> pcollection?
>>> >
>>> > Thanks
>>> > -Yushu
>>> >
>>>
>>
Re: RDD (Spark dataframe) into a PCollection?
Posted by Reuven Lax <re...@google.com>.
Something like this seems reasonable. Beam PCollections also have a
timestamp associated with every element, so the importRDD function probably
needs a way to specify the timestamp (could be an attribute name for
dataframes or a timestamp extraction function for regular RDDs).
On Tue, May 24, 2022 at 2:40 AM Jan Lukavský <je...@seznam.cz> wrote:
> Hi,
> I think this feature is valid. Every runner for which Beam is not a
> 'native' SDK uses some form of translation context, which maps PCollection
> to internal representation of the particular SDK of the runner (RDD in this
> case). It should be possible to "import" an RDD into the specific runner
> via something like
>
> SparkRunner runner = ....;
> PCollection<...> pCollection = runner.importRDD(rdd);
>
> and similarly
>
> RDD<...> rdd = runner.exportRDD(pCollection);
>
> Yes, apparently this would be runner specific, but that is the point,
> actually. This would enable using features and libraries, that Beam does
> not have, or micro-optimize some particular step using runner-specific
> features, that we don't have in Beam. We actually had this feature (at
> least in a prototype) many years ago when Euphoria was a separate project.
>
> Jan
> On 5/23/22 20:58, Alexey Romanenko wrote:
>
>
>
> On 23 May 2022, at 20:40, Brian Hulette <bh...@google.com> wrote:
>
> Yeah I'm not sure of any simple way to do this. I wonder if it's worth
> considering building some Spark runner-specific feature around this, or at
> least packaging up Robert's proposed solution?
>
>
> I’m not sure that a runner specific feature is a good way to do this since
> the other runners won’t be able to support it or I’m missing something?
>
> There could be other interesting integrations in this space too, e.g.
> using Spark RDDs as a cache for Interactive Beam.
>
>
> Another option could be to add something like SparkIO (or
> FlinkIO/whatever) to read/write data from/to Spark data structures for such
> cases (Spark schema to Beam schema convention also could be supported). And
> dreaming a bit more, for those who need to have a mixed pipeline (e.g.
> Spark + Beam) such connectors could support the push-downs of pure Spark
> pipelines and then use the result downstream in Beam.
>
> —
> Alexey
>
>
>
> Brian
>
> On Mon, May 23, 2022 at 11:35 AM Robert Bradshaw <ro...@google.com>
> wrote:
>
>> The easiest way to do this would be to write the RDD somewhere then
>> read it from Beam.
>>
>> On Mon, May 23, 2022 at 9:39 AM Yushu Yao <ya...@gmail.com> wrote:
>> >
>> > Hi Folks,
>> >
>> > I know this is not the optimal way to use beam :-) But assume I only
>> use the spark runner.
>> >
>> > I have a spark library (very complex) that emits a spark dataframe (or
>> RDD).
>> > I also have an existing complex beam pipeline that can do post
>> processing on the data inside the dataframe.
>> >
>> > However, the beam part needs a pcollection to start with. The question
>> is, how can I convert a spark RDD into a pcollection?
>> >
>> > Thanks
>> > -Yushu
>> >
>>
>
>
Re: RDD (Spark dataframe) into a PCollection?
Posted by Jan Lukavský <je...@seznam.cz>.
Hi,
I think this feature is valid. Every runner for which Beam is not a
'native' SDK uses some form of translation context, which maps
PCollection to internal representation of the particular SDK of the
runner (RDD in this case). It should be possible to "import" an RDD into
the specific runner via something like
SparkRunner runner = ....;
PCollection<...> pCollection = runner.importRDD(rdd);
and similarly
RDD<...> rdd = runner.exportRDD(pCollection);
Yes, apparently this would be runner specific, but that is the point,
actually. This would enable using features and libraries, that Beam does
not have, or micro-optimize some particular step using runner-specific
features, that we don't have in Beam. We actually had this feature (at
least in a prototype) many years ago when Euphoria was a separate project.
Jan
On 5/23/22 20:58, Alexey Romanenko wrote:
>
>
>> On 23 May 2022, at 20:40, Brian Hulette <bh...@google.com> wrote:
>>
>> Yeah I'm not sure of any simple way to do this. I wonder if it's
>> worth considering building some Spark runner-specific feature around
>> this, or at least packaging up Robert's proposed solution?
>
> I’m not sure that a runner specific feature is a good way to do this
> since the other runners won’t be able to support it or I’m missing
> something?
>
>> There could be other interesting integrations in this space too, e.g.
>> using Spark RDDs as a cache for Interactive Beam.
>
> Another option could be to add something like SparkIO (or
> FlinkIO/whatever) to read/write data from/to Spark data structures for
> such cases (Spark schema to Beam schema convention also could be
> supported). And dreaming a bit more, for those who need to have a
> mixed pipeline (e.g. Spark + Beam) such connectors could support the
> push-downs of pure Spark pipelines and then use the result downstream
> in Beam.
>
> —
> Alexey
>
>
>>
>> Brian
>>
>> On Mon, May 23, 2022 at 11:35 AM Robert Bradshaw
>> <ro...@google.com> wrote:
>>
>> The easiest way to do this would be to write the RDD somewhere then
>> read it from Beam.
>>
>> On Mon, May 23, 2022 at 9:39 AM Yushu Yao <ya...@gmail.com>
>> wrote:
>> >
>> > Hi Folks,
>> >
>> > I know this is not the optimal way to use beam :-) But assume I
>> only use the spark runner.
>> >
>> > I have a spark library (very complex) that emits a spark
>> dataframe (or RDD).
>> > I also have an existing complex beam pipeline that can do post
>> processing on the data inside the dataframe.
>> >
>> > However, the beam part needs a pcollection to start with. The
>> question is, how can I convert a spark RDD into a pcollection?
>> >
>> > Thanks
>> > -Yushu
>> >
>>
>
Re: RDD (Spark dataframe) into a PCollection?
Posted by Alexey Romanenko <ar...@gmail.com>.
> On 23 May 2022, at 20:40, Brian Hulette <bh...@google.com> wrote:
>
> Yeah I'm not sure of any simple way to do this. I wonder if it's worth considering building some Spark runner-specific feature around this, or at least packaging up Robert's proposed solution?
I’m not sure that a runner specific feature is a good way to do this since the other runners won’t be able to support it or I’m missing something?
> There could be other interesting integrations in this space too, e.g. using Spark RDDs as a cache for Interactive Beam.
Another option could be to add something like SparkIO (or FlinkIO/whatever) to read/write data from/to Spark data structures for such cases (Spark schema to Beam schema convention also could be supported). And dreaming a bit more, for those who need to have a mixed pipeline (e.g. Spark + Beam) such connectors could support the push-downs of pure Spark pipelines and then use the result downstream in Beam.
—
Alexey
>
> Brian
>
> On Mon, May 23, 2022 at 11:35 AM Robert Bradshaw <robertwb@google.com <ma...@google.com>> wrote:
> The easiest way to do this would be to write the RDD somewhere then
> read it from Beam.
>
> On Mon, May 23, 2022 at 9:39 AM Yushu Yao <yao.yushu@gmail.com <ma...@gmail.com>> wrote:
> >
> > Hi Folks,
> >
> > I know this is not the optimal way to use beam :-) But assume I only use the spark runner.
> >
> > I have a spark library (very complex) that emits a spark dataframe (or RDD).
> > I also have an existing complex beam pipeline that can do post processing on the data inside the dataframe.
> >
> > However, the beam part needs a pcollection to start with. The question is, how can I convert a spark RDD into a pcollection?
> >
> > Thanks
> > -Yushu
> >
Re: RDD (Spark dataframe) into a PCollection?
Posted by Brian Hulette <bh...@google.com>.
Yeah I'm not sure of any simple way to do this. I wonder if it's worth
considering building some Spark runner-specific feature around this, or at
least packaging up Robert's proposed solution?
There could be other interesting integrations in this space too, e.g. using
Spark RDDs as a cache for Interactive Beam.
Brian
On Mon, May 23, 2022 at 11:35 AM Robert Bradshaw <ro...@google.com>
wrote:
> The easiest way to do this would be to write the RDD somewhere then
> read it from Beam.
>
> On Mon, May 23, 2022 at 9:39 AM Yushu Yao <ya...@gmail.com> wrote:
> >
> > Hi Folks,
> >
> > I know this is not the optimal way to use beam :-) But assume I only use
> the spark runner.
> >
> > I have a spark library (very complex) that emits a spark dataframe (or
> RDD).
> > I also have an existing complex beam pipeline that can do post
> processing on the data inside the dataframe.
> >
> > However, the beam part needs a pcollection to start with. The question
> is, how can I convert a spark RDD into a pcollection?
> >
> > Thanks
> > -Yushu
> >
>
Re: RDD (Spark dataframe) into a PCollection?
Posted by Alexey Romanenko <ar...@gmail.com>.
To add a bit more to what Robert suggested. Right, in general we can’t read Spark RDD directly with Beam (Spark runner uses RDD under the hood but it’s a different story) but you can write the results to any storage and in data format that Beam supports and then read it with a corespondent Beam IO connector.
—
Alexey
> On 23 May 2022, at 20:35, Robert Bradshaw <ro...@google.com> wrote:
>
> The easiest way to do this would be to write the RDD somewhere then
> read it from Beam.
>
> On Mon, May 23, 2022 at 9:39 AM Yushu Yao <ya...@gmail.com> wrote:
>>
>> Hi Folks,
>>
>> I know this is not the optimal way to use beam :-) But assume I only use the spark runner.
>>
>> I have a spark library (very complex) that emits a spark dataframe (or RDD).
>> I also have an existing complex beam pipeline that can do post processing on the data inside the dataframe.
>>
>> However, the beam part needs a pcollection to start with. The question is, how can I convert a spark RDD into a pcollection?
>>
>> Thanks
>> -Yushu
>>
Re: RDD (Spark dataframe) into a PCollection?
Posted by Robert Bradshaw <ro...@google.com>.
The easiest way to do this would be to write the RDD somewhere then
read it from Beam.
On Mon, May 23, 2022 at 9:39 AM Yushu Yao <ya...@gmail.com> wrote:
>
> Hi Folks,
>
> I know this is not the optimal way to use beam :-) But assume I only use the spark runner.
>
> I have a spark library (very complex) that emits a spark dataframe (or RDD).
> I also have an existing complex beam pipeline that can do post processing on the data inside the dataframe.
>
> However, the beam part needs a pcollection to start with. The question is, how can I convert a spark RDD into a pcollection?
>
> Thanks
> -Yushu
>