MapDF.scala

import clojure.lang.IFn
import org.apache.spark.sql.Row
import clojure.java.api.Clojure
import org.apache.spark.sql.types.StructType

def map_clojure(df: DataFrame, source: String, schema: StructType) = {
    var clojure_compiled: IFn = null
    val rdd = df.map({row => 
        if (clojure_compiled == null) {
            clojure_compiled = Clojure.`var`("clojure.core", "load-string").invoke(source).asInstanceOf[IFn]
        }
        Row.fromSeq(clojure_compiled.invoke(row.toSeq.toArray).asInstanceOf[Array[Any]])
    })
    sqlContext.createDataFrame(rdd, schema)
}