<commit_before>package com.intel.PRCcloud.streamBench

import com.intel.PRCcloud.streamBench.entity.ParamEntity
import com.intel.PRCcloud.streamBench.util._
import com.intel.PRCcloud.streamBench.microbench._

object RunBench {
	def main(args: Array[String]) {
		//val args2=Array("micro/identity","spark_test_1_1","spark://sr119:7077","1","sr464:2181","spark_consumer_1","1","100000")
		this.run(args)
	}

	def run(args: Array[String]) {
		//		if(args.length < 8){
		//			BenchLogUtil.handleError("Usage: <BENCHNAME> <TOPIC> <MASTER> <BATCH_INTERVAL> <ZKHOST> <CONSUMER_GROUP> <KAFKA_THREADS>")
		//		}
		if (args.length < 1) {
			BenchLogUtil.handleError("Usage: RunBench <ConfigFile>")
		}

		val conf = new ConfigLoader(args(0))
		//		val benchName=args(0)
		//		val topic=args(1)
		//		val master=args(2)
		//		val batchInterval=args(3).toInt
		//		val zkHost=args(4)
		//		val consumerGroup=args(5)
		//		val kafkaThreads=args(6).toInt
		//		val recordCount=args(7).toLong
		//		val copies=args(8).toInt
		//		val testWAL=args(9).toBoolean
		//		var path=""
		//		var nexti=10
		//		if (testWAL) {
		//						path=args(10)
		//						nexti=11
		//		}
		//
		//		val debug = args(nexti).toBoolean
		//		val directMode = args(nexti+1).toBoolean
		//var brokerList = ""
		//		if (directMode) {
		//			brokerList = args(nexti+2)
		//		}

		val benchName = conf.getPropertiy("hibench.streamingbench.benchname")
		val topic = conf.getPropertiy("hibench.streamingbench.topic_name")
		val master = conf.getPropertiy("hibench.spark.master")
		val batchInterval = conf.getPropertiy("hibench.streamingbench.batch_interval").toInt
		val zkHost = conf.getPropertiy("hibench.streamingbench.zookeeper.host")
		val consumerGroup = conf.getPropertiy("hibench.streamingbench.consumer_group")
		val kafkaThreads = conf.getPropertiy("hibench.streamingbench.receiver_nodes").toInt
		val recordCount = conf.getPropertiy("hibench.streamingbench.record_count").toLong
		val copies = conf.getPropertiy("hibench.streamingbench.copies").toInt
		val testWAL = conf.getPropertiy("hibench.streamingbench.testWAL").toBoolean
		val path = if (testWAL) conf.getPropertiy("hibench.streamingbench.checkpoint_path") else ""
		val debug = conf.getPropertiy("hibench.streamingbench.debug").toBoolean
		val directMode = conf.getPropertiy("hibench.streamingbench.direct_mode").toBoolean
		val brokerList = if (directMode) conf.getPropertiy("hibench.streamingbench.broker_list_with_quote") else ""


		val param = ParamEntity(master, benchName, batchInterval, zkHost, consumerGroup, topic, kafkaThreads, recordCount, copies, testWAL, path, debug, directMode, brokerList)
    benchName match {
      case "micro-projection" =>
        val fieldIndex = conf.getPropertiy("hibench.streamingbench.field_index").toInt
        val separator = conf.getPropertiy("hibench.streamingbench.separator")
        val ProjectTest = new StreamProjectionJob(param, fieldIndex, separator)
        ProjectTest.run()
      case "micro-sample" =>
        val prob = conf.getPropertiy("hibench.streamingbench.prob").toDouble
        val SampleTest = new SampleStreamJob(param, prob)
        SampleTest.run()
      case "micro-statistics" =>
        val fieldIndex = conf.getPropertiy("hibench.streamingbench.field_index").toInt
        val separator = conf.getPropertiy("hibench.streamingbench.separator")
        val numericCalc = new NumericCalcJob(param, fieldIndex, separator)
        numericCalc.run()
      case "micro-wordcount" =>
        val separator = conf.getPropertiy("hibench.streamingbench.separator")
        val wordCount = new Wordcount(param, separator)
        wordCount.run()
      case "micro-grep" =>
        val pattern = conf.getPropertiy("hibench.streamingbench.pattern")
        val GrepStream = new GrepStreamJob(param, pattern)
        GrepStream.run()
      case "micro-distinctcount" =>
        val fieldIndex = conf.getPropertiy("hibench.streamingbench.field_index").toInt
        val separator = conf.getPropertiy("hibench.streamingbench.separator")
        val distinct = new DistinctCountJob(param, fieldIndex, separator)
        distinct.run()<commit_msg>fix minor bug for spark streaming<commit_after>package com.intel.PRCcloud.streamBench

import com.intel.PRCcloud.streamBench.entity.ParamEntity
import com.intel.PRCcloud.streamBench.util._
import com.intel.PRCcloud.streamBench.microbench._

object RunBench {
	def main(args: Array[String]) {
		//val args2=Array("micro/identity","spark_test_1_1","spark://sr119:7077","1","sr464:2181","spark_consumer_1","1","100000")
		this.run(args)
	}

	def run(args: Array[String]) {
		//		if(args.length < 8){
		//			BenchLogUtil.handleError("Usage: <BENCHNAME> <TOPIC> <MASTER> <BATCH_INTERVAL> <ZKHOST> <CONSUMER_GROUP> <KAFKA_THREADS>")
		//		}
		if (args.length < 1) {
			BenchLogUtil.handleError("Usage: RunBench <ConfigFile>")
		}

		val conf = new ConfigLoader(args(0))
		//		val benchName=args(0)
		//		val topic=args(1)
		//		val master=args(2)
		//		val batchInterval=args(3).toInt
		//		val zkHost=args(4)
		//		val consumerGroup=args(5)
		//		val kafkaThreads=args(6).toInt
		//		val recordCount=args(7).toLong
		//		val copies=args(8).toInt
		//		val testWAL=args(9).toBoolean
		//		var path=""
		//		var nexti=10
		//		if (testWAL) {
		//						path=args(10)
		//						nexti=11
		//		}
		//
		//		val debug = args(nexti).toBoolean
		//		val directMode = args(nexti+1).toBoolean
		//var brokerList = ""
		//		if (directMode) {
		//			brokerList = args(nexti+2)
		//		}

		val benchName = conf.getPropertiy("hibench.streamingbench.benchname")
		val topic = conf.getPropertiy("hibench.streamingbench.topic_name")
		val master = conf.getPropertiy("hibench.spark.master")
		val batchInterval = conf.getPropertiy("hibench.streamingbench.batch_interval").toInt
		val zkHost = conf.getPropertiy("hibench.streamingbench.zookeeper.host")
		val consumerGroup = conf.getPropertiy("hibench.streamingbench.consumer_group")
		val kafkaThreads = conf.getPropertiy("hibench.streamingbench.receiver_nodes").toInt
		val recordCount = conf.getPropertiy("hibench.streamingbench.record_count").toLong
		val copies = conf.getPropertiy("hibench.streamingbench.copies").toInt
		val testWAL = conf.getPropertiy("hibench.streamingbench.testWAL").toBoolean
		val path = if (testWAL) conf.getPropertiy("hibench.streamingbench.checkpoint_path") else ""
		val debug = conf.getPropertiy("hibench.streamingbench.debug").toBoolean
		val directMode = conf.getPropertiy("hibench.streamingbench.direct_mode").toBoolean
		val brokerList = if (directMode) conf.getPropertiy("hibench.streamingbench.brokerList") else ""


		val param = ParamEntity(master, benchName, batchInterval, zkHost, consumerGroup, topic, kafkaThreads, recordCount, copies, testWAL, path, debug, directMode, brokerList)
    benchName match {
      case "micro-projection" =>
        val fieldIndex = conf.getPropertiy("hibench.streamingbench.field_index").toInt
        val separator = conf.getPropertiy("hibench.streamingbench.separator")
        val ProjectTest = new StreamProjectionJob(param, fieldIndex, separator)
        ProjectTest.run()
      case "micro-sample" =>
        val prob = conf.getPropertiy("hibench.streamingbench.prob").toDouble
        val SampleTest = new SampleStreamJob(param, prob)
        SampleTest.run()
      case "micro-statistics" =>
        val fieldIndex = conf.getPropertiy("hibench.streamingbench.field_index").toInt
        val separator = conf.getPropertiy("hibench.streamingbench.separator")
        val numericCalc = new NumericCalcJob(param, fieldIndex, separator)
        numericCalc.run()
      case "micro-wordcount" =>
        val separator = conf.getPropertiy("hibench.streamingbench.separator")
        val wordCount = new Wordcount(param, separator)
        wordCount.run()
      case "micro-grep" =>
        val pattern = conf.getPropertiy("hibench.streamingbench.pattern")
        val GrepStream = new GrepStreamJob(param, pattern)
        GrepStream.run()
      case "micro-distinctcount" =>
        val fieldIndex = conf.getPropertiy("hibench.streamingbench.field_index").toInt
        val separator = conf.getPropertiy("hibench.streamingbench.separator")
        val distinct = new DistinctCountJob(param, fieldIndex, separator)
        distinct.run()
{'fraction_non_alphanumeric': 0.1286577992744861, 'fraction_numerical': 0.011366384522370012, 'mean_word_length': 3.952694610778443, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4559764', 'n_tokens_mistral': 3047, 'n_tokens_neox': 2836, 'n_words': 450}