Apache Storm with Python

環(huán)境:


  • 系統(tǒng):centos7
    ps: 請(qǐng)確認(rèn)kafka漂彤,zookeeper弦牡,storm部署完成(本文基于Apache ambari搭建的一個(gè)集群兼都,進(jìn)行測(cè)試)

  • 安裝包:

    1. $ yum install -y gcc python-devel java cyrus-sasl-devel cyrus-sasl-gssapi cyrus-sasl-md5 cyrus-sasl-plain librdkafka-devel redis
    2. Install lein
      $ wget https://raw.githubusercontent.com/technomancy/leiningen/stable/bin/lein
      $ mv lein /usr/bin/
      $ chmod a+x /usr/bin/lein
      $ wget https://github.com/technomancy/leiningen/releases/download/2.8.1/leiningen-2.8.1-standalone.zip
      $ mv leiningen-2.8.1-standalone.zip /root/.lein/self-installs/leiningen-2.8.1-standalone.jar
      $ export LEIN_ROOT = 1
      $ lein version # test lein version
      image.png
  • Create virtualenv
    $ pip install streamparse confluent-kafka redis kazoo

整體架構(gòu)


image.png

Start demo


  • get kafka brokers

    1. find zookeeper cluster(through Ambari)


      image.png
    2. get brokers
    from kazoo.client import KazooClient
    import json
    
    
    def get_kafka_brokers(host):
        zookeeper = KazooClient(hosts=host, read_only=True)
        zookeeper.start()
        for node in zookeeper.get_children('/brokers/ids'):
            data, stats = zookeeper.get('/brokers/ids/'+node)
            props = json.loads(data)
            yield props['host']+':'+str(props['port'])
        zookeeper.stop()
    
    
    if __name__ == "__main__":
        print ','.join(get_kafka_brokers("cluster1.dc.com, cluster2.dc.com"))
    
    

    輸出: cluster2.dc.com:6667
    通過Ambari 確認(rèn)kafka集群,如圖

    image.png

  • producer往brokers生產(chǎn)數(shù)據(jù)(用了confluent-kafka)

    # -*- coding:utf-8 -*-
    import confluent_kafka
    import random, time
    import json
    from get_broker_list import get_kafka_brokers
    
    
    def error_cb(err):
        print('Error: %s' % err)
    
    
    def main():
        # bootstrap_servers = 'cluster2.dc.com:6667'
        zk_host = 'cluster1.dc.com,cluster2.dc.com'
        bootstrap_servers = ','.join(get_kafka_brokers(zk_host))
        api_version_request = True
        conf = {'bootstrap.servers': bootstrap_servers,
                'api.version.request': api_version_request,
                'error_cb': error_cb,
                'debug': 'protocol',
                'broker.address.family': 'v4'}
        producer = confluent_kafka.Producer(**conf)
        user_list = ['jason', 'jane', 'tom', 'jack']
        while True:
            data = {"user": random.choice(user_list),
                    "timestamp": time.time(),
                    "log_level": random.randint(0, 5)
                   }
            try:
                producer.produce('test', value=json.dumps(data))
                # time.sleep(random.randint(1, 2))
            except BufferError:
                producer.poll(100)
            continue
        producer.flush()
    
    
    if __name__ == '__main__':
        main()
    

    部分結(jié)果如圖

    image.png

  • 創(chuàng)建一個(gè)consumer進(jìn)行驗(yàn)證:

    #!/usr/bin/env python
    import time
    import json
    from confluent_kafka import Consumer, KafkaException, KafkaError
    from get_broker_list import get_kafka_brokers
    
    
    def main():
        # broker = 'cluster2.dc.com:6667'
        zk_host = 'cluster1.dc.com,cluster2.dc.com'
        bootstrap_servers = ','.join(get_kafka_brokers(zk_host))
        group = 'test.py'
        conf = {'bootstrap.servers': bootstrap_servers, 'group.id': group, 'session.timeout.ms': 6000,
                'default.topic.config': {'auto.offset.reset': 'smallest'}}
        consumer = Consumer(**conf)
        consumer.subscribe(['test'])
        while True:
            msg = consumer.poll()
            try:
                print json.loads(msg.value())
            except Exception:
                time.sleep(1)
            continue
        consumer.close()
    
    if __name__ == '__main__':
        main()
    

    部分結(jié)果如圖

    image.png

  • integrate with Storm(use package streamparse

    上面kafka producer產(chǎn)生了一條用戶記錄套媚,storm demo以計(jì)算5分鐘內(nèi)產(chǎn)生了多少條記錄(實(shí)際效果producer >> consumer缚态,所以導(dǎo)致延遲問題,測(cè)試數(shù)據(jù)大概5分鐘寫入150w-180w條凑阶,資源限制導(dǎo)致的性能問題猿规,僅供參考)

  1. sparse quickstart onlineuser

項(xiàng)目結(jié)構(gòu)如下:
image.png

其中topologies,bolts宙橱,以及spouts中的文件名可能是wordcount相關(guān)命名姨俩,修改或不修改均可,只需要確認(rèn)topologies文件中的topology能與spouts中的spout师郑,bolts中的bolt對(duì)應(yīng)起來即可

  1. vim spout/user.py

    import sys, os
    # sys.path.append(os.path.dirname(os.path.abspath(__file__)) + '/../../../kafka_example')
    abspath = "" # must fill with abs path, cannot use os.path.abspath, run as jar in /tmp directory
    if not abspath:
        raise Exception("setting kafka_exmaple directory abspath to import get_broker_list")
    sys.path.append(abspath)
    from confluent_kafka import Consumer
    from streamparse import Spout
    from get_broker_list import get_kafka_brokers
    
    
    class OnlineUserSpout(Spout):
        outputs = ['log']
    
        def initialize(self, stormconf, context):
            # broker = 'cluster2.dc.com:6667'
            zk_host = 'cluster1.dc.com,cluster2.dc.com'
            broker = ','.join(get_kafka_brokers(zk_host))
            group = 'test.py'
            conf = {'bootstrap.servers': broker, 'group.id': group, 'session.timeout.ms': 6000,
                    'default.topic.config': {'auto.offset.reset': 'smallest'}}
            self.consumer = Consumer(**conf)
    
    
        def activate(self):
            self.consumer.subscribe(['test'])
    
        def next_tuple(self):
            msg = self.consumer.poll()
            if msg.value():
                self.emit([msg.value()])
    
        def deactivate(self):
            self.consumer.close()
    
    
  2. vim bolts/serializer_log.py這部分沒有考慮用戶重復(fù)問題

    import json
    import time
    from datetime import datetime, timedelta
    from redis import StrictRedis
    from streamparse import Bolt
    
    
    class RedisLog(Bolt):
    
    
        def initialize(self, conf, ctx):
            self.redis = StrictRedis()
            self.interval_minute = 5
    
        def _increment(self, duration):
            return self.redis.incr(duration)
    
    
        def process(self, tup):
            data = json.loads(tup.values[0])
            user = data['user'] # useless
            timestamp = data["timestamp"]
            now = datetime.fromtimestamp(int(timestamp))
            now = now - timedelta(minutes=now.minute % self.interval_minute,
                              seconds=now.second, microseconds=now.microsecond)
            now_timestamp = int(time.mktime(now.timetuple()))
            duration = '{0}-{1}'.format(now_timestamp, now_timestamp + self.interval_minute * 60)
            count = self._increment(duration)
            self.emit([duration, count])
    
    
  3. vim topologies/onlineuser.py

    """
    Online User topology
    """
    from streamparse import Topology
    from bolts.serializer_log import RedisLog
    from spouts.user import OnlineUserSpout
    
    
    class OnlineUserCount(Topology):
        log_spout = OnlineUserSpout.spec()
        count_bolt = RedisLog.spec(inputs=[log_spout])
    
    
  4. $ sparse run # 必須在sparse quickstart 項(xiàng)目路徑下(耗時(shí)較久环葵,需要build成jar到/tmp下執(zhí)行)
    部分結(jié)果如圖(可能有一些warn,這是由于zookeeper日記文件相關(guān)寫入延遲宝冕,會(huì)影響storm性能张遭,測(cè)試先忽略)

    image.png

  5. 可以通過redis檢測(cè)key value(key是以時(shí)間戳區(qū)間,整形地梨,格式 'timestamp1-timestamp2')

    import time
    from redis import StrictRedis
    redis = StrictRedis()
    while 1:
        keys = redis.keys()
        vals = redis.mget(keys)
        kv = zip(keys, vals)
        print kv
        time.sleep(10)
    

    result:大致如圖


    image.png
  6. 可能出現(xiàn)的一些問題解決辦法:

    • 運(yùn)行sparse run 時(shí)菊卷,爆storm版本不一致問題,修改project.clj宝剖,由于可能storm也是通過ambari進(jìn)行安裝洁闰,輸出版本的格式不一致(Hortonworks data platform 版本號(hào),類似‘1.1.0.2.6.2.0-205’万细,這時(shí)候需要去vim xxx/xxx/site-packages/streamparse/cli/run.py 大概48,49修改一下判斷or去掉檢測(cè))


      image.png
    • 運(yùn)行sparse run時(shí)扑眉,可能出現(xiàn)NoClassDefFoundError: org/apache/commons/lang/StringUtils.
      解決的辦法
      wget https://www.apache.org/dist/commons/lang/binaries/commons-lang-2.6-bin.zip.md5
      unzip commons-lang-2.6-bin.zip
      cd commons-lang-2.6-bin.zip && mv commons-lang.jar storm/lib
      

Ending


整個(gè)過程中,可能還會(huì)出現(xiàn)一些issue赖钞,可以到對(duì)應(yīng)的項(xiàng)目去查看文檔腰素。
Finally,本文原創(chuàng)雪营,未經(jīng)許可弓千,謝絕轉(zhuǎn)載。=_=!

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
  • 序言:七十年代末献起,一起剝皮案震驚了整個(gè)濱河市计呈,隨后出現(xiàn)的幾起案子砰诵,更是在濱河造成了極大的恐慌,老刑警劉巖捌显,帶你破解...
    沈念sama閱讀 222,627評(píng)論 6 517
  • 序言:濱河連續(xù)發(fā)生了三起死亡事件,死亡現(xiàn)場(chǎng)離奇詭異总寒,居然都是意外死亡扶歪,警方通過查閱死者的電腦和手機(jī),發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 95,180評(píng)論 3 399
  • 文/潘曉璐 我一進(jìn)店門摄闸,熙熙樓的掌柜王于貴愁眉苦臉地迎上來善镰,“玉大人,你說我怎么就攤上這事年枕§牌郏” “怎么了?”我有些...
    開封第一講書人閱讀 169,346評(píng)論 0 362
  • 文/不壞的土叔 我叫張陵熏兄,是天一觀的道長(zhǎng)品洛。 經(jīng)常有香客問我,道長(zhǎng)摩桶,這世上最難降的妖魔是什么桥状? 我笑而不...
    開封第一講書人閱讀 60,097評(píng)論 1 300
  • 正文 為了忘掉前任,我火速辦了婚禮硝清,結(jié)果婚禮上辅斟,老公的妹妹穿的比我還像新娘。我一直安慰自己芦拿,他們只是感情好士飒,可當(dāng)我...
    茶點(diǎn)故事閱讀 69,100評(píng)論 6 398
  • 文/花漫 我一把揭開白布。 她就那樣靜靜地躺著蔗崎,像睡著了一般酵幕。 火紅的嫁衣襯著肌膚如雪。 梳的紋絲不亂的頭發(fā)上蚁趁,一...
    開封第一講書人閱讀 52,696評(píng)論 1 312
  • 那天裙盾,我揣著相機(jī)與錄音,去河邊找鬼他嫡。 笑死番官,一個(gè)胖子當(dāng)著我的面吹牛,可吹牛的內(nèi)容都是我干的钢属。 我是一名探鬼主播徘熔,決...
    沈念sama閱讀 41,165評(píng)論 3 422
  • 文/蒼蘭香墨 我猛地睜開眼,長(zhǎng)吁一口氣:“原來是場(chǎng)噩夢(mèng)啊……” “哼淆党!你這毒婦竟也來了酷师?” 一聲冷哼從身側(cè)響起讶凉,我...
    開封第一講書人閱讀 40,108評(píng)論 0 277
  • 序言:老撾萬榮一對(duì)情侶失蹤,失蹤者是張志新(化名)和其女友劉穎山孔,沒想到半個(gè)月后懂讯,有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體,經(jīng)...
    沈念sama閱讀 46,646評(píng)論 1 319
  • 正文 獨(dú)居荒郊野嶺守林人離奇死亡台颠,尸身上長(zhǎng)有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點(diǎn)故事閱讀 38,709評(píng)論 3 342
  • 正文 我和宋清朗相戀三年褐望,在試婚紗的時(shí)候發(fā)現(xiàn)自己被綠了。 大學(xué)時(shí)的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片串前。...
    茶點(diǎn)故事閱讀 40,861評(píng)論 1 353
  • 序言:一個(gè)原本活蹦亂跳的男人離奇死亡瘫里,死狀恐怖,靈堂內(nèi)的尸體忽然破棺而出荡碾,到底是詐尸還是另有隱情谨读,我是刑警寧澤,帶...
    沈念sama閱讀 36,527評(píng)論 5 351
  • 正文 年R本政府宣布坛吁,位于F島的核電站劳殖,受9級(jí)特大地震影響,放射性物質(zhì)發(fā)生泄漏阶冈。R本人自食惡果不足惜闷尿,卻給世界環(huán)境...
    茶點(diǎn)故事閱讀 42,196評(píng)論 3 336
  • 文/蒙蒙 一、第九天 我趴在偏房一處隱蔽的房頂上張望女坑。 院中可真熱鬧填具,春花似錦、人聲如沸匆骗。這莊子的主人今日做“春日...
    開封第一講書人閱讀 32,698評(píng)論 0 25
  • 文/蒼蘭香墨 我抬頭看了看天上的太陽碉就。三九已至盟广,卻和暖如春,著一層夾襖步出監(jiān)牢的瞬間瓮钥,已是汗流浹背筋量。 一陣腳步聲響...
    開封第一講書人閱讀 33,804評(píng)論 1 274
  • 我被黑心中介騙來泰國打工, 沒想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留碉熄,地道東北人桨武。 一個(gè)月前我還...
    沈念sama閱讀 49,287評(píng)論 3 379
  • 正文 我出身青樓,卻偏偏與公主長(zhǎng)得像锈津,于是被迫代替她去往敵國和親呀酸。 傳聞我的和親對(duì)象是個(gè)殘疾皇子,可洞房花燭夜當(dāng)晚...
    茶點(diǎn)故事閱讀 45,860評(píng)論 2 361

推薦閱讀更多精彩內(nèi)容