You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by MoTao <mo...@sensetime.com> on 2016/08/08 09:00:51 UTC
how to generate a column using mapParition and then add it back to
the df?
Hi all,
I'm trying to append a column to a df.
I understand that the new column must be created by
1) using literals,
2) transforming an existing column in df,
or 3) generated from udf over this df
In my case, the column to be appended is created by processing each row,
like
val df = spark.createDataFrame(Seq(1.0, 2.0, 3.0)).toDF("value")
val func = udf {
v: Double => {
val model = initModel()
model.process(v)
}
}
val df2 = df.withColumn("valueWithBias", func(col("value")))
This works fine. However, for performance reason, I want to avoid
initModel() for each row.
So I come with mapParitions, like
val df = spark.createDataFrame(Seq(1.0, 2.0, 3.0)).toDF("value")
val df2 = df.mapPartitions(rows => {
val model = initModel()
rows.map(row => model.process(row.getAs[Double](0)))
})
val df3 = df.withColumn("valueWithBias", df2.col("value")) // FAIL
But this is wrong as a column of df2 *CANNOT* be appended to df.
The only solution I got is to force mapPartitions to return a whole row
instead of the new column,
( Something like "row => Row.fromSeq(row.toSeq ++
Array(model.process(...)))" )
which requires a lot of copy as well.
I wonder how to deal with this problem with as few overhead as possible?
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/how-to-generate-a-column-using-mapParition-and-then-add-it-back-to-the-df-tp27493.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe e-mail: user-unsubscribe@spark.apache.org
答复: 答复: how to generate a column using mapParition and then add it back to the df?
Posted by 莫涛 <mo...@sensetime.com>.
Hi guha,
Thanks a lot!
This is perfectly what I want and I'll try to implement it.
MoTao
________________________________
发件人: ayan guha <gu...@gmail.com>
发送时间: 2016年8月8日 18:05:37
收件人: 莫涛
抄送: ndjido@gmail.com; user@spark.apache.org
主题: Re: 答复: how to generate a column using mapParition and then add it back to the df?
Hi
I think you should modify initModel() function to getOrCreateModel() and create the model as singleton object. You may want to refer this link<http://stackoverflow.com/questions/30450763/spark-streaming-and-connection-pool-implementation>
On Mon, Aug 8, 2016 at 7:44 PM, 莫涛 <mo...@sensetime.com>> wrote:
Hi Ndjido,
Thanks for your reply.
Yes, it is good idea if the model can be broadcast.
I'm working with a built library (on Linux, say classifier.so and classifier.h) and it requires the model file is in the local file system.
As I don't have access to the library code, I write JNI to wrap the classifier.
The model file can be sent to each executor efficiently by addFile and getFile.
But initModel() is still expensive as it actually loads a local file into C++ heap memory, which is not serializable.
That's the reason I can not broadcast the model and I have to avoid load model as possible as I can.
Best
________________________________
发件人: ndjido@gmail.com<ma...@gmail.com> <nd...@gmail.com>>
发送时间: 2016年8月8日 17:16:27
收件人: 莫涛
抄送: user@spark.apache.org<ma...@spark.apache.org>
主题: Re: how to generate a column using mapParition and then add it back to the df?
Hi MoTao,
What about broadcasting the model?
Cheers,
Ndjido.
> On 08 Aug 2016, at 11:00, MoTao <mo...@sensetime.com>> wrote:
>
> Hi all,
>
> I'm trying to append a column to a df.
> I understand that the new column must be created by
> 1) using literals,
> 2) transforming an existing column in df,
> or 3) generated from udf over this df
>
> In my case, the column to be appended is created by processing each row,
> like
>
> val df = spark.createDataFrame(Seq(1.0, 2.0, 3.0)).toDF("value")
> val func = udf {
> v: Double => {
> val model = initModel()
> model.process(v)
> }
> }
> val df2 = df.withColumn("valueWithBias", func(col("value")))
>
> This works fine. However, for performance reason, I want to avoid
> initModel() for each row.
> So I come with mapParitions, like
>
> val df = spark.createDataFrame(Seq(1.0, 2.0, 3.0)).toDF("value")
> val df2 = df.mapPartitions(rows => {
> val model = initModel()
> rows.map(row => model.process(row.getAs[Double](0)))
> })
> val df3 = df.withColumn("valueWithBias", df2.col("value")) // FAIL
>
> But this is wrong as a column of df2 *CANNOT* be appended to df.
>
> The only solution I got is to force mapPartitions to return a whole row
> instead of the new column,
> ( Something like "row => Row.fromSeq(row.toSeq ++
> Array(model.process(...)))" )
> which requires a lot of copy as well.
>
> I wonder how to deal with this problem with as few overhead as possible?
>
>
>
>
> --
> View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/how-to-generate-a-column-using-mapParition-and-then-add-it-back-to-the-df-tp27493.html
> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>
> ---------------------------------------------------------------------
> To unsubscribe e-mail: user-unsubscribe@spark.apache.org<ma...@spark.apache.org>
>
--
Best Regards,
Ayan Guha
Re: 答复: how to generate a column using mapParition and then add it back to the df?
Posted by ayan guha <gu...@gmail.com>.
Hi
I think you should modify initModel() function to getOrCreateModel() and
create the model as singleton object. You may want to refer this link
<http://stackoverflow.com/questions/30450763/spark-streaming-and-connection-pool-implementation>
On Mon, Aug 8, 2016 at 7:44 PM, 莫涛 <mo...@sensetime.com> wrote:
> Hi Ndjido,
>
> Thanks for your reply.
>
> Yes, it is good idea if the model can be broadcast.
>
> I'm working with a built library (on Linux, say classifier.so and
> classifier.h) and it requires the model file is in the local file system.
> As I don't have access to the library code, I write JNI to wrap the
> classifier.
> The model file can be sent to each executor efficiently by addFile and
> getFile.
> But initModel() is still expensive as it actually loads a local file into
> C++ heap memory, which is not serializable.
>
> That's the reason I can not broadcast the model and I have to avoid load
> model as possible as I can.
>
> Best
>
> ------------------------------
> *发件人:* ndjido@gmail.com <nd...@gmail.com>
> *发送时间:* 2016年8月8日 17:16:27
> *收件人:* 莫涛
> *抄送:* user@spark.apache.org
> *主题:* Re: how to generate a column using mapParition and then add it back
> to the df?
>
>
> Hi MoTao,
> What about broadcasting the model?
>
> Cheers,
> Ndjido.
>
> > On 08 Aug 2016, at 11:00, MoTao <mo...@sensetime.com> wrote:
> >
> > Hi all,
> >
> > I'm trying to append a column to a df.
> > I understand that the new column must be created by
> > 1) using literals,
> > 2) transforming an existing column in df,
> > or 3) generated from udf over this df
> >
> > In my case, the column to be appended is created by processing each row,
> > like
> >
> > val df = spark.createDataFrame(Seq(1.0, 2.0, 3.0)).toDF("value")
> > val func = udf {
> > v: Double => {
> > val model = initModel()
> > model.process(v)
> > }
> > }
> > val df2 = df.withColumn("valueWithBias", func(col("value")))
> >
> > This works fine. However, for performance reason, I want to avoid
> > initModel() for each row.
> > So I come with mapParitions, like
> >
> > val df = spark.createDataFrame(Seq(1.0, 2.0, 3.0)).toDF("value")
> > val df2 = df.mapPartitions(rows => {
> > val model = initModel()
> > rows.map(row => model.process(row.getAs[Double](0)))
> > })
> > val df3 = df.withColumn("valueWithBias", df2.col("value")) // FAIL
> >
> > But this is wrong as a column of df2 *CANNOT* be appended to df.
> >
> > The only solution I got is to force mapPartitions to return a whole row
> > instead of the new column,
> > ( Something like "row => Row.fromSeq(row.toSeq ++
> > Array(model.process(...)))" )
> > which requires a lot of copy as well.
> >
> > I wonder how to deal with this problem with as few overhead as possible?
> >
> >
> >
> >
> > --
> > View this message in context: http://apache-spark-user-list.
> 1001560.n3.nabble.com/how-to-generate-a-column-using-
> mapParition-and-then-add-it-back-to-the-df-tp27493.html
> > Sent from the Apache Spark User List mailing list archive at Nabble.com.
> >
> > ---------------------------------------------------------------------
> > To unsubscribe e-mail: user-unsubscribe@spark.apache.org
> >
>
--
Best Regards,
Ayan Guha
答复: how to generate a column using mapParition and then add it back to the df?
Posted by 莫涛 <mo...@sensetime.com>.
Hi Ndjido,
Thanks for your reply.
Yes, it is good idea if the model can be broadcast.
I'm working with a built library (on Linux, say classifier.so and classifier.h) and it requires the model file is in the local file system.
As I don't have access to the library code, I write JNI to wrap the classifier.
The model file can be sent to each executor efficiently by addFile and getFile.
But initModel() is still expensive as it actually loads a local file into C++ heap memory, which is not serializable.
That's the reason I can not broadcast the model and I have to avoid load model as possible as I can.
Best
________________________________
发件人: ndjido@gmail.com <nd...@gmail.com>
发送时间: 2016年8月8日 17:16:27
收件人: 莫涛
抄送: user@spark.apache.org
主题: Re: how to generate a column using mapParition and then add it back to the df?
Hi MoTao,
What about broadcasting the model?
Cheers,
Ndjido.
> On 08 Aug 2016, at 11:00, MoTao <mo...@sensetime.com> wrote:
>
> Hi all,
>
> I'm trying to append a column to a df.
> I understand that the new column must be created by
> 1) using literals,
> 2) transforming an existing column in df,
> or 3) generated from udf over this df
>
> In my case, the column to be appended is created by processing each row,
> like
>
> val df = spark.createDataFrame(Seq(1.0, 2.0, 3.0)).toDF("value")
> val func = udf {
> v: Double => {
> val model = initModel()
> model.process(v)
> }
> }
> val df2 = df.withColumn("valueWithBias", func(col("value")))
>
> This works fine. However, for performance reason, I want to avoid
> initModel() for each row.
> So I come with mapParitions, like
>
> val df = spark.createDataFrame(Seq(1.0, 2.0, 3.0)).toDF("value")
> val df2 = df.mapPartitions(rows => {
> val model = initModel()
> rows.map(row => model.process(row.getAs[Double](0)))
> })
> val df3 = df.withColumn("valueWithBias", df2.col("value")) // FAIL
>
> But this is wrong as a column of df2 *CANNOT* be appended to df.
>
> The only solution I got is to force mapPartitions to return a whole row
> instead of the new column,
> ( Something like "row => Row.fromSeq(row.toSeq ++
> Array(model.process(...)))" )
> which requires a lot of copy as well.
>
> I wonder how to deal with this problem with as few overhead as possible?
>
>
>
>
> --
> View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/how-to-generate-a-column-using-mapParition-and-then-add-it-back-to-the-df-tp27493.html
> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>
> ---------------------------------------------------------------------
> To unsubscribe e-mail: user-unsubscribe@spark.apache.org
>
Re: how to generate a column using mapParition and then add it back to the df?
Posted by nd...@gmail.com.
Hi MoTao,
What about broadcasting the model?
Cheers,
Ndjido.
> On 08 Aug 2016, at 11:00, MoTao <mo...@sensetime.com> wrote:
>
> Hi all,
>
> I'm trying to append a column to a df.
> I understand that the new column must be created by
> 1) using literals,
> 2) transforming an existing column in df,
> or 3) generated from udf over this df
>
> In my case, the column to be appended is created by processing each row,
> like
>
> val df = spark.createDataFrame(Seq(1.0, 2.0, 3.0)).toDF("value")
> val func = udf {
> v: Double => {
> val model = initModel()
> model.process(v)
> }
> }
> val df2 = df.withColumn("valueWithBias", func(col("value")))
>
> This works fine. However, for performance reason, I want to avoid
> initModel() for each row.
> So I come with mapParitions, like
>
> val df = spark.createDataFrame(Seq(1.0, 2.0, 3.0)).toDF("value")
> val df2 = df.mapPartitions(rows => {
> val model = initModel()
> rows.map(row => model.process(row.getAs[Double](0)))
> })
> val df3 = df.withColumn("valueWithBias", df2.col("value")) // FAIL
>
> But this is wrong as a column of df2 *CANNOT* be appended to df.
>
> The only solution I got is to force mapPartitions to return a whole row
> instead of the new column,
> ( Something like "row => Row.fromSeq(row.toSeq ++
> Array(model.process(...)))" )
> which requires a lot of copy as well.
>
> I wonder how to deal with this problem with as few overhead as possible?
>
>
>
>
> --
> View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/how-to-generate-a-column-using-mapParition-and-then-add-it-back-to-the-df-tp27493.html
> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>
> ---------------------------------------------------------------------
> To unsubscribe e-mail: user-unsubscribe@spark.apache.org
>
---------------------------------------------------------------------
To unsubscribe e-mail: user-unsubscribe@spark.apache.org