Pyspark DataFrame - How to use variables to make join?
<p>I'm having a bit of trouble to make a join on two Data Frames using Spark Data Frames on python. I have two data frames that I had to change the name of the columns in order to make them unique for each data frame, so later I could tell which column is which. I did this to rename the columns (firstDf and secondDf are Spark DataFrames created using the function createDataFrame):</p>

<pre><code>oldColumns = firstDf.schema.names
newColumns = list(map(lambda x: "{}.{}".format('firstDf', x), oldColumns))
firstDf = firstDf.toDF(*newColumns)
</code></pre>

<p>I repeated this for the second DataFrame. Then I tried to join them, using the following code:</p>

<pre><code>from pyspark.sql.functions import *

firstColumn = 'firstDf.firstColumn'
secondColumn = 'secondDf.firstColumn'
joinedDF = firstDf.join(secondDf, col(firstColumn) == col(secondColumn), 'inner')
</code></pre>

<p>Using it like this I get the following error:</p>

<blockquote>
  <p>AnalysisException "cannot resolve 'firstDf.firstColumn' given input columns: [firstDf.firstColumn, ...];"</p>
</blockquote>

<p>This was only to illustrate that the column exists in the input columns array.</p>

<p>If I don't rename the DataFrames columns I'm able to join them using this piece of code:</p>

<pre><code>joinedDf = firstDf.join(secondDf, firstDf.firstColumn == secondDf.firstColumn, 'inner')
</code></pre>

<p>But this give me a DataFrame with ambiguous column names.</p>

<p>Any ideas on how to approach this?</p>

<p>Generally speaking don't use dots in names. These have special meaning (can be used either to determine the table or to access <code>struct</code> fields) and require some additional work to be correctly recognized.</p>

<p>For equi joins all you need is a column name:</p>

<pre><code>from pyspark.sql.functions import col

firstDf = spark.createDataFrame([(1, "foo")], ("firstColumn", "secondColumn"))
secondDf = spark.createDataFrame([(1, "foo")], ("firstColumn", "secondColumn"))

column = 'firstColumn'
firstDf.join(secondDf, [column], 'inner')

## DataFrame[firstColumn: bigint, secondColumn: string, secondColumn: string]
</code></pre>

<p>For complex cases use table aliases:</p>

<pre><code>firstColumn = 'firstDf.firstColumn'
secondColumn = 'secondDf.firstColumn'

firstDf.alias("firstDf").join(
    secondDf.alias("secondDf"),
    # After alias prefix resolves to table name
    col(firstColumn) == col(secondColumn),
   "inner"
)

## DataFrame[firstColumn: bigint, secondColumn: string, firstColumn: bigint, secondColumn: string]
</code></pre>

<p>You could also use parent frames directly:</p>

<pre><code>column = 'firstColumn'

firstDf.join(secondDf, firstDf[column] == secondDf[column])
</code></pre>

python apache-spark pyspark spark-dataframe pyspark-sql