//设置package包名称以及导入依赖的类
package spark.feature

import org.apache.spark.ml.Transformer
import org.apache.spark.ml.attribute.AttributeGroup
import org.apache.spark.ml.param.{ParamMap, _}
import org.apache.spark.ml.util._
import org.apache.spark.sql.functions.{col, udf}
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.{DataFrame, UserDefinedFunction}


  def setFunction(value: String=>Double) = set(function, value)
  def getFunction() =  $(function)

  override def transform(dataset: DataFrame): DataFrame = {
    val outputSchema = transformSchema(dataset.schema)
    val metadata = outputSchema($(outputCol)).metadata
    val dummy = udf { x: Any => $(expr) }
    var data = dataset.select(col("*"), dummy(col($(inputCols).head)).as("0"))
    val substitute: (String => ((String, Double) => String)) = name => (exp, elem) => exp.replace(name, elem.toString)
    def subst(v: String) = udf(substitute(v))
    $(inputCols).view.zipWithIndex foreach { case (v, i) => data = data.select(col("*"), subst(v)(data(i.toString), data(v)).as((i + 1).toString)).drop(i.toString) }
    val eval = udf($(function))
    data.select(col("*"), eval(data($(inputCols).length.toString)).as($(outputCol), metadata)).drop($(inputCols).length.toString)
  }


  override def transformSchema(schema: StructType): StructType = {
    // TODO: Assertions on inputCols
    val attrGroup = new AttributeGroup($(outputCol), $(numFeatures))
    val col = attrGroup.toStructField()
    require(!schema.fieldNames.contains(col.name), s"Column ${col.name} already exists.")
    StructType(schema.fields :+ col)
  }

  override def copy(extra: ParamMap): FeatureFuTransformer = defaultCopy(extra)
} 

//设置package包名称以及导入依赖的类
package spark.progress

import org.apache.spark.ml.Transformer
import org.apache.spark.ml.attribute.AttributeGroup
import org.apache.spark.ml.param.{ParamMap, _}
import org.apache.spark.ml.util._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.{DataFrame, Row}


	def getOutputCol() = $(outputcol)

	def getExpr() = $(expr)

	def getInputCols() = $(inputcols).toArray

	def getNumFeatures() = $(numFeatures)

	def getFunction() = $(function)

	override def transform(dataset: DataFrame): DataFrame = {
		val outputSchema = transformSchema(dataset.schema)
		val metadata = outputSchema($(outputcol)).metadata
		val f = udf {(r: Row) => {
			val exp = $(expr)
			for (i <- 1 to $(numFeatures)) {
				exp.replace(dataset.columns.toSeq(i), r.getInt(i).toString)
			}
			$(function)(exp)
		}}
		val x = lit($(expr))
		dataset.select(col("*"), f(struct(dataset.columns.map(dataset(_)) : _*)).as($(outputcol), metadata))
	}


	override def transformSchema(schema: StructType): StructType = {
		val attrGroup = new AttributeGroup($(outputcol), $(numFeatures))
		val col = attrGroup.toStructField()
		require(!schema.fieldNames.contains(col.name), s"Column ${col.name} already exists.")
		StructType(schema.fields :+ col)
	}

	override def copy(extra: ParamMap): ExpressionEval = defaultCopy(extra)
}