udpate

streamingpro-mlsql/src/main/java/streaming/core/compositor/spark/udf/FVectors.scala

@@ -70,4 +70,15 @@ object FVectors {
     }.unzip
     new SparseVector(selectedIndices.length, sliceInds.toArray, sliceVals.toArray)
   }
+
+  def range(vector: SparseVector, range: Array[Int]): SparseVector = {
+    val start = range(0)
+    val end = range(1)
+
+    val (sliceVals, sliceInds) = vector.values.zip(vector.indices).filter { d =>
+      val (v, i) = d
+      i < end && i >= start
+    }.unzip
+    new SparseVector(end - start, sliceInds.toArray, sliceVals.toArray)
+  }
 }

streamingpro-mlsql/src/main/java/streaming/core/compositor/spark/udf/Functions.scala

@@ -119,6 +119,17 @@ object Functions {
     })
   }
 
+  def vec_range(uDFRegistration: UDFRegistration) = {
+    uDFRegistration.register("vec_range", (vec: Vector, inds: Seq[Int]) => {
+      assert(inds.size == 2)
+      vec match {
+        case features: DenseVector => Vectors.dense(features.toArray.slice(inds(0), inds(1)))
+        case features: SparseVector => FVectors.range(features, inds.toArray)
+      }
+    })
+  }
+
+
   /*
     1 - x.dot(y)/(x.norm(2)*y.norm(2))
    */

streamingpro-mlsql/src/main/java/streaming/dsl/IncludeAdaptor.scala

@@ -67,6 +67,7 @@ object IncludeAdaptor {
   val mapping = Map[String, String](
     "hdfs" -> "streaming.dsl.mmlib.algs.includes.HDFSIncludeSource",
     "http" -> "streaming.dsl.mmlib.algs.includes.HTTPIncludeSource",
+    "store" -> "tech.mlsql.dsl.includes.StoreIncludeSource",
 
     "function" -> "streaming.dsl.mmlib.algs.includes.analyst.HttpBaseDirIncludeSource",
     "view" -> "streaming.dsl.mmlib.algs.includes.analyst.HttpBaseDirIncludeSource",

streamingpro-mlsql/src/main/java/streaming/dsl/mmlib/algs/feature/StringFeature.scala

@@ -198,7 +198,7 @@ object StringFeature extends BaseFeatureFunctions with Logging with WowLog {
     val spark = df.sparkSession
 
     //tfidf feature
-    val tfIDFconfig = Map("inputCol" -> inputCol, "numFeatures" -> wordCount.toString, "binary" -> "true")
+    val tfIDFconfig = Map("inputCol" -> inputCol, "numFeatures" -> wordCount.toString, "binary" -> "false")
     logInfo(format(s"[TFIDF] run tf/idf estimator with follow configuration ${tfIDFconfig.map(s => s"${s._1}->${s._2}").mkString(" ; ")} "))
     val tfidf = new SQLTfIdf()
     tfidf.train(newDF, TF_IDF_PATH(metaPath, inputCol), tfIDFconfig)

streamingpro-mlsql/src/main/java/tech/mlsql/dsl/includes/StoreIncludeSource.scala

@@ -0,0 +1,31 @@
+package tech.mlsql.dsl.includes
+
+import org.apache.http.client.fluent.Request
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.mlsql.session.MLSQLException
+import streaming.common.PathFun
+import streaming.dsl.IncludeSource
+import streaming.log.Logging
+
+/**
+  * 2019-05-12 WilliamZhu([email protected])
+  */
+class StoreIncludeSource extends IncludeSource with Logging {
+  override def fetchSource(sparkSession: SparkSession, path: String, options: Map[String, String]): String = {
+
+    val res = Request.Get(PathFun("http://repo.store.mlsql.tech").add(path).toPath)
+      .connectTimeout(60 * 1000)
+      .socketTimeout(10 * 60 * 1000)
+      .execute().returnContent().asString()
+
+    if (res == null) {
+      throw new MLSQLException(
+        s"""
+           |${path} is not found
+         """.stripMargin)
+    }
+    res
+  }
+
+  override def skipPathPrefix: Boolean = true
+}