xxx/message-folder

              |--report_push-0
              |--report_push-1
              |--report_push-2
              |--report_push-3
              |--launch_info-0
              |--launch_info-1
              |--launch_info-2
              |--launch_info-3

from kafka import KafkaConsumer

class KafkaStreamTest:
    '''
    This class consume all external Kafka topics'''

    def __init__(self):

        self.appName = "kafkatest"

        self.kafkaHosts = "192.168.4.201:6667,192.168.4.231:6667"
        self.kafkaAutoOffsetReset = "largest"
        self._kafka_topic = "test"

    def start(self):
        reload(sys)
        sys.setdefaultencoding('utf-8')

        elogging.debug(self.appName, elogging.normalCID(), "receiver starting")
        consumer = KafkaConsumer('test',  bootstrap_servers=['192.168.4.201:6667','192.168.4.231:6667'], enable_auto_commit=True, auto_offset_reset='earliest')
        #consumer = KafkaConsumer('test', bootstrap_servers=['192.168.4.201:6667', '192.168.4.231:6667'], auto_offset_reset='earliest')
        while True:
            # The definition of KafkaMessage:
            # KafkaMessage = namedtuple("KafkaMessage",
            #     ["topic", "partition", "offset", "key", "value"])
            kafkaMsg = consumer.next()

            # for debug
            print kafkaMsg.topic, kafkaMsg.partition, kafkaMsg.offset, kafkaMsg.key, kafkaMsg.value

if __name__ =="__main__":
    test = KafkaStreamTest()
    test.start()

import os
import sys
sys.path.append("..")
sys.path.append(sys.argv[0][:sys.argv[0].rfind(os.path.join('com','ericsson'))])

import copy
import traceback
import redis
from pyspark import SparkContext, SparkConf
from pyspark.streaming import StreamingContext, DStream
from pyspark.sql import SQLContext
import simplejson as json
from com.ericsson.analytics.fms.common.common import ELogForDistributedApp,getSqlContextInstance
from pyspark.streaming.kafka import KafkaUtils,TopicAndPartition
from com.ericsson.analytics.oamf.client.logging import elogging
from com.ericsson.analytics.fms.common.common import HDFSOperation

class KafkaStreamTest:
    '''
    This class consume all external Kafka topics, store the data into Parquet and send the data to internal Kafka topics
    '''

    def __init__(self):

        self.appName = "kafkatest"

        self.kafkaHosts = "192.168.4.201:6667,192.168.4.231:6667"
        self.kafkaAutoOffsetReset = "largest"
        self.kafka_offset_redis_db = 6
        self._kafka_topic = "test"
        self.redisHost = "192.168.4.231"
        self.redisPort = 6379

        self.spark_batch_duration = 20

    def createStreamingContext(self, sc):
        ssc = StreamingContext(sc, self.spark_batch_duration)
        ds = self.getDStreamFromKafka(ssc)
        if ds is not None:
            elogging.info(self.appName, elogging.normalCID(), "Kafka succeeded to getting the data")
            return ssc, ds
        else:
            return None, None

    def getDStreamFromKafka(self, ssc):
        kafkaParams = {"metadata.broker.list": self.kafkaHosts}
        elogging.debug(self.appName, elogging.normalCID(), kafkaParams)

        sc = ssc.sparkContext
        dstream = None
        try:
            redisConn = self.getRedisConnection(self.kafka_offset_redis_db)
            if redisConn.exists(self.appName):
                elogging.debug(self.appName, elogging.normalCID(), "key " + self.appName + " exists in redis")
                fromOffset = {}
                offsetListStr = redisConn.get(self.appName)
                offsetList = 
                for offset in offsetList:
                    elogging.debug(self.appName, elogging.normalCID(), str(offset))
                    topicPartion = TopicAndPartition(offset["topic"], offset["partition"])
                    fromOffset[topicPartion] = offset["untilOffset"]
                dstream = KafkaUtils.createDirectStream(ssc, [self._kafka_topic], kafkaParams, fromOffset)
            else:
                kafkaParams = {"metadata.broker.list": self.kafkaHosts, "auto.offset.reset": self.kafkaAutoOffsetReset}
                elogging.debug(self.appName, elogging.normalCID(), "key " + self.appName + " doesn't exist in redis")
                dstream = KafkaUtils.createDirectStream(ssc, [self._kafka_topic], kafkaParams)
        except:
            traceInfo = traceback.format_exc()
            elogging.error(self.appName, elogging.faultCID(), "failed to create DStream : " + traceInfo)

        return dstream

    def getRedisConnection(self, redisDB):
        try:
            pool = redis.ConnectionPool(host=self.redisHost, port=self.redisPort, db=redisDB)
            redisConn = redis.Redis(connection_pool=pool)
        except:
            traceInfo = traceback.format_exc()
            elogging.error(self.appName, elogging.faultCID(), "failed to create DStream : " + traceInfo)
            return None

        return redisConn

    def getOffSetRangesFromRDD(self, rdd):
        try:
            offsetRanges = rdd.offsetRanges()
        except:
            traceInfo = traceback.format_exc()
            elogging.error(self.appName, elogging.faultCID(), "failed to call rdd.offsetRanges() function : " + traceInfo)
            return None

        offsetList = []
        for offset in offsetRanges:
            offsetList.append({"topic": offset.topic, "partition": offset.partition, "fromOffset": offset.fromOffset,
                               "untilOffset": offset.untilOffset})

        elogging.info(self.appName, elogging.normalCID(), "getOffSetRangesFromRDD, offsetList: " + str(offsetList))
        return offsetList

    def saveOffSetRangesToRedis(self, offsetList):
        redisConn = self.getRedisConnection(self.kafka_offset_redis_db)
        if redisConn is not None:
            redisConn.set(self.appName, offsetList)
            elogging.info(self.appName, elogging.normalCID(), "saveOffSetRangesToRedis, offsetList : " + str(offsetList))

    def handleMessages(self, runTime, rdd):
        elogging.debug(self.appName, elogging.normalCID(), "========= %s =========" % str(runTime))
        offsetList = self.getOffSetRangesFromRDD(rdd)
        if offsetList is not None:
            self.saveOffSetRangesToRedis(offsetList)

        rddFilter = rdd.map(lambda p: p[1])

        counts = rddFilter.flatMap(lambda line: line.split(" ")) \
            .map(lambda word: (word, 1)) \
            .reduceByKey(lambda a, b: a + b)
        sqlContext = getSqlContextInstance(rddFilter.context)
        if counts is not None:
            df = sqlContext.createDataFrame(counts)
            df.show()

    def start(self):
        reload(sys)
        sys.setdefaultencoding('utf-8')
        sc = SparkContext(appName=self.appName)
        eloggingConfig = None
        try:
            eloggingConfig = HDFSOperation.getConfigFromHDFS(ELogForDistributedApp.LOGHDFSPATH, sc)
            elogging.initLogFromDict(eloggingConfig)
        except StandardError, se:
            pass

        elogging.debug(self.appName, elogging.normalCID(), "receiver starting")
        configInfoStr = 'kafkaHosts:' + str(self.kafkaHosts) + ', kafkaAutoOffsetReset:' + str(self.kafkaAutoOffsetReset) + \
                        ', kafka_offset_redis_db:' + str(self.kafka_offset_redis_db) + ', spark_batch_duration:' + str(self.spark_batch_duration) + \
                        ', redisHost:' + str(self.redisHost) + ', redisPort:' + str(self.redisPort)
        elogging.info(self.appName, elogging.normalCID(), configInfoStr)
        ssc, newDS = self.createStreamingContext(sc)
        if newDS is not None:
            newDS.foreachRDD(self.handleMessages)
            ssc.start()
            elogging.debug(self.appName, elogging.normalCID(), "StreamingContext start")
            ssc.awaitTermination()
            elogging.debug(self.appName, elogging.normalCID(), "receiver end")
        else:
            traceInfo = traceback.format_exc()
            elogging.error(self.appName, elogging.faultCID(), "Failed to create DStream " + traceInfo)

if __name__ =="__main__":
    test = KafkaStreamTest()
    test.start()