How to perform "Lookup" operation on Spark dataframes given multiple conditions
<p>I'm a newbie on Spark (My version is 1.6.0) and now I'm trying to solve the problem given below:</p>
<p>Suppose there are two source files:</p>
<ul>
<li>The first one (A for short) is a big one which contains columns named A1, B1, C1 and other 80 columns. There are 230K records inside.</li>
<li>The second one (B for short) is a small lookup table which contains columns named A2, B2, C2 and D2. There are 250 records inside.</li>
</ul>
<p>Now we need to insert a new column into A, given the logic below:</p>
<ul>
<li>First lookup A1, B1 and C1 in B (corresponding columns are A2, B2 and C2), if succeeds, return D2 as the value of the new added column. If nothing found...</li>
<li>Then lookup A1, B1 in B. If successful, return D2. If nothing found...</li>
<li>Set the default value &quot;NA&quot;</li>
</ul>
<p>I have already read in the files and converted them into data frames. For the first situation, I got the result by left outer joining them together. But I cannot find good way in the next step.</p>
<p>My current trying is to build a new data frame by joining A and B using a less strict condition. However I've no clue how to update the current data frame from the other one. Or is there any other more intuitive and efficient way to tackle the whole problem?</p>
<p>Thanks for all the answers.</p>
<p>-----------------------------Update on 20160309--------------------------------</p>
<p>Finally accepted @mlk 's answer. Still great thanks to @zero323 for his/her great comments on UDF versus join, the Tungsten code generation is really another problem we are facing now. But since we need to do scores of lookup and average 4 conditions for every lookup, the former solution is more suitable...</p>
<p>The final solution is somehow looks like below snippet:</p>
<pre><code>```
import sqlContext.implicits._
import com.github.marklister.collections.io._

case class TableType(A: String, B: String, C: String, D: String)
val tableBroadcast = sparkContext.broadcast(CsvParser(TableType).parseFile(&quot;...&quot;))
val lkupD = udf {
  (aStr: String, bStr: String, cStr: String) =&gt;
    tableBroadcast.value.find {
      case TableType(a, b, c, _) =&gt;
        (a == aStr &amp;&amp; b == bStr &amp;&amp; c == cStr) ||
        (a == aStr &amp;&amp; b == bStr)
    }.getOrElse(TableType(&quot;&quot;, &quot;&quot;, &quot;&quot;, &quot;NA&quot;)).D
}
df = df.withColumn(&quot;NEW_COL&quot;, lkupD($&quot;A&quot;, $&quot;B&quot;, $&quot;C&quot;))
```
</code></pre>