KafkaPipeline.scala

package com.powerpanel.kso;

import com.powerpanel.kso._
import com.powerpanel.kso.model._
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.SparkConf
import org.apache.log4j.Logger;

class KafkaPipeline extends Serializable {

    object Holder extends Serializable {
        @transient lazy val logger = Logger.getLogger(getClass.getName)
    }

    def create() = {

        val props = AppConfig.loadProperties();
        val checkpointDirectory = props.getProperty("app.checkpoint_path");
        val batchSizeSeconds = Integer.parseInt(props.getProperty("app.batch_size_seconds"));

        val sparkConf = new SparkConf();
        Holder.logger.info("Creating new spark context with checkpoint directory: " + checkpointDirectory)
        val ssc = new StreamingContext(sparkConf, Seconds(batchSizeSeconds));

        if (checkpointDirectory.length() > 0) {
            ssc.checkpoint(checkpointDirectory);
        }

        val dataStream = new KafkaInput().readFromKafka(ssc);
        val writeCounts: DStream[Integer] =
        
        new OpenTSDBOutput().putOpentsdb(
            props.getProperty("opentsdb.ip"),
            dataStream);

        writeCounts.reduce(_ + _).print(1);
        ssc;
    }: StreamingContext
}