[
https://issues.apache.org/jira/browse/SPARK-23650?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16401525#comment-16401525
]
Felix Cheung edited comment on SPARK-23650 at 3/16/18 7:16 AM:
---------------------------------------------------------------
do you mean this?
RRunner: Times: boot = 0.010 s, init = 0.005 s, broadcast = 1.894 s, read-input
= 0.001 s, compute = 0.062 s, write-output = 0.002 s, total = 1.974 s
Under the cover it is working with the same R process.
I see
SparkContext: Created broadcast 3 from broadcast at DAGScheduler.scala:1006
each time it is creating a new broadcast which would then needs to be
transferred.
IMO there are a few things to look into:
# it should detect if the broadcast is the same (not sure if it does that)
# if it is attributed to the same broadcast in use.daemon mode then it perhaps
doesn't have to transfer it again (but it would need to keep track of the stage
executed before and broadcast that was sent before etc)
# data transfer can be faster (SPARK-18924)
However, as of now RRunner simply picks up the broadcast that is pass to it and
sends it.
was (Author: felixcheung):
do you mean this?
RRunner: Times: boot = 0.010 s, init = 0.005 s, broadcast = 1.894 s, read-input
= 0.001 s, compute = 0.062 s, write-output = 0.002 s, total = 1.974 s
Under the cover it is working with the same R process.
I see
SparkContext: Created broadcast 3 from broadcast at DAGScheduler.scala:1006
each time it is creating a new broadcast which would then needs to be
transferred.
IMO there are a few things to look into:
# it should detect if the broadcast is the same (not sure if it does that)
# if it is attributed to the same broadcast in use.daemon mode then it perhaps
doesn't have to transfer it again
# data transfer can be faster (SPARK-18924)
However, as of now RRunner simply picks up the broadcast that is pass to it and
sends it.
> Slow SparkR udf (dapply)
> ------------------------
>
> Key: SPARK-23650
> URL: https://issues.apache.org/jira/browse/SPARK-23650
> Project: Spark
> Issue Type: Improvement
> Components: Spark Shell, SparkR, Structured Streaming
> Affects Versions: 2.2.0
> Reporter: Deepansh
> Priority: Major
> Attachments: sparkR_log2.txt, sparkRlag.txt
>
>
> For eg, I am getting streams from Kafka and I want to implement a model made
> in R for those streams. For this, I am using dapply.
> My code is:
> iris_model <- readRDS("./iris_model.rds")
> randomBr <- SparkR:::broadcast(sc, iris_model)
> kafka <- read.stream("kafka",subscribe = "source", kafka.bootstrap.servers =
> "localhost:9092", topic = "source")
> lines<- select(kafka, cast(kafka$value, "string"))
> schema<-schema(lines)
> df1<-dapply(lines,function(x){
> i_model<-SparkR:::value(randomMatBr)
> for (row in 1:nrow(x))
> { y<-fromJSON(as.character(x[row,"value"])) y$predict=predict(i_model,y)
> y<-toJSON(y) x[row,"value"] = y }
> x
> },schema)
> Every time when Kafka streams are fetched the dapply method creates new
> runner thread and ships the variables again, which causes a huge lag(~2s for
> shipping model) every time. I even tried without broadcast variables but it
> takes same time to ship variables. Can some other techniques be applied to
> improve its performance?
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
