You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@s2graph.apache.org by st...@apache.org on 2018/12/06 12:49:50 UTC
[06/13] incubator-s2graph git commit: Implement and add test cases on
below. - WalLogAggregateProcess - BuildTopFeaturesProcess -
FilterTopFeaturesProcess
Implement and add test cases on below.
- WalLogAggregateProcess
- BuildTopFeaturesProcess
- FilterTopFeaturesProcess
Project: http://git-wip-us.apache.org/repos/asf/incubator-s2graph/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-s2graph/commit/651ca2cf
Tree: http://git-wip-us.apache.org/repos/asf/incubator-s2graph/tree/651ca2cf
Diff: http://git-wip-us.apache.org/repos/asf/incubator-s2graph/diff/651ca2cf
Branch: refs/heads/master
Commit: 651ca2cf9fad57e4613e22dde39c65fe88755a5c
Parents: d9af2b8
Author: DO YUNG YOON <st...@apache.org>
Authored: Wed Aug 22 19:15:48 2018 +0900
Committer: DO YUNG YOON <st...@apache.org>
Committed: Thu Aug 23 00:52:24 2018 +0900
----------------------------------------------------------------------
.../org/apache/s2graph/s2jobs/task/Task.scala | 16 +++
.../org/apache/s2graph/s2jobs/wal/WalLog.scala | 60 ++++++---
.../wal/process/BuildTopFeaturesProcess.scala | 126 +++++++++++++++++++
.../wal/process/FeatureIndexProcess.scala | 99 ---------------
.../wal/process/FilterTopFeaturesProcess.scala | 90 +++++++++++++
.../wal/process/FilterWalLogAggProcess.scala | 75 -----------
.../wal/process/WalLogAggregateProcess.scala | 20 +--
.../wal/process/params/AggregateParam.scala | 19 ++-
.../process/params/BuildTopFeaturesParam.scala | 17 +++
.../wal/process/params/FeatureIndexParam.scala | 15 ---
.../process/params/FilterTopFeaturesParam.scala | 7 ++
.../process/params/FilterWalLogAggParam.scala | 7 --
.../wal/transformer/DefaultTransformer.scala | 4 +-
.../s2jobs/wal/transformer/ExtractDomain.scala | 11 +-
.../wal/transformer/ExtractServiceName.scala | 13 +-
.../s2jobs/wal/transformer/Transformer.scala | 3 +-
.../s2graph/s2jobs/wal/udfs/WalLogUDF.scala | 26 ++--
.../apache/s2graph/s2jobs/wal/TestData.scala | 42 +++++++
.../s2graph/s2jobs/wal/TransformerTest.scala | 10 +-
.../process/BuildTopFeaturesProcessTest.scala | 31 +++++
.../process/FilterTopFeaturesProcessTest.scala | 84 +++++++++++++
.../process/WalLogAggregateProcessTest.scala | 64 ++--------
.../s2jobs/wal/udafs/WalLogUDAFTest.scala | 42 +++++++
23 files changed, 574 insertions(+), 307 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/task/Task.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/task/Task.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/task/Task.scala
index ab02900..62081df 100644
--- a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/task/Task.scala
+++ b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/task/Task.scala
@@ -21,9 +21,13 @@ package org.apache.s2graph.s2jobs.task
import org.apache.s2graph.core.S2GraphConfigs
import org.apache.s2graph.s2jobs.Logger
+import org.apache.s2graph.s2jobs.wal.transformer.Transformer
+import play.api.libs.json.Json
//import org.apache.s2graph.s2jobs.loader.GraphFileOptions
object TaskConf {
+ val Empty = new TaskConf(name = "empty", `type` = "empty", inputs = Nil, options = Map.empty)
+
// def toGraphFileOptions(taskConf: TaskConf): GraphFileOptions = {
// val args = taskConf.options.filterKeys(GraphFileOptions.OptionKeys)
// .flatMap(kv => Seq(kv._1, kv._2)).toSeq.toArray
@@ -42,6 +46,18 @@ object TaskConf {
def parseLocalCacheConfigs(taskConf: TaskConf): Map[String, Any] = {
taskConf.options.filterKeys(S2GraphConfigs.CacheConfigs.DEFAULTS.keySet).mapValues(_.toInt)
}
+
+ def parseTransformers(taskConf: TaskConf): Seq[Transformer] = {
+ val classes = Json.parse(taskConf.options.getOrElse("transformClasses",
+ """["org.apache.s2graph.s2jobs.wal.transformer.DefaultTransformer"]""")).as[Seq[String]]
+
+ classes.map { className =>
+ Class.forName(className)
+ .getConstructor(classOf[TaskConf])
+ .newInstance(taskConf)
+ .asInstanceOf[Transformer]
+ }
+ }
}
case class TaskConf(name:String, `type`:String, inputs:Seq[String] = Nil, options:Map[String, String] = Map.empty, cache:Option[Boolean]=None)
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/WalLog.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/WalLog.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/WalLog.scala
index e4c3467..9617ca1 100644
--- a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/WalLog.scala
+++ b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/WalLog.scala
@@ -3,6 +3,7 @@ package org.apache.s2graph.s2jobs.wal
import com.google.common.hash.Hashing
import org.apache.s2graph.core.JSONParser
import org.apache.s2graph.s2jobs.wal.process.params.AggregateParam
+import org.apache.s2graph.s2jobs.wal.transformer.Transformer
import org.apache.s2graph.s2jobs.wal.utils.BoundedPriorityQueue
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{LongType, StringType, StructField, StructType}
@@ -17,22 +18,12 @@ object WalLogAgg {
new WalLogAgg(walLog.from, Seq(walLog), walLog.timestamp, walLog.timestamp)
}
+ def toFeatureHash(dimVal: DimVal): Long = toFeatureHash(dimVal.dim, dimVal.value)
+
def toFeatureHash(dim: String, value: String): Long = {
Hashing.murmur3_128().hashBytes(s"$dim:$value".getBytes("UTF-8")).asLong()
}
- def filter(walLogAgg: WalLogAgg, validFeatureHashKeys: Set[Long]) = {
- val filtered = walLogAgg.logs.map { walLog =>
- val fields = Json.parse(walLog.props).as[JsObject].fields.filter { case (dim, jsValue) =>
- validFeatureHashKeys(toFeatureHash(dim, JSONParser.jsValueToString(jsValue)))
- }
-
- walLog.copy(props = Json.toJson(fields).as[JsObject].toString)
- }
-
- walLogAgg.copy(logs = filtered)
- }
-
def merge(iter: Iterator[WalLogAgg],
param: AggregateParam)(implicit ord: Ordering[WalLog]) = {
val heap = new BoundedPriorityQueue[WalLog](param.heapSize)
@@ -51,6 +42,47 @@ object WalLogAgg {
WalLogAgg(topItems.head.from, topItems, maxTs, minTs)
}
+
+ def filterProps(walLogAgg: WalLogAgg,
+ transformers: Seq[Transformer],
+ validFeatureHashKeys: Set[Long]) = {
+ val filtered = walLogAgg.logs.map { walLog =>
+ val fields = walLog.propsJson.fields.filter { case (propKey, propValue) =>
+ val filtered = transformers.flatMap { transformer =>
+ transformer.toDimValLs(walLog, propKey, JSONParser.jsValueToString(propValue)).filter(dimVal => validFeatureHashKeys(toFeatureHash(dimVal)))
+ }
+ filtered.nonEmpty
+ }
+
+ walLog.copy(props = Json.toJson(fields.toMap).as[JsObject].toString)
+ }
+
+ walLogAgg.copy(logs = filtered)
+ }
+}
+
+object DimValCountRank {
+ def fromRow(row: Row): DimValCountRank = {
+ val dim = row.getAs[String]("dim")
+ val value = row.getAs[String]("value")
+ val count = row.getAs[Long]("count")
+ val rank = row.getAs[Long]("rank")
+
+ new DimValCountRank(DimVal(dim, value), count, rank)
+ }
+}
+
+case class DimValCountRank(dimVal: DimVal, count: Long, rank: Long)
+
+case class DimValCount(dimVal: DimVal, count: Long)
+
+object DimVal {
+ def fromRow(row: Row): DimVal = {
+ val dim = row.getAs[String]("dim")
+ val value = row.getAs[String]("value")
+
+ new DimVal(dim, value)
+ }
}
case class DimVal(dim: String, value: String)
@@ -71,8 +103,8 @@ case class WalLog(timestamp: Long,
val id = from
val columnName = label
val serviceName = to
-
- lazy val propsKeyValues = Json.parse(props).as[JsObject].fields.map { case (key, jsValue) =>
+ lazy val propsJson = Json.parse(props).as[JsObject]
+ lazy val propsKeyValues = propsJson.fields.map { case (key, jsValue) =>
key -> JSONParser.jsValueToString(jsValue)
}
}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/BuildTopFeaturesProcess.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/BuildTopFeaturesProcess.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/BuildTopFeaturesProcess.scala
new file mode 100644
index 0000000..048b470
--- /dev/null
+++ b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/BuildTopFeaturesProcess.scala
@@ -0,0 +1,126 @@
+package org.apache.s2graph.s2jobs.wal.process
+
+import org.apache.s2graph.s2jobs.task.TaskConf
+import org.apache.s2graph.s2jobs.wal.WalLogAgg.toFeatureHash
+import org.apache.s2graph.s2jobs.wal.process.params.BuildTopFeaturesParam
+import org.apache.s2graph.s2jobs.wal.transformer._
+import org.apache.s2graph.s2jobs.wal.udfs.WalLogUDF
+import org.apache.s2graph.s2jobs.wal.{DimVal, DimValCount, WalLog, WalLogAgg}
+import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql._
+import org.apache.spark.sql.functions._
+import play.api.libs.json.{JsObject, Json}
+
+import scala.collection.mutable
+
+object BuildTopFeaturesProcess {
+ def extractDimValuesWithCount(transformers: Seq[Transformer]) = {
+ udf((rows: Seq[Row]) => {
+ val logs = rows.map(WalLog.fromRow)
+ val dimValCounts = mutable.Map.empty[DimVal, Int]
+
+ logs.foreach { walLog =>
+ walLog.propsKeyValues.foreach { case (propsKey, propsValue) =>
+ transformers.foreach { transformer =>
+ transformer.toDimValLs(walLog, propsKey, propsValue).foreach { dimVal =>
+ val newCount = dimValCounts.getOrElse(dimVal, 0) + 1
+ dimValCounts += (dimVal -> newCount)
+ }
+ }
+ }
+ }
+
+ dimValCounts.toSeq.sortBy(-_._2)map { case (dimVal, count) =>
+ DimValCount(dimVal, count)
+ }
+ })
+ }
+
+ def extractDimValues(transformers: Seq[Transformer]) = {
+ udf((rows: Seq[Row]) => {
+ val logs = rows.map(WalLog.fromRow)
+ // TODO: this can be changed into Map to count how many times each dimVal exist in sequence of walLog
+ // then change this to mutable.Map.empty[DimVal, Int], then aggregate.
+ val distinctDimValues = mutable.Set.empty[DimVal]
+
+ logs.foreach { walLog =>
+ walLog.propsKeyValues.foreach { case (propsKey, propsValue) =>
+ transformers.foreach { transformer =>
+ transformer.toDimValLs(walLog, propsKey, propsValue).foreach { dimVal =>
+ distinctDimValues += dimVal
+ }
+ }
+ }
+ }
+
+ distinctDimValues.toSeq
+ })
+ }
+
+ def buildDictionary(ss: SparkSession,
+ allDimVals: DataFrame,
+ param: BuildTopFeaturesParam,
+ dimValColumnName: String = "dimVal"): DataFrame = {
+ import ss.implicits._
+
+ val rawFeatures = allDimVals
+ .select(col(param._countColumnName), col(s"$dimValColumnName.dim").as("dim"), col(s"$dimValColumnName.value").as("value"))
+ .groupBy("dim", "value")
+ .agg(countDistinct(param._countColumnName).as("count"))
+ .filter(s"count > ${param._minUserCount}")
+
+ val ds: Dataset[((String, Long), String)] =
+ rawFeatures.select("dim", "value", "count").as[(String, String, Long)]
+ .map { case (dim, value, uv) =>
+ (dim, uv) -> value
+ }
+
+
+ implicit val ord = Ordering.Tuple2(Ordering.String, Ordering.Long.reverse)
+
+ val rdd: RDD[(Long, (String, Long), String)] = WalLogUDF.appendRank(ds, param.numOfPartitions, param.samplePointsPerPartitionHint)
+
+ rdd.toDF("rank", "dim_count", "value")
+ .withColumn("dim", col("dim_count._1"))
+ .withColumn("count", col("dim_count._2"))
+ .select("dim", "value", "count", "rank")
+ }
+}
+
+case class BuildTopFeaturesProcess(taskConf: TaskConf) extends org.apache.s2graph.s2jobs.task.Process(taskConf) {
+
+ import BuildTopFeaturesProcess._
+
+ override def execute(ss: SparkSession, inputMap: Map[String, DataFrame]): DataFrame = {
+ val countColumnName = taskConf.options.getOrElse("countColumnName", "from")
+ val numOfPartitions = taskConf.options.get("numOfPartitions").map(_.toInt)
+ val samplePointsPerPartitionHint = taskConf.options.get("samplePointsPerPartitionHint").map(_.toInt)
+ val minUserCount = taskConf.options.get("minUserCount").map(_.toLong)
+
+ numOfPartitions.map { d => ss.sqlContext.setConf("spark.sql.shuffle.partitions", d.toString) }
+
+ val param = BuildTopFeaturesParam(minUserCount = minUserCount, countColumnName = Option(countColumnName),
+ numOfPartitions = numOfPartitions, samplePointsPerPartitionHint = samplePointsPerPartitionHint
+ )
+
+ val edges = taskConf.inputs.tail.foldLeft(inputMap(taskConf.inputs.head)) { case (prev, cur) =>
+ prev.union(inputMap(cur))
+ }
+
+ //TODO: user expect to inject transformers that transform (WalLog, propertyKey, propertyValue) to Seq[DimVal].
+ val transformers = TaskConf.parseTransformers(taskConf)
+ val dimValExtractUDF = extractDimValues(transformers)
+ val dimValColumnName = "dimVal"
+
+ val rawFeatures = edges
+ .withColumn(dimValColumnName, explode(dimValExtractUDF(col("logs"))))
+
+ val dict = buildDictionary(ss, rawFeatures, param, dimValColumnName)
+
+ dict
+ }
+
+
+ override def mandatoryOptions: Set[String] = Set.empty
+}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/FeatureIndexProcess.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/FeatureIndexProcess.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/FeatureIndexProcess.scala
deleted file mode 100644
index 073b0cf..0000000
--- a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/FeatureIndexProcess.scala
+++ /dev/null
@@ -1,99 +0,0 @@
-package org.apache.s2graph.s2jobs.wal.process
-
-import org.apache.s2graph.s2jobs.task.TaskConf
-import org.apache.s2graph.s2jobs.wal.process.params.FeatureIndexParam
-import org.apache.s2graph.s2jobs.wal.transformer._
-import org.apache.s2graph.s2jobs.wal.udfs.WalLogUDF
-import org.apache.s2graph.s2jobs.wal.{DimVal, WalLog}
-import org.apache.spark.rdd.RDD
-import org.apache.spark.sql._
-import org.apache.spark.sql.functions._
-
-import scala.collection.mutable
-
-object FeatureIndexProcess {
- def extractDimValues(transformers: Seq[Transformer]) = {
- udf((rows: Seq[Row]) => {
- val logs = rows.map(WalLog.fromRow)
- // TODO: this can be changed into Map to count how many times each dimVal exist in sequence of walLog
- // then change this to mutable.Map.empty[DimVal, Int], then aggregate.
- val distinctDimValues = mutable.Set.empty[DimVal]
-
- logs.foreach { walLog =>
- walLog.propsKeyValues.foreach { case (propsKey, propsValue) =>
- transformers.foreach { transformer =>
- transformer.toDimValLs(walLog, propsKey, propsValue).foreach(distinctDimValues += _)
- }
- }
- }
-
- distinctDimValues.toSeq
- })
- }
-
- def buildDictionary(ss: SparkSession,
- allDimVals: DataFrame,
- param: FeatureIndexParam,
- dimValColumnName: String = "dimVal"): DataFrame = {
- import ss.implicits._
-
- val rawFeatures = allDimVals
- .select(col(param._countColumnName), col(s"$dimValColumnName.dim").as("dim"), col(s"$dimValColumnName.value").as("value"))
- .groupBy("dim", "value")
- .agg(countDistinct(param._countColumnName).as("count"))
- .filter(s"count > ${param._minUserCount}")
-
- val ds: Dataset[((String, Long), String)] =
- rawFeatures.select("dim", "value", "count").as[(String, String, Long)]
- .map { case (dim, value, uv) =>
- (dim, uv) -> value
- }
-
-
- implicit val ord = Ordering.Tuple2(Ordering.String, Ordering.Long.reverse)
-
- val rdd: RDD[(Long, (String, Long), String)] = WalLogUDF.appendRank(ds, param.numOfPartitions, param.samplePointsPerPartitionHint)
-
- rdd.toDF("rank", "dim_count", "value")
- .withColumn("dim", col("dim_count._1"))
- .withColumn("count", col("dim_count._2"))
- .select("dim", "value", "count", "rank")
- }
-}
-
-case class FeatureIndexProcess(taskConf: TaskConf) extends org.apache.s2graph.s2jobs.task.Process(taskConf) {
-
- import FeatureIndexProcess._
-
- override def execute(ss: SparkSession, inputMap: Map[String, DataFrame]): DataFrame = {
- val countColumnName = taskConf.options.getOrElse("countColumnName", "from")
- val numOfPartitions = taskConf.options.get("numOfPartitions").map(_.toInt)
- val samplePointsPerPartitionHint = taskConf.options.get("samplePointsPerPartitionHint").map(_.toInt)
- val minUserCount = taskConf.options.get("minUserCount").map(_.toLong)
-
- numOfPartitions.map { d => ss.sqlContext.setConf("spark.sql.shuffle.partitions", d.toString) }
-
- val param = FeatureIndexParam(minUserCount = minUserCount, countColumnName = Option(countColumnName),
- numOfPartitions = numOfPartitions, samplePointsPerPartitionHint = samplePointsPerPartitionHint
- )
-
- val edges = taskConf.inputs.tail.foldLeft(inputMap(taskConf.inputs.head)) { case (prev, cur) =>
- prev.union(inputMap(cur))
- }
-
- //TODO: user expect to inject transformers that transform (WalLog, propertyKey, propertyValue) to Seq[DimVal].
- val transformers = Seq(DefaultTransformer())
- val dimValExtractUDF = extractDimValues(transformers)
- val dimValColumnName = "dimVal"
-
- val rawFeatures = edges
- .withColumn(dimValColumnName, explode(dimValExtractUDF(col("logs"))))
-
- val dict = buildDictionary(ss, rawFeatures, param, dimValColumnName)
-
- dict
- }
-
-
- override def mandatoryOptions: Set[String] = Set.empty
-}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/FilterTopFeaturesProcess.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/FilterTopFeaturesProcess.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/FilterTopFeaturesProcess.scala
new file mode 100644
index 0000000..ec7e645
--- /dev/null
+++ b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/FilterTopFeaturesProcess.scala
@@ -0,0 +1,90 @@
+package org.apache.s2graph.s2jobs.wal.process
+
+import org.apache.s2graph.s2jobs.task.TaskConf
+import org.apache.s2graph.s2jobs.wal.WalLogAgg
+import org.apache.s2graph.s2jobs.wal.transformer.{DefaultTransformer, Transformer}
+import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.sql.functions.{col, udf}
+import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
+import play.api.libs.json.{JsObject, Json}
+
+object FilterTopFeaturesProcess {
+ private var validFeatureHashKeys: Set[Long] = null
+ def getValidFeatureHashKeys(validFeatureHashKeysBCast: Broadcast[Array[Long]]): Set[Long] = {
+ if (validFeatureHashKeys == null) {
+ validFeatureHashKeys = validFeatureHashKeysBCast.value.toSet
+ }
+
+ validFeatureHashKeys
+ }
+
+ def collectDistinctFeatureHashes(ss: SparkSession,
+ filteredDict: DataFrame): Array[Long] = {
+ import ss.implicits._
+
+ val featureHashUDF = udf((dim: String, value: String) => WalLogAgg.toFeatureHash(dim, value))
+
+ filteredDict.withColumn("featureHash", featureHashUDF(col("dim"), col("value")))
+ .select("featureHash")
+ .distinct().as[Long].collect()
+ }
+
+ def filterTopKsPerDim(dict: DataFrame,
+ maxRankPerDim: Broadcast[Map[String, Int]],
+ defaultMaxRank: Int): DataFrame = {
+ val filterUDF = udf((dim: String, rank: Long) => {
+ rank < maxRankPerDim.value.getOrElse(dim, defaultMaxRank)
+ })
+
+ dict.filter(filterUDF(col("dim"), col("rank")))
+ }
+
+ def filterWalLogAgg(ss: SparkSession,
+ walLogAgg: Dataset[WalLogAgg],
+ transformers: Seq[Transformer],
+ validFeatureHashKeysBCast: Broadcast[Array[Long]]) = {
+ import ss.implicits._
+ walLogAgg.mapPartitions { iter =>
+ val validFeatureHashKeys = getValidFeatureHashKeys(validFeatureHashKeysBCast)
+
+ iter.map { walLogAgg =>
+ WalLogAgg.filterProps(walLogAgg, transformers, validFeatureHashKeys)
+ }
+ }
+ }
+}
+
+class FilterTopFeaturesProcess(taskConf: TaskConf) extends org.apache.s2graph.s2jobs.task.Process(taskConf) {
+
+ import FilterTopFeaturesProcess._
+
+ /*
+ filter topKs per dim, then build valid dimValLs.
+ then broadcast valid dimValLs to original dataframe, and filter out not valid dimVal.
+ */
+ override def execute(ss: SparkSession, inputMap: Map[String, DataFrame]): DataFrame = {
+ import ss.implicits._
+
+ val maxRankPerDim = taskConf.options.get("maxRankPerDim").map { s =>
+ Json.parse(s).as[JsObject].fields.map { case (k, jsValue) =>
+ k -> jsValue.as[Int]
+ }.toMap
+ }
+ val maxRankPerDimBCast = ss.sparkContext.broadcast(maxRankPerDim.getOrElse(Map.empty))
+
+ val defaultMaxRank = taskConf.options.get("defaultMaxRank").map(_.toInt)
+
+ val featureDict = inputMap(taskConf.options("featureDict"))
+ val walLogAgg = inputMap(taskConf.options("walLogAgg")).as[WalLogAgg]
+
+ val transformers = TaskConf.parseTransformers(taskConf)
+
+ val filteredDict = filterTopKsPerDim(featureDict, maxRankPerDimBCast, defaultMaxRank.getOrElse(Int.MaxValue))
+ val validFeatureHashKeys = collectDistinctFeatureHashes(ss, filteredDict)
+ val validFeatureHashKeysBCast = ss.sparkContext.broadcast(validFeatureHashKeys)
+
+ filterWalLogAgg(ss, walLogAgg, transformers, validFeatureHashKeysBCast).toDF()
+ }
+
+ override def mandatoryOptions: Set[String] = Set("featureDict", "walLogAgg")
+}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/FilterWalLogAggProcess.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/FilterWalLogAggProcess.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/FilterWalLogAggProcess.scala
deleted file mode 100644
index 3c546a7..0000000
--- a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/FilterWalLogAggProcess.scala
+++ /dev/null
@@ -1,75 +0,0 @@
-package org.apache.s2graph.s2jobs.wal.process
-
-import org.apache.s2graph.s2jobs.task.TaskConf
-import org.apache.s2graph.s2jobs.wal.WalLogAgg
-import org.apache.spark.broadcast.Broadcast
-import org.apache.spark.sql.functions.{col, udf}
-import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
-import play.api.libs.json.{JsObject, Json}
-
-object FilterWalLogAggProcess {
-
- def collectDistinctFeatureHashes(ss: SparkSession,
- filteredDict: DataFrame): Array[Long] = {
- import ss.implicits._
-
- val featureHashUDF = udf((dim: String, value: String) => WalLogAgg.toFeatureHash(dim, value))
-
- filteredDict.withColumn("featureHash", featureHashUDF(col("dim"), col("value")))
- .select("featureHash")
- .distinct().as[Long].collect()
- }
-
- def filterTopKsPerDim(dict: DataFrame,
- maxRankPerDim: Broadcast[Map[String, Int]],
- defaultMaxRank: Int): DataFrame = {
- val filterUDF = udf((dim: String, rank: Long) => {
- rank < maxRankPerDim.value.getOrElse(dim, defaultMaxRank)
- })
-
- dict.filter(filterUDF(col("dim"), col("rank")))
- }
-
- def filterWalLogAgg(walLogAgg: Dataset[WalLogAgg],
- validFeatureHashKeysBCast: Broadcast[Array[Long]]) = {
- walLogAgg.mapPartitions { iter =>
- val validFeatureHashKeys = validFeatureHashKeysBCast.value.toSet
-
- iter.map { walLogAgg =>
- WalLogAgg.filter(walLogAgg, validFeatureHashKeys)
- }
- }
- }
-}
-
-class FilterWalLogAggProcess(taskConf: TaskConf) extends org.apache.s2graph.s2jobs.task.Process(taskConf) {
-
- import FilterWalLogAggProcess._
-
- /*
- filter topKs per dim, then build valid dimValLs.
- then broadcast valid dimValLs to original dataframe, and filter out not valid dimVal.
- */
- override def execute(ss: SparkSession, inputMap: Map[String, DataFrame]): DataFrame = {
- val maxRankPerDim = taskConf.options.get("maxRankPerDim").map { s =>
- Json.parse(s).as[JsObject].fields.map { case (k, jsValue) =>
- k -> jsValue.as[Int]
- }.toMap
- }
- val maxRankPerDimBCast = ss.sparkContext.broadcast(maxRankPerDim.getOrElse(Map.empty))
-
- val defaultMaxRank = taskConf.options.get("defaultMaxRank").map(_.toInt)
-
- val featureDict = inputMap(taskConf.options("featureDict"))
- val walLogAgg = inputMap(taskConf.options("walLogAgg")).as[WalLogAgg]
-
-
- val filteredDict = filterTopKsPerDim(featureDict, maxRankPerDimBCast, defaultMaxRank.getOrElse(Int.MaxValue))
- val validFeatureHashKeys = collectDistinctFeatureHashes(ss, filteredDict)
- val validFeatureHashKeysBCast = ss.sparkContext.broadcast(validFeatureHashKeys)
-
- filterWalLogAgg(walLogAgg, validFeatureHashKeysBCast).toDF()
- }
-
- override def mandatoryOptions: Set[String] = Set("featureDict", "walLogAgg")
-}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/WalLogAggregateProcess.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/WalLogAggregateProcess.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/WalLogAggregateProcess.scala
index 2b42e20..ee4debd 100644
--- a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/WalLogAggregateProcess.scala
+++ b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/WalLogAggregateProcess.scala
@@ -41,26 +41,18 @@ class WalLogAggregateProcess(taskConf: TaskConf) extends org.apache.s2graph.s2jo
override def execute(ss: SparkSession, inputMap: Map[String, DataFrame]): DataFrame = {
import ss.implicits._
- val groupByKeys = taskConf.options.get("groupByKeys").map(_.split(",").filter(_.nonEmpty).toSeq)
- val maxNumOfEdges = taskConf.options.get("maxNumOfEdges").map(_.toInt).getOrElse(1000)
- val arrayType = taskConf.options.get("arrayType").map(_.toBoolean).getOrElse(false)
- val sortTopItems = taskConf.options.get("sortTopItems").map(_.toBoolean).getOrElse(false)
- val numOfPartitions = taskConf.options.get("numOfPartitions")
- numOfPartitions.foreach(d => ss.sqlContext.setConf("spark.sql.shuffle.partitions", d))
+ //TODO: Current implementation only expect taskConf.options as Map[String, String].
+ //Once we change taskConf.options as JsObject, then we can simply parse input paramter as following.
+ //implicit val paramReads = Json.reads[AggregateParam]
+ val param = AggregateParam.fromTaskConf(taskConf)
+ param.numOfPartitions.foreach(d => ss.sqlContext.setConf("spark.sql.shuffle.partitions", d.toString))
implicit val ord = WalLog.orderByTsAsc
val walLogs = taskConf.inputs.tail.foldLeft(inputMap(taskConf.inputs.head)) { case (prev, cur) =>
prev.union(inputMap(cur))
}
- //TODO: Current implementation only expect taskConf.options as Map[String, String].
- //Once we change taskConf.options as JsObject, then we can simply parse input paramter as following.
- //implicit val paramReads = Json.reads[AggregateParam]
-
- val param = AggregateParam(groupByKeys, topK = Option(maxNumOfEdges),
- isArrayType = Option(arrayType), shouldSortTopItems = Option(sortTopItems))
-
- if (arrayType) aggregate(ss, walLogs.as[WalLogAgg], param)
+ if (param.arrayType) aggregate(ss, walLogs.as[WalLogAgg], param)
else aggregateRaw(ss, walLogs.as[WalLog], param)
}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/AggregateParam.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/AggregateParam.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/AggregateParam.scala
index 89c36f4..4cb1377 100644
--- a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/AggregateParam.scala
+++ b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/AggregateParam.scala
@@ -1,17 +1,34 @@
package org.apache.s2graph.s2jobs.wal.process.params
+import org.apache.s2graph.s2jobs.task.TaskConf
object AggregateParam {
val defaultGroupByKeys = Seq("from")
val defaultTopK = 1000
val defaultIsArrayType = false
val defaultShouldSortTopItems = true
+
+ def fromTaskConf(taskConf: TaskConf): AggregateParam = {
+ val groupByKeys = taskConf.options.get("groupByKeys").map(_.split(",").filter(_.nonEmpty).toSeq)
+ val maxNumOfEdges = taskConf.options.get("maxNumOfEdges").map(_.toInt).getOrElse(defaultTopK)
+ val arrayType = taskConf.options.get("arrayType").map(_.toBoolean).getOrElse(defaultIsArrayType)
+ val sortTopItems = taskConf.options.get("sortTopItems").map(_.toBoolean).getOrElse(defaultShouldSortTopItems)
+ val numOfPartitions = taskConf.options.get("numOfPartitions").map(_.toInt)
+
+ new AggregateParam(groupByKeys = groupByKeys,
+ topK = Option(maxNumOfEdges),
+ isArrayType = Option(arrayType),
+ shouldSortTopItems = Option(sortTopItems),
+ numOfPartitions = numOfPartitions
+ )
+ }
}
case class AggregateParam(groupByKeys: Option[Seq[String]],
topK: Option[Int],
isArrayType: Option[Boolean],
- shouldSortTopItems: Option[Boolean]) {
+ shouldSortTopItems: Option[Boolean],
+ numOfPartitions: Option[Int]) {
import AggregateParam._
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/BuildTopFeaturesParam.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/BuildTopFeaturesParam.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/BuildTopFeaturesParam.scala
new file mode 100644
index 0000000..3e8bae5
--- /dev/null
+++ b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/BuildTopFeaturesParam.scala
@@ -0,0 +1,17 @@
+package org.apache.s2graph.s2jobs.wal.process.params
+
+object BuildTopFeaturesParam {
+ val defaultMinUserCount = 0L
+ val defaultCountColumnName = "from"
+}
+
+case class BuildTopFeaturesParam(minUserCount: Option[Long],
+ countColumnName: Option[String],
+ samplePointsPerPartitionHint: Option[Int],
+ numOfPartitions: Option[Int]) {
+
+ import BuildTopFeaturesParam._
+
+ val _countColumnName = countColumnName.getOrElse(defaultCountColumnName)
+ val _minUserCount = minUserCount.getOrElse(defaultMinUserCount)
+}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/FeatureIndexParam.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/FeatureIndexParam.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/FeatureIndexParam.scala
deleted file mode 100644
index 4ede70b..0000000
--- a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/FeatureIndexParam.scala
+++ /dev/null
@@ -1,15 +0,0 @@
-package org.apache.s2graph.s2jobs.wal.process.params
-
-object FeatureIndexParam {
- val defaultMinUserCount = 0L
- val defaultCountColumnName = "from"
-}
-
-case class FeatureIndexParam(minUserCount: Option[Long],
- countColumnName: Option[String],
- samplePointsPerPartitionHint: Option[Int],
- numOfPartitions: Option[Int]) {
- import FeatureIndexParam._
- val _countColumnName = countColumnName.getOrElse(defaultCountColumnName)
- val _minUserCount = minUserCount.getOrElse(defaultMinUserCount)
-}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/FilterTopFeaturesParam.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/FilterTopFeaturesParam.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/FilterTopFeaturesParam.scala
new file mode 100644
index 0000000..3b9d868
--- /dev/null
+++ b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/FilterTopFeaturesParam.scala
@@ -0,0 +1,7 @@
+package org.apache.s2graph.s2jobs.wal.process.params
+
+class FilterTopFeaturesParam(maxRankPerDim: Option[Map[String, Int]],
+ defaultMaxRank: Option[Int]) {
+ val _maxRankPerDim = maxRankPerDim.getOrElse(Map.empty)
+ val _defaultMaxRank = defaultMaxRank.getOrElse(Int.MaxValue)
+}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/FilterWalLogAggParam.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/FilterWalLogAggParam.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/FilterWalLogAggParam.scala
deleted file mode 100644
index 4b2209f..0000000
--- a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/process/params/FilterWalLogAggParam.scala
+++ /dev/null
@@ -1,7 +0,0 @@
-package org.apache.s2graph.s2jobs.wal.process.params
-
-class FilterWalLogAggParam(maxRankPerDim: Option[Map[String, Int]],
- defaultMaxRank: Option[Int]) {
- val _maxRankPerDim = maxRankPerDim.getOrElse(Map.empty)
- val _defaultMaxRank = defaultMaxRank.getOrElse(Int.MaxValue)
-}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/DefaultTransformer.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/DefaultTransformer.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/DefaultTransformer.scala
index 2839d6a..de328e5 100644
--- a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/DefaultTransformer.scala
+++ b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/DefaultTransformer.scala
@@ -1,3 +1,5 @@
package org.apache.s2graph.s2jobs.wal.transformer
-case class DefaultTransformer() extends Transformer
+import org.apache.s2graph.s2jobs.task.TaskConf
+
+case class DefaultTransformer(taskConf: TaskConf) extends Transformer(taskConf)
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/ExtractDomain.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/ExtractDomain.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/ExtractDomain.scala
index 9b09d71..45bbe66 100644
--- a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/ExtractDomain.scala
+++ b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/ExtractDomain.scala
@@ -1,12 +1,15 @@
package org.apache.s2graph.s2jobs.wal.transformer
+import org.apache.s2graph.s2jobs.task.TaskConf
import org.apache.s2graph.s2jobs.wal.utils.UrlUtils
import org.apache.s2graph.s2jobs.wal.{DimVal, WalLog}
+import play.api.libs.json.Json
-case class ExtractDomain(urlDimensions: Set[String],
- hostDimName: String = "host",
- domainDimName: String = "domain",
- keywordDimName: String = "uri_keywords") extends Transformer {
+case class ExtractDomain(taskConf: TaskConf) extends Transformer(taskConf) {
+ val urlDimensions = Json.parse(taskConf.options.getOrElse("urlDimensions", "[]")).as[Set[String]]
+ val hostDimName = taskConf.options.getOrElse("hostDimName", "host")
+ val domainDimName= taskConf.options.getOrElse("domainDimName", "domain")
+ val keywordDimName = taskConf.options.getOrElse("keywordDimName", "uri_keywords")
override def toDimValLs(walLog: WalLog, propertyKey: String, propertyValue: String): Seq[DimVal] = {
if (!urlDimensions(propertyKey)) Nil
else {
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/ExtractServiceName.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/ExtractServiceName.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/ExtractServiceName.scala
index 7ad7dfd..15efe9f 100644
--- a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/ExtractServiceName.scala
+++ b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/ExtractServiceName.scala
@@ -1,10 +1,17 @@
package org.apache.s2graph.s2jobs.wal.transformer
+import org.apache.s2graph.core.JSONParser
+import org.apache.s2graph.s2jobs.task.TaskConf
import org.apache.s2graph.s2jobs.wal.{DimVal, WalLog}
+import play.api.libs.json.{JsObject, Json}
+
+class ExtractServiceName(taskConf: TaskConf) extends Transformer(taskConf) {
+ val serviceDims = Json.parse(taskConf.options.getOrElse("serviceDims", "[]")).as[Set[String]]
+ val domainServiceMap = Json.parse(taskConf.options.getOrElse("domainServiceMap", "{}")).as[JsObject].fields.map { case (k, v) =>
+ k -> JSONParser.jsValueToString(v)
+ }.toMap
+ val serviceDimName = taskConf.options.getOrElse("serviceDimName", "serviceDimName")
-class ExtractServiceName(serviceDims: Set[String],
- domainServiceMap: Map[String, String] = Map.empty,
- serviceDimName: String = "serviceName") extends Transformer {
override def toDimValLs(walLog: WalLog, propertyKey: String, propertyValue: String): Seq[DimVal] = {
if (!serviceDims(propertyKey)) Nil
else {
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/Transformer.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/Transformer.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/Transformer.scala
index 523f58d..68be2ad 100644
--- a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/Transformer.scala
+++ b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/transformer/Transformer.scala
@@ -1,11 +1,12 @@
package org.apache.s2graph.s2jobs.wal.transformer
+import org.apache.s2graph.s2jobs.task.TaskConf
import org.apache.s2graph.s2jobs.wal.{DimVal, WalLog}
/**
* decide how to transform walLog's each property key value to Seq[DimVal]
*/
-trait Transformer {
+abstract class Transformer(taskConf: TaskConf) extends Serializable {
def toDimValLs(walLog: WalLog, propertyKey: String, propertyValue: String): Seq[DimVal] = {
val dim = s"${walLog.label}:${propertyKey}"
val value = propertyValue
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/udfs/WalLogUDF.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/udfs/WalLogUDF.scala b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/udfs/WalLogUDF.scala
index 34f4a2b..d213d6c 100644
--- a/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/udfs/WalLogUDF.scala
+++ b/s2jobs/src/main/scala/org/apache/s2graph/s2jobs/wal/udfs/WalLogUDF.scala
@@ -137,22 +137,16 @@ object WalLogUDF {
val sorted = rdd.repartitionAndSortWithinPartitions(partitioner)
def rank(idx: Int, iter: Iterator[((K1, K2), V)]) = {
- val initialK1: K1 = null.asInstanceOf[K1]
- val initialK2: K2 = null.asInstanceOf[K2]
- val initialV: V = null.asInstanceOf[V]
- val zero = List((1L, initialK1, initialK2, initialV))
-
- def merge(acc: List[(Long, K1, K2, V)], x: ((K1, K2), V)) =
- (acc.head, x) match {
- case ((offset, prevKey1, _, _), ((curKey1: K1, curKey2: K2), curVal: V)) => {
- val newOffset = if (prevKey1 == curKey1) offset + 1L else 1L
- (newOffset, curKey1, curKey2, curVal) :: acc
- }
- }
-
- iter.foldLeft(zero)(merge).reverse.drop(1).map { case (offset, key1, key2, value) =>
- (idx, offset, key1, key2, value)
- }.toIterator
+ var curOffset = 1L
+ var curK1 = null.asInstanceOf[K1]
+
+ iter.map{ case ((key1, key2), value) =>
+ // println(s">>>[$idx] curK1: $curK1, curOffset: $curOffset")
+ val newOffset = if (curK1 == key1) curOffset + 1L else 1L
+ curOffset = newOffset
+ curK1 = key1
+ (idx, newOffset, key1, key2, value)
+ }
}
def getOffset(idx: Int, iter: Iterator[((K1, K2), V)]) = {
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/TestData.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/TestData.scala b/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/TestData.scala
new file mode 100644
index 0000000..fae9265
--- /dev/null
+++ b/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/TestData.scala
@@ -0,0 +1,42 @@
+package org.apache.s2graph.s2jobs.wal
+
+object TestData {
+ val testServiceName = "s2graph"
+ val walLogsLs = Seq(
+ WalLog(1L, "insert", "edge", "u1", "i1", s"$testServiceName", "click", """{"item_name":"awesome item"}"""),
+ WalLog(2L, "insert", "edge", "u1", "i1", s"$testServiceName", "purchase", """{"price":2}"""),
+ WalLog(3L, "insert", "edge", "u1", "q1", s"$testServiceName", "search", """{"referrer":"www.google.com"}"""),
+ WalLog(4L, "insert", "edge", "u2", "i1", s"$testServiceName", "click", """{"item_name":"awesome item"}"""),
+ WalLog(5L, "insert", "edge", "u2", "q2", s"$testServiceName", "search", """{"referrer":"www.bing.com"}"""),
+ WalLog(6L, "insert", "edge", "u3", "i2", s"$testServiceName", "click", """{"item_name":"bad item"}"""),
+ WalLog(7L, "insert", "edge", "u4", "q1", s"$testServiceName", "search", """{"referrer":"www.google.com"}""")
+ )
+
+ // order by from
+ val aggExpected = Array(
+ WalLogAgg("u1", Seq(
+ WalLog(3L, "insert", "edge", "u1", "q1", s"$testServiceName", "search", """{"referrer":"www.google.com"}"""),
+ WalLog(2L, "insert", "edge", "u1", "i1", s"$testServiceName", "purchase", """{"price":2}"""),
+ WalLog(1L, "insert", "edge", "u1", "i1", s"$testServiceName", "click", """{"item_name":"awesome item"}""")
+ ), 3L, 1L),
+ WalLogAgg("u2", Seq(
+ WalLog(5L, "insert", "edge", "u2", "q2", s"$testServiceName", "search", """{"referrer":"www.bing.com"}"""),
+ WalLog(4L, "insert", "edge", "u2", "i1", s"$testServiceName", "click", """{"item_name":"awesome item"}""")
+ ), 5L, 4L),
+ WalLogAgg("u3", Seq(
+ WalLog(6L, "insert", "edge", "u3", "i2", s"$testServiceName", "click", """{"item_name":"bad item"}""")
+ ), 6L, 6L),
+ WalLogAgg("u4", Seq(
+ WalLog(7L, "insert", "edge", "u4", "q1", s"$testServiceName", "search", """{"referrer":"www.google.com"}""")
+ ), 7L, 7L)
+ )
+
+ // order by dim, rank
+ val featureDictExpected = Array(
+ DimValCountRank(DimVal("click:item_name", "awesome item"), 2, 1),
+ DimValCountRank(DimVal("click:item_name", "bad item"), 1, 2),
+ DimValCountRank(DimVal("purchase:price", "2"), 1, 1),
+ DimValCountRank(DimVal("search:referrer", "www.google.com"), 2, 1),
+ DimValCountRank(DimVal("search:referrer", "www.bing.com"), 1, 2)
+ )
+}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/TransformerTest.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/TransformerTest.scala b/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/TransformerTest.scala
index 5089390..974002b 100644
--- a/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/TransformerTest.scala
+++ b/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/TransformerTest.scala
@@ -1,20 +1,26 @@
package org.apache.s2graph.s2jobs.wal
+import org.apache.s2graph.s2jobs.task.TaskConf
import org.apache.s2graph.s2jobs.wal.transformer._
import org.scalatest.{FunSuite, Matchers}
+import play.api.libs.json.Json
class TransformerTest extends FunSuite with Matchers {
val walLog = WalLog(1L, "insert", "edge", "a", "b", "s2graph", "friends", """{"name": 1, "url": "www.google.com"}""")
test("test default transformer") {
- val transformer = new DefaultTransformer
+ val taskConf = TaskConf.Empty
+ val transformer = new DefaultTransformer(taskConf)
val dimVals = transformer.toDimValLs(walLog, "name", "1")
dimVals shouldBe Seq(DimVal("friends:name", "1"))
}
test("test ExtractDomain from URL") {
- val transformer = new ExtractDomain(urlDimensions = Set("url"))
+ val taskConf = TaskConf.Empty.copy(options =
+ Map("urlDimensions" -> Json.toJson(Seq("url")).toString())
+ )
+ val transformer = new ExtractDomain(taskConf)
val dimVals = transformer.toDimValLs(walLog, "url", "http://www.google.com/abc")
dimVals shouldBe Seq(
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/process/BuildTopFeaturesProcessTest.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/process/BuildTopFeaturesProcessTest.scala b/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/process/BuildTopFeaturesProcessTest.scala
new file mode 100644
index 0000000..4d2f079
--- /dev/null
+++ b/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/process/BuildTopFeaturesProcessTest.scala
@@ -0,0 +1,31 @@
+package org.apache.s2graph.s2jobs.wal.process
+
+import com.holdenkarau.spark.testing.DataFrameSuiteBase
+import org.apache.s2graph.s2jobs.task.TaskConf
+import org.apache.s2graph.s2jobs.wal.DimValCountRank
+import org.scalatest.{BeforeAndAfterAll, FunSuite, Matchers}
+
+class BuildTopFeaturesProcessTest extends FunSuite with Matchers with BeforeAndAfterAll with DataFrameSuiteBase {
+
+ import org.apache.s2graph.s2jobs.wal.TestData._
+
+ test("test entire process.") {
+ import spark.implicits._
+ val df = spark.createDataset(aggExpected).toDF()
+
+ val taskConf = new TaskConf(name = "test", `type` = "test", inputs = Seq("input"),
+ options = Map("minUserCount" -> "0")
+ )
+ val job = new BuildTopFeaturesProcess(taskConf)
+
+
+ val inputMap = Map("input" -> df)
+ val featureDicts = job.execute(spark, inputMap)
+ .orderBy("dim", "rank")
+ .map(DimValCountRank.fromRow)
+ .collect()
+
+ featureDicts shouldBe featureDictExpected
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/process/FilterTopFeaturesProcessTest.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/process/FilterTopFeaturesProcessTest.scala b/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/process/FilterTopFeaturesProcessTest.scala
new file mode 100644
index 0000000..cd8295a
--- /dev/null
+++ b/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/process/FilterTopFeaturesProcessTest.scala
@@ -0,0 +1,84 @@
+package org.apache.s2graph.s2jobs.wal.process
+
+import com.holdenkarau.spark.testing.DataFrameSuiteBase
+import org.apache.s2graph.s2jobs.task.TaskConf
+import org.apache.s2graph.s2jobs.wal.transformer.DefaultTransformer
+import org.apache.s2graph.s2jobs.wal.{DimValCountRank, WalLogAgg}
+import org.scalatest.{BeforeAndAfterAll, FunSuite, Matchers}
+
+class FilterTopFeaturesProcessTest extends FunSuite with Matchers with BeforeAndAfterAll with DataFrameSuiteBase {
+ import org.apache.s2graph.s2jobs.wal.TestData._
+
+ test("test filterTopKsPerDim.") {
+ import spark.implicits._
+ val featureDf = spark.createDataset(featureDictExpected).map { x =>
+ (x.dimVal.dim, x.dimVal.value, x.count, x.rank)
+ }.toDF("dim", "value", "count", "rank")
+
+ val maxRankPerDim = spark.sparkContext.broadcast(Map.empty[String, Int])
+
+ // filter nothing because all feature has rank < 10
+ val filtered = FilterTopFeaturesProcess.filterTopKsPerDim(featureDf, maxRankPerDim, 10)
+
+ val real = filtered.orderBy("dim", "rank").map(DimValCountRank.fromRow).collect()
+ real.zip(featureDictExpected).foreach { case (real, expected) =>
+ real shouldBe expected
+ }
+ // filter rank >= 2
+ val filtered2 = FilterTopFeaturesProcess.filterTopKsPerDim(featureDf, maxRankPerDim, 2)
+ val real2 = filtered2.orderBy("dim", "rank").map(DimValCountRank.fromRow).collect()
+ real2 shouldBe featureDictExpected.filter(_.rank < 2)
+ }
+
+
+ test("test filterWalLogAgg.") {
+ import spark.implicits._
+ val walLogAgg = spark.createDataset(aggExpected)
+ val featureDf = spark.createDataset(featureDictExpected).map { x =>
+ (x.dimVal.dim, x.dimVal.value, x.count, x.rank)
+ }.toDF("dim", "value", "count", "rank")
+ val maxRankPerDim = spark.sparkContext.broadcast(Map.empty[String, Int])
+
+ val transformers = Seq(DefaultTransformer(TaskConf.Empty))
+ // filter nothing. so input, output should be same.
+ val featureFiltered = FilterTopFeaturesProcess.filterTopKsPerDim(featureDf, maxRankPerDim, 10)
+ val validFeatureHashKeys = FilterTopFeaturesProcess.collectDistinctFeatureHashes(spark, featureFiltered)
+ val validFeatureHashKeysBCast = spark.sparkContext.broadcast(validFeatureHashKeys)
+ val real = FilterTopFeaturesProcess.filterWalLogAgg(spark, walLogAgg, transformers, validFeatureHashKeysBCast)
+ .collect().sortBy(_.from)
+
+ real.zip(aggExpected).foreach { case (real, expected) =>
+ real shouldBe expected
+ }
+ }
+
+ test("test entire process. filter nothing.") {
+ import spark.implicits._
+ val df = spark.createDataset(aggExpected).toDF()
+ val featureDf = spark.createDataset(featureDictExpected).map { x =>
+ (x.dimVal.dim, x.dimVal.value, x.count, x.rank)
+ }.toDF("dim", "value", "count", "rank")
+
+ val inputKey = "input"
+ val featureDictKey = "feature"
+ // filter nothing since we did not specified maxRankPerDim and defaultMaxRank.
+ val taskConf = new TaskConf(name = "test", `type` = "test",
+ inputs = Seq(inputKey, featureDictKey),
+ options = Map(
+ "featureDict" -> featureDictKey,
+ "walLogAgg" -> inputKey
+ )
+ )
+ val inputMap = Map(inputKey -> df, featureDictKey -> featureDf)
+ val job = new FilterTopFeaturesProcess(taskConf)
+ val filtered = job.execute(spark, inputMap)
+ .orderBy("from")
+ .as[WalLogAgg]
+ .collect()
+
+ filtered.zip(aggExpected).foreach { case (real, expected) =>
+ real shouldBe expected
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/process/WalLogAggregateProcessTest.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/process/WalLogAggregateProcessTest.scala b/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/process/WalLogAggregateProcessTest.scala
index 5753411..1bb7426 100644
--- a/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/process/WalLogAggregateProcessTest.scala
+++ b/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/process/WalLogAggregateProcessTest.scala
@@ -2,35 +2,20 @@ package org.apache.s2graph.s2jobs.wal.process
import com.holdenkarau.spark.testing.DataFrameSuiteBase
import org.apache.s2graph.s2jobs.task.TaskConf
-import org.apache.s2graph.s2jobs.wal.WalLog
-import org.apache.s2graph.s2jobs.wal.udafs._
-import org.apache.s2graph.s2jobs.wal.utils.BoundedPriorityQueue
+import org.apache.s2graph.s2jobs.wal._
import org.scalatest.{BeforeAndAfterAll, FunSuite, Matchers}
-import scala.collection.mutable
-import scala.util.Random
-
class WalLogAggregateProcessTest extends FunSuite with Matchers with BeforeAndAfterAll with DataFrameSuiteBase {
- val walLogsLs = Seq(
- WalLog(1L, "insert", "edge", "a", "b", "s2graph", "friends", """{"name": 1}"""),
- WalLog(2L, "insert", "edge", "a", "c", "s2graph", "friends", """{"name": 2}"""),
- WalLog(3L, "insert", "edge", "a", "d", "s2graph", "friends", """{"name": 3}"""),
- WalLog(4L, "insert", "edge", "a", "b", "s2graph", "friends", """{"name": 4}""")
- )
- val walLogsLs2 = Seq(
- WalLog(5L, "insert", "edge", "a", "b", "s2graph", "friends", """{"name": 1}"""),
- WalLog(6L, "insert", "edge", "a", "c", "s2graph", "friends", """{"name": 2}"""),
- WalLog(7L, "insert", "edge", "a", "d", "s2graph", "friends", """{"name": 3}"""),
- WalLog(8L, "insert", "edge", "a", "b", "s2graph", "friends", """{"name": 4}""")
- )
-
+ import org.apache.s2graph.s2jobs.wal.TestData._
- test("test S2EdgeDataAggregateProcess") {
+ test("test entire process") {
import spark.sqlContext.implicits._
val edges = spark.createDataset(walLogsLs).toDF()
- val inputMap = Map("edges" -> edges)
- val taskConf = new TaskConf(name = "test", `type` = "agg", inputs = Seq("edges"),
+ val processKey = "agg"
+ val inputMap = Map(processKey -> edges)
+
+ val taskConf = new TaskConf(name = "test", `type` = "agg", inputs = Seq(processKey),
options = Map("maxNumOfEdges" -> "10")
)
@@ -38,40 +23,9 @@ class WalLogAggregateProcessTest extends FunSuite with Matchers with BeforeAndAf
val processed = job.execute(spark, inputMap)
processed.printSchema()
- processed.collect().foreach { row =>
- println(row)
- }
- }
-
- test("mergeTwoSeq") {
- val prev: Array[Int] = Array(3, 2, 1)
- val cur: Array[Int] = Array(4, 2, 2)
-
- val ls = WalLogUDAF.mergeTwoSeq(prev, cur, 10)
- println(ls.size)
-
- ls.foreach { x =>
- println(x)
+ processed.orderBy("from").as[WalLogAgg].collect().zip(aggExpected).foreach { case (real, expected) =>
+ real shouldBe expected
}
}
- test("addToTopK test.") {
- import WalLogUDAF._
- val numOfTest = 100
- val numOfNums = 100
- val maxNum = 10
-
- (0 until numOfTest).foreach { testNum =>
- val maxSize = 1 + Random.nextInt(numOfNums)
- val pq = new BoundedPriorityQueue[Int](maxSize)
- val arr = (0 until numOfNums).map(x => Random.nextInt(maxNum))
- var result: mutable.Seq[Int] = mutable.ArrayBuffer.empty[Int]
-
- arr.foreach { i =>
- pq += i
- result = addToTopK(result, maxSize, i)
- }
- result.sorted shouldBe pq.toSeq.sorted
- }
- }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/651ca2cf/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/udafs/WalLogUDAFTest.scala
----------------------------------------------------------------------
diff --git a/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/udafs/WalLogUDAFTest.scala b/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/udafs/WalLogUDAFTest.scala
new file mode 100644
index 0000000..aded56d
--- /dev/null
+++ b/s2jobs/src/test/scala/org/apache/s2graph/s2jobs/wal/udafs/WalLogUDAFTest.scala
@@ -0,0 +1,42 @@
+package org.apache.s2graph.s2jobs.wal.udafs
+
+import org.apache.s2graph.s2jobs.wal.utils.BoundedPriorityQueue
+import org.scalatest._
+
+import scala.collection.mutable
+import scala.util.Random
+
+class WalLogUDAFTest extends FunSuite with Matchers {
+
+ test("mergeTwoSeq") {
+ val prev: Array[Int] = Array(3, 2, 1)
+ val cur: Array[Int] = Array(4, 2, 2)
+
+ val ls = WalLogUDAF.mergeTwoSeq(prev, cur, 10)
+ println(ls.size)
+
+ ls.foreach { x =>
+ println(x)
+ }
+ }
+
+ test("addToTopK test.") {
+ import WalLogUDAF._
+ val numOfTest = 100
+ val numOfNums = 100
+ val maxNum = 10
+
+ (0 until numOfTest).foreach { testNum =>
+ val maxSize = 1 + Random.nextInt(numOfNums)
+ val pq = new BoundedPriorityQueue[Int](maxSize)
+ val arr = (0 until numOfNums).map(x => Random.nextInt(maxNum))
+ var result: mutable.Seq[Int] = mutable.ArrayBuffer.empty[Int]
+
+ arr.foreach { i =>
+ pq += i
+ result = addToTopK(result, maxSize, i)
+ }
+ result.sorted shouldBe pq.toSeq.sorted
+ }
+ }
+}