There are two distinct parts here. Optimisation + execution.
Spark does not have a Cost Based Optimizer (CBO) yet but that does not
matter for now.
When we do such operation say outer join between (s) and (t) DFs below, we
see
scala> val rs = s.join(t,s("time_id")===t("time_id"), "fullouter")
rs: org.apache.spark.sql.DataFrame = [AMOUNT_SOLD: decimal(10,0), TIME_ID:
timestamp ... 3 more fields]
But the optimizer knows the access path at the dataframe level itself
before converting DF to rdd
scala> val rs = s.join(t,s("time_id")===t("time_id"), "fullouter").
*explain*
== Physical Plan ==
SortMergeJoin [time_id#5], [time_id#40], FullOuter
:- *Sort [time_id#5 ASC], false, 0
: +- Exchange hashpartitioning(time_id#5, 200)
: +- HiveTableScan [AMOUNT_SOLD#9, TIME_ID#5, CHANNEL_ID#6L],
MetastoreRelation oraclehadoop, sales
+- *Sort [time_id#40 ASC], false, 0
+- Exchange hashpartitioning(time_id#40, 200)
+- HiveTableScan [TIME_ID#40, CALENDAR_MONTH_DESC#50],
MetastoreRelation oraclehadoop, times
rs: Unit = ()
scala> val rs = s.join(t,s("time_id")===t("time_id"), "fullouter").rdd
rs: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] =
MapPartitionsRDD[1310] at rdd at <console>:27
HTH
Dr Mich Talebzadeh
LinkedIn *
https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw
<https://www.linkedin.com/profile/view?id=AAEAAAAWh2gBxianrbJd6zP6AcPCCdOABUrV8Pw>*
http://talebzadehmich.wordpress.com
*Disclaimer:* Use it at your own risk. Any and all responsibility for any
loss, damage or destruction of data or any other property which may arise
from relying on this email's technical content is explicitly disclaimed.
The author will in no case be liable for any monetary damages arising from
such loss, damage or destruction.
On 14 August 2016 at 09:56, ayan guha <[email protected]> wrote:
> I do not think so. What I understand Spark will still use Catalyst to
> join. DF always has an RDD underneath, but that does not mean any action
> will force less optimal path.
>
> On Sun, Aug 14, 2016 at 3:04 PM, mayur bhole <[email protected]>
> wrote:
>
>> HI All,
>>
>> Lets say, we have
>>
>> val df = bigTableA.join(bigTableB,bigTableA("A")===bigTableB("A"),"left")
>> val rddFromDF = df.rdd
>> println(rddFromDF.count)
>>
>> My understanding is that spark will convert all data frame operations
>> before "rddFromDF.count" into RDD equivalent operation as we are not
>> performing any action on dataframe directly. In that case, spark will not
>> be using optimization engine. Is my assumption right? Please point me to
>> right resources.
>>
>> [ Note : I have posted same question on so : http://stackoverflow.com/que
>> stions/38889812/how-spark-dataframe-optimization-engine-works-with-dag ]
>>
>> Thanks
>>
>
>
>
> --
> Best Regards,
> Ayan Guha
>
