You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@spark.apache.org by Enrico Minack <ma...@Enrico.Minack.dev> on 2019/11/07 08:44:42 UTC
[DISCUSS] Expensive deterministic UDFs
Hi all,
Running expensive deterministic UDFs that return complex types, followed
by multiple references to those results cause Spark to evaluate the UDF
multiple times per row. This has been reported and discussed before:
SPARK-18748 SPARK-17728
val f: Int => Array[Int]
val udfF = udf(f)
df
.select($"id", udfF($"id").as("array"))
.select($"array"(0).as("array0"), $"array"(1).as("array1"))
A common approach to make Spark evaluate the UDF only once is to cache
the intermediate result right after projecting the UDF:
df
.select($"id", udfF($"id").as("array"))
.cache()
.select($"array"(0).as("array0"), $"array"(1).as("array1"))
There are scenarios where this intermediate result is too big for the
cluster to cache. Also this is bad design.
The best approach is to mark the UDF as non-deterministic. Then Spark
optimizes the query in a way that the UDF gets called only once per row,
exactly what you want.
val udfF = udf(f).asNondeterministic()
*However, stating a UDF is non-deterministic though it clearly is
deterministic is counter-intuitive and makes your code harder to read.*
Spark should provide a better way to flag the UDF. Calling it expensive
would be a better naming here.
val udfF = udf(f).asExpensive()
I understand that deterministic is a notion that Expression provides,
and there is no equivalent to expensive that is understood by the
optimizer. However, that asExpensive() could just set the
ScalaUDF.udfDeterministic =deterministic &&!expensive, which implements
the best available approach behind a better naming.
What are your thoughts on asExpensive()?
Regards,
Enrico
Re: [DISCUSS] Expensive deterministic UDFs
Posted by Guy Khazma <Gu...@ibm.com>.
Hi All,
For the use case where the expensive UDF has constant inputs (literals) we
have proposed the following JIRA and PR which calculates the UDF only once
in the driver:
https://issues.apache.org/jira/browse/SPARK-27692
https://github.com/apache/spark/pull/24593
If considering revisiting the optimization for UDFs the above PR can be
considered as well.
Thanks,
Guy
--
Sent from: http://apache-spark-developers-list.1001551.n3.nabble.com/
---------------------------------------------------------------------
To unsubscribe e-mail: dev-unsubscribe@spark.apache.org
Re: [DISCUSS] Expensive deterministic UDFs
Posted by Enrico Minack <ma...@Enrico.Minack.dev>.
I agree that 'non-deterministic' is the right term for what it currently
does: mark an expression as non-deterministic (returns different values
for the same input, e.g. rand()), and the optimizer does its best to not
break semantics when moving expressions around.
In case of expensive deterministic UDFs, or any expensive deterministic
expression, the optimizer should not multiply effort. Even in case of
cheap expressions like a * 5, where performance impact is comparably
small, it simply should not execute things twice. So this is not about
expensive deterministic expressions but deterministic expressions that
get referenced multiple times.
Pushing those expressions into other expressions that reference them is
useful in order to simplify those other expressions, e.g.
df.withColumn("b", not($"a")).where(not($"b")) will eliminate the double
negation of a.
So if expressions are referenced multiple times, they should not be
collapsed, unless referencing expressions get simplified. And then the
simplification must pay off for evaluating the referenced expression
twice. This needs some kind of cost-model, or at least heuristics.
In case of UDFs, I think they should never be collapsed because they
cannot be used to simplify other expressions (can they?). They should
rather be materialised as close to the first reference as possible. If
executing the UDF and referencing it multiple times happens in the same
stage, hence the same generated code, we end up with the perfect
situation where that materialisation of the result per call is hold in
memory and processed by all referencing expressions.
Marking UDFs as expensive is not the right approach here, I agree, they
should simply not be executed multiple times.
Enrico
Am 08.11.19 um 04:26 schrieb Wenchen Fan:
> We really need some documents to define what non-deterministic means.
> AFAIK, non-deterministic expressions may produce a different result
> for the same input row, if the already processed input rows are
> different.
>
> The optimizer tries its best to not change the input sequence
> of non-deterministic expressions. For example, `df.select(...,
> nonDeterministicExpr).filter...` can't do filter pushdown. An
> exception is filter condition. For `df.filter(nonDeterministic &&
> cond)`, Spark still pushes down `cond` even if it may change the input
> sequence of the first condition. This is to respect the SQL semantic
> that filter conditions ANDed together are order-insensitive. Users
> should write `df.filter(nonDeterministic).filter(cond)` to guarantee
> the order.
>
> For this particular problem, I think it's not only about UDF, but a
> general problem with how Spark collapses Projects.
> For example, `df.select('a * 5 as 'b).select('b + 2, 'b + 3)`, Spark
> optimizes it to `df.select('a * 5 + 2, 'a * 5 + 3)`, and execute 'a *
> 5 twice.
>
> I think we should revisit this optimization and think about when we
> can collapse.
>
> On Thu, Nov 7, 2019 at 6:20 PM Rubén Berenguel <rberenguel@gmail.com
> <ma...@gmail.com>> wrote:
>
> That was very interesting, thanks Enrico.
>
> Sean, IIRC it also prevents push down of the UDF in Catalyst in
> some cases.
>
> Regards,
>
> Ruben
>
> > On 7 Nov 2019, at 11:09, Sean Owen <srowen@gmail.com
> <ma...@gmail.com>> wrote:
> >
> > Interesting, what does non-deterministic do except have this effect?
> > aside from the naming, it could be a fine use of this flag if that's
> > all it effectively does. I'm not sure I'd introduce another flag
> with
> > the same semantics just over naming. If anything 'expensive' also
> > isn't the right word, more like 'try not to evaluate multiple
> times'.
> >
> > Why isn't caching the answer? I realize it's big, but you can
> cache to
> > disk. This may be faster than whatever plan reordering has to happen
> > to evaluate once.
> >
> > Usually I'd say, can you redesign your UDF and code to be more
> > efficient too? or use a big a cluster if that's really what you
> need.
> >
> > At first look, no I don't think this Spark-side workaround for
> naming
> > for your use case is worthwhile. There are existing better
> solutions.
> >
> > On Thu, Nov 7, 2019 at 2:45 AM Enrico Minack
> <mail@enrico.minack.dev <ma...@enrico.minack.dev>> wrote:
> >>
> >> Hi all,
> >>
> >> Running expensive deterministic UDFs that return complex types,
> followed by multiple references to those results cause Spark to
> evaluate the UDF multiple times per row. This has been reported
> and discussed before: SPARK-18748 SPARK-17728
> >>
> >> val f: Int => Array[Int]
> >> val udfF = udf(f)
> >> df
> >> .select($"id", udfF($"id").as("array"))
> >> .select($"array"(0).as("array0"), $"array"(1).as("array1"))
> >>
> >> A common approach to make Spark evaluate the UDF only once is
> to cache the intermediate result right after projecting the UDF:
> >>
> >> df
> >> .select($"id", udfF($"id").as("array"))
> >> .cache()
> >> .select($"array"(0).as("array0"), $"array"(1).as("array1"))
> >>
> >> There are scenarios where this intermediate result is too big
> for the cluster to cache. Also this is bad design.
> >>
> >> The best approach is to mark the UDF as non-deterministic. Then
> Spark optimizes the query in a way that the UDF gets called only
> once per row, exactly what you want.
> >>
> >> val udfF = udf(f).asNondeterministic()
> >>
> >> However, stating a UDF is non-deterministic though it clearly
> is deterministic is counter-intuitive and makes your code harder
> to read.
> >>
> >> Spark should provide a better way to flag the UDF. Calling it
> expensive would be a better naming here.
> >>
> >> val udfF = udf(f).asExpensive()
> >>
> >> I understand that deterministic is a notion that Expression
> provides, and there is no equivalent to expensive that is
> understood by the optimizer. However, that asExpensive() could
> just set the ScalaUDF.udfDeterministic = deterministic &&
> !expensive, which implements the best available approach behind a
> better naming.
> >>
> >> What are your thoughts on asExpensive()?
> >>
> >> Regards,
> >> Enrico
> >
> >
> ---------------------------------------------------------------------
> > To unsubscribe e-mail: dev-unsubscribe@spark.apache.org
> <ma...@spark.apache.org>
> >
>
>
> ---------------------------------------------------------------------
> To unsubscribe e-mail: dev-unsubscribe@spark.apache.org
> <ma...@spark.apache.org>
>
Re: [DISCUSS] Expensive deterministic UDFs
Posted by Wenchen Fan <cl...@gmail.com>.
We really need some documents to define what non-deterministic means.
AFAIK, non-deterministic expressions may produce a different result for the
same input row, if the already processed input rows are different.
The optimizer tries its best to not change the input sequence
of non-deterministic expressions. For example, `df.select(...,
nonDeterministicExpr).filter...` can't do filter pushdown. An exception is
filter condition. For `df.filter(nonDeterministic && cond)`, Spark still
pushes down `cond` even if it may change the input sequence of the first
condition. This is to respect the SQL semantic that filter conditions ANDed
together are order-insensitive. Users should write `
df.filter(nonDeterministic).filter(cond)` to guarantee the order.
For this particular problem, I think it's not only about UDF, but a general
problem with how Spark collapses Projects.
For example, `df.select('a * 5 as 'b).select('b + 2, 'b + 3)`, Spark
optimizes it to `df.select('a * 5 + 2, 'a * 5 + 3)`, and execute 'a * 5
twice.
I think we should revisit this optimization and think about when we can
collapse.
On Thu, Nov 7, 2019 at 6:20 PM Rubén Berenguel <rb...@gmail.com> wrote:
> That was very interesting, thanks Enrico.
>
> Sean, IIRC it also prevents push down of the UDF in Catalyst in some cases.
>
> Regards,
>
> Ruben
>
> > On 7 Nov 2019, at 11:09, Sean Owen <sr...@gmail.com> wrote:
> >
> > Interesting, what does non-deterministic do except have this effect?
> > aside from the naming, it could be a fine use of this flag if that's
> > all it effectively does. I'm not sure I'd introduce another flag with
> > the same semantics just over naming. If anything 'expensive' also
> > isn't the right word, more like 'try not to evaluate multiple times'.
> >
> > Why isn't caching the answer? I realize it's big, but you can cache to
> > disk. This may be faster than whatever plan reordering has to happen
> > to evaluate once.
> >
> > Usually I'd say, can you redesign your UDF and code to be more
> > efficient too? or use a big a cluster if that's really what you need.
> >
> > At first look, no I don't think this Spark-side workaround for naming
> > for your use case is worthwhile. There are existing better solutions.
> >
> > On Thu, Nov 7, 2019 at 2:45 AM Enrico Minack <ma...@enrico.minack.dev>
> wrote:
> >>
> >> Hi all,
> >>
> >> Running expensive deterministic UDFs that return complex types,
> followed by multiple references to those results cause Spark to evaluate
> the UDF multiple times per row. This has been reported and discussed
> before: SPARK-18748 SPARK-17728
> >>
> >> val f: Int => Array[Int]
> >> val udfF = udf(f)
> >> df
> >> .select($"id", udfF($"id").as("array"))
> >> .select($"array"(0).as("array0"), $"array"(1).as("array1"))
> >>
> >> A common approach to make Spark evaluate the UDF only once is to cache
> the intermediate result right after projecting the UDF:
> >>
> >> df
> >> .select($"id", udfF($"id").as("array"))
> >> .cache()
> >> .select($"array"(0).as("array0"), $"array"(1).as("array1"))
> >>
> >> There are scenarios where this intermediate result is too big for the
> cluster to cache. Also this is bad design.
> >>
> >> The best approach is to mark the UDF as non-deterministic. Then Spark
> optimizes the query in a way that the UDF gets called only once per row,
> exactly what you want.
> >>
> >> val udfF = udf(f).asNondeterministic()
> >>
> >> However, stating a UDF is non-deterministic though it clearly is
> deterministic is counter-intuitive and makes your code harder to read.
> >>
> >> Spark should provide a better way to flag the UDF. Calling it expensive
> would be a better naming here.
> >>
> >> val udfF = udf(f).asExpensive()
> >>
> >> I understand that deterministic is a notion that Expression provides,
> and there is no equivalent to expensive that is understood by the
> optimizer. However, that asExpensive() could just set the
> ScalaUDF.udfDeterministic = deterministic && !expensive, which implements
> the best available approach behind a better naming.
> >>
> >> What are your thoughts on asExpensive()?
> >>
> >> Regards,
> >> Enrico
> >
> > ---------------------------------------------------------------------
> > To unsubscribe e-mail: dev-unsubscribe@spark.apache.org
> >
>
>
> ---------------------------------------------------------------------
> To unsubscribe e-mail: dev-unsubscribe@spark.apache.org
>
>
Re: [DISCUSS] Expensive deterministic UDFs
Posted by Rubén Berenguel <rb...@gmail.com>.
That was very interesting, thanks Enrico.
Sean, IIRC it also prevents push down of the UDF in Catalyst in some cases.
Regards,
Ruben
> On 7 Nov 2019, at 11:09, Sean Owen <sr...@gmail.com> wrote:
>
> Interesting, what does non-deterministic do except have this effect?
> aside from the naming, it could be a fine use of this flag if that's
> all it effectively does. I'm not sure I'd introduce another flag with
> the same semantics just over naming. If anything 'expensive' also
> isn't the right word, more like 'try not to evaluate multiple times'.
>
> Why isn't caching the answer? I realize it's big, but you can cache to
> disk. This may be faster than whatever plan reordering has to happen
> to evaluate once.
>
> Usually I'd say, can you redesign your UDF and code to be more
> efficient too? or use a big a cluster if that's really what you need.
>
> At first look, no I don't think this Spark-side workaround for naming
> for your use case is worthwhile. There are existing better solutions.
>
> On Thu, Nov 7, 2019 at 2:45 AM Enrico Minack <ma...@enrico.minack.dev> wrote:
>>
>> Hi all,
>>
>> Running expensive deterministic UDFs that return complex types, followed by multiple references to those results cause Spark to evaluate the UDF multiple times per row. This has been reported and discussed before: SPARK-18748 SPARK-17728
>>
>> val f: Int => Array[Int]
>> val udfF = udf(f)
>> df
>> .select($"id", udfF($"id").as("array"))
>> .select($"array"(0).as("array0"), $"array"(1).as("array1"))
>>
>> A common approach to make Spark evaluate the UDF only once is to cache the intermediate result right after projecting the UDF:
>>
>> df
>> .select($"id", udfF($"id").as("array"))
>> .cache()
>> .select($"array"(0).as("array0"), $"array"(1).as("array1"))
>>
>> There are scenarios where this intermediate result is too big for the cluster to cache. Also this is bad design.
>>
>> The best approach is to mark the UDF as non-deterministic. Then Spark optimizes the query in a way that the UDF gets called only once per row, exactly what you want.
>>
>> val udfF = udf(f).asNondeterministic()
>>
>> However, stating a UDF is non-deterministic though it clearly is deterministic is counter-intuitive and makes your code harder to read.
>>
>> Spark should provide a better way to flag the UDF. Calling it expensive would be a better naming here.
>>
>> val udfF = udf(f).asExpensive()
>>
>> I understand that deterministic is a notion that Expression provides, and there is no equivalent to expensive that is understood by the optimizer. However, that asExpensive() could just set the ScalaUDF.udfDeterministic = deterministic && !expensive, which implements the best available approach behind a better naming.
>>
>> What are your thoughts on asExpensive()?
>>
>> Regards,
>> Enrico
>
> ---------------------------------------------------------------------
> To unsubscribe e-mail: dev-unsubscribe@spark.apache.org
>
---------------------------------------------------------------------
To unsubscribe e-mail: dev-unsubscribe@spark.apache.org
Re: [DISCUSS] Expensive deterministic UDFs
Posted by Sean Owen <sr...@gmail.com>.
Interesting, what does non-deterministic do except have this effect?
aside from the naming, it could be a fine use of this flag if that's
all it effectively does. I'm not sure I'd introduce another flag with
the same semantics just over naming. If anything 'expensive' also
isn't the right word, more like 'try not to evaluate multiple times'.
Why isn't caching the answer? I realize it's big, but you can cache to
disk. This may be faster than whatever plan reordering has to happen
to evaluate once.
Usually I'd say, can you redesign your UDF and code to be more
efficient too? or use a big a cluster if that's really what you need.
At first look, no I don't think this Spark-side workaround for naming
for your use case is worthwhile. There are existing better solutions.
On Thu, Nov 7, 2019 at 2:45 AM Enrico Minack <ma...@enrico.minack.dev> wrote:
>
> Hi all,
>
> Running expensive deterministic UDFs that return complex types, followed by multiple references to those results cause Spark to evaluate the UDF multiple times per row. This has been reported and discussed before: SPARK-18748 SPARK-17728
>
> val f: Int => Array[Int]
> val udfF = udf(f)
> df
> .select($"id", udfF($"id").as("array"))
> .select($"array"(0).as("array0"), $"array"(1).as("array1"))
>
> A common approach to make Spark evaluate the UDF only once is to cache the intermediate result right after projecting the UDF:
>
> df
> .select($"id", udfF($"id").as("array"))
> .cache()
> .select($"array"(0).as("array0"), $"array"(1).as("array1"))
>
> There are scenarios where this intermediate result is too big for the cluster to cache. Also this is bad design.
>
> The best approach is to mark the UDF as non-deterministic. Then Spark optimizes the query in a way that the UDF gets called only once per row, exactly what you want.
>
> val udfF = udf(f).asNondeterministic()
>
> However, stating a UDF is non-deterministic though it clearly is deterministic is counter-intuitive and makes your code harder to read.
>
> Spark should provide a better way to flag the UDF. Calling it expensive would be a better naming here.
>
> val udfF = udf(f).asExpensive()
>
> I understand that deterministic is a notion that Expression provides, and there is no equivalent to expensive that is understood by the optimizer. However, that asExpensive() could just set the ScalaUDF.udfDeterministic = deterministic && !expensive, which implements the best available approach behind a better naming.
>
> What are your thoughts on asExpensive()?
>
> Regards,
> Enrico
---------------------------------------------------------------------
To unsubscribe e-mail: dev-unsubscribe@spark.apache.org