yueawang/oap-perf-suite

OAP Cluster Performance TestSuite

oap-perf-suite's Issues

A guide is needed to understand aims of the test cases

I can also link this project to OAP.

OAP PR test support.

SO it can be used to pre-check PR's readiness.

Add OAP commit diff to report.

So we can further make sure if there is a real regression or not.

Add on-demand data gen.

We need to precisely control the data scale, format, column, and row group for more test purposes.

object OapBenchmarkOnDemandDataBuilder extends OapPerfSuiteContext with Logging {

private val defaultProperties = Map(
"oap.benchmark.compression.codec" -> "gzip",
"oap.benchmark.support.oap.version" -> "0.4.0",
"oap.benchmark.tpcds.tool.dir" -> "/home/oap/tpcds-kit/tools",
"oap.benchmark.hdfs.file.root.dir" -> "/user/oap/oaptest/",
"oap.benchmark.database.prefix" -> "",
"oap.benchmark.database.postfix" -> "",
"oap.benchmark.tpcds.data.scale" -> "200",
"oap.benchmark.tpcds.data.partition" -> "80"
)

def getDatabase(format: String) : String = {
val prefix = properties.get("oap.benchmark.database.prefix").get
val postfix = properties.get("oap.benchmark.database.postfix").get
val dataScale = properties.get("oap.benchmark.tpcds.data.scale").get.toInt
val baseName = format match {
case "oap" => s"oap_tpcds_$dataScale"
case "parquet" => s"parquet_tpcds_$dataScale"
case _ => "default"
}

prefix + baseName + postfix

}

def formatTableLocation(rootDir: String, versionNum: String, tableFormat: String): String = {
s"${rootDir}/${versionNum}/ondemand/${getDatabase(tableFormat)}/"
}

private val properties = {
try {
new mutable.HashMapString, String ++=
Utils.getPropertiesFromFile("./conf/oap-benchmark-default.conf")
} catch {
case e: IllegalArgumentException => {
logWarning(e.getMessage + ". Use default setting!")
defaultProperties
}
}
}

override def beforeAll(conf: Map[String, String] = Map.empty): Unit = {
super.beforeAll(conf)
}

def generateTables(dataFormats: Array[String] = Array("oap", "parquet")): Unit = {
val versionNum = properties.get("oap.benchmark.support.oap.version").get
val codec = properties.get("oap.benchmark.compression.codec").get
val scale = properties.get("oap.benchmark.tpcds.data.scale").get.toInt
val partitions = properties.get("oap.benchmark.tpcds.data.partition").get.toInt
val hdfsRootDir = properties.get("oap.benchmark.hdfs.file.root.dir").get
val tpcdsToolPath = properties.get("oap.benchmark.tpcds.tool.dir").get

protected object testImplicits extends SQLImplicits {
  protected override def _sqlContext: SQLContext = spark.sqlContext
}
import testImplicits._

dataFormats.foreach{ format =>
  sqlContext.setConf(s"spark.sql.$format.compression.codec", codec)
  val loc = formatTableLocation(hdfsRootDir, versionNum, format)
  val constant = scale
  val data = spark.sparkContext.parallelize(1 to partitions, partitions).flatMap { i => {
    new Iterator[Long] {
            val rand = new java.util.Random(System.currentTimeMillis())
            var count = constant
            def hasNext = count > 0
            def next(): Long = {
              count -= 1
              rand.nextLong()
            }
          }
        }
      }
  data.toDF("value").write.mode("overwrite").format("oap").save(loc + "store_sales")
}

}

def generateDatabases() {
// TODO: get from OapFileFormatConfigSet
val dataFormats: Seq[String] = Seq("oap", "parquet")
dataFormats.foreach { format =>
spark.sql(s"create database if not exists ${getDatabase(format)}")
val versionNum = properties.get("oap.benchmark.support.oap.version").get
val hdfsRootDir = properties.get("oap.benchmark.hdfs.file.root.dir").get
val dataLocation = formatTableLocation(hdfsRootDir, versionNum, format)

  spark.sql(s"use ${getDatabase(format)}")
  spark.sql("drop table if exists store_sales")
  spark.sqlContext.createExternalTable("store_sales", dataLocation + "store_sales", format)
}

}
}

Should be easy to re-run failed test.