title: Hashing multiple columns of spark dataframe
body: <p>I need to hash specific columns of spark dataframe. Some columns have specific datatype which are basically the extensions of standard spark's DataType class. The problem is because for some reason in <em>when</em> case some conditions don't work as expected.</p>
  
  <p>As a hash table I have a map. Let's call it tableConfig:</p>
  
  <pre class="lang-scala prettyprint-override"><code>val tableConfig = Map("a" -&gt; "KEEP", "b" -&gt; "HASH", "c" -&gt; "KEEP", "d" -&gt; "HASH", "e" -&gt; "KEEP")
  </code></pre>
  
  <p>The salt variable is used to concatenate with column:</p>
  
  <pre class="lang-scala prettyprint-override"><code>val salt = "abc"
  </code></pre>
  
  <p>The function for hashing looks like this:</p>
  
  <pre class="lang-scala prettyprint-override"><code>def hashColumns(tableConfig: Map[String, String], salt: String, df: DataFrame): DataFrame = {
  
      val removedColumns = tableConfig.filter(_._2 == "REMOVE").keys.toList
      val hashedColumns = tableConfig.filter(_._2 == "HASH").keys.toList
  
      val cleanedDF = df.drop(removedColumns: _ *)
  
      val colTypes = cleanedDF.dtypes.toMap
  
      def typeFromString(s: String): DataType = s match {
        case "StringType" =&gt; StringType
        case "BooleanType" =&gt; BooleanType
        case "IntegerType" =&gt; IntegerType
        case "DateType" =&gt; DateType
        case "ShortType" =&gt; ShortType
        case "DecimalType(15,7)" =&gt; DecimalType(15,7)
        case "DecimalType(18,2)" =&gt; DecimalType(18,2)
        case "DecimalType(11,7)" =&gt; DecimalType(11,7)
        case "DecimalType(17,2)" =&gt; DecimalType(17,2)
        case "DecimalType(38,2)" =&gt; DecimalType(38,2)
        case _ =&gt; throw new TypeNotPresentException(
          "Please check types in the dataframe. The following column type is missing: ".concat(s), null
        )
      }
  
      val getType = colTypes.map{case (k, _) =&gt; (k, typeFromString(colTypes(k)))}
  
      val hashedDF = cleanedDF.columns.foldLeft(cleanedDF) {
        (memoDF, colName) =&gt;
          memoDF.withColumn(
            colName,
            when(col(colName).isin(hashedColumns: _*) &amp;&amp; col(colName).isNull, null).
              when(col(colName).isin(hashedColumns: _*) &amp;&amp; col(colName).isNotNull,
                sha2(concat(col(colName), lit(salt)), 256)).otherwise(col(colName)
              )
          )
      }
  
      hashedDF
    }
  </code></pre>
  
  <p>I am getting error regarding to specific column. Namely the error is the following:</p>
  
  <blockquote>
    <p>org.apache.spark.sql.AnalysisException: cannot resolve '(<code>c</code> IN ('a', 'b', 'd', 'e'))' due to data type mismatch: Arguments must be same type but were: boolean != string;;</p>
  </blockquote>
  
  <p><em>Column names are changed.</em></p>
  
  <p>My search didn't give any clear explanation why isin or isNull functions don't work as expected. Also I follow specific style of implementation and want to avoid the following approaches:</p>
  
  <p>1) No UDFs. They are painful for me.</p>
  
  <p>2) No for loops over spark dataframe columns. The data could contain more than billion samples and it's going to be a headache in terms of performance.</p>