راهنمای به‌کارگیری Kafka Python برای پردازش سریع داده‌ها

$ cd kafka/
$ ./kafka-mesos.sh topic add admintome-pages --broker=0 --api=http://mslave2.admintome.lab:7000

$ ./kafka-mesos.sh topic list --api=http://mslave2.admintome.lab:7000
topics:
name: __consumer_offsets
partitions: 0:[0], 1:[0], 2:[0], 3:[0], 4:[0], 5:[0], 6:[0], 7:[0], 8:[0], 9:[0], 10:[0], 11:[0], 12:[0], 13:[0], 14:[0], 15:[0], 16:[0], 17:[0], 18:[0], 19:[0], 20:[0], 21:[0], 22:[0], 23:[0], 24:[0], 25:[0], 26:[0], 27:[0], 28:[0], 29:[0], 30:[0], 31:[0], 32:[0], 33:[0], 34:[0], 35:[0], 36:[0], 37:[0], 38:[0], 39:[0], 40:[0], 41:[0], 42:[0], 43:[0], 44:[0], 45:[0], 46:[0], 47:[0], 48:[0], 49:[0]
options: segment.bytes=104857600,cleanup.policy=compact,compression.type=producer
name: admintome
partitions: 0:[0]
name: admintome-pages
partitions: 0:[0]

$ mkdir ~/Development/python/venvs
$ mkdir ~/Development/python/site-stats-intake
$ cd ~/Development/python/site-stats-intake
$ virtualenv ../venvs/intake
$ source ../venvs/intake/bin/activate
(intake) $ pip install kafka-python requests
(intake) $ pip freeze > requirements.txt

import requests
import json
class Clicky(object):
    def __init__(self, site_id, sitekey):
        self.site_id = site_id
        self.sitekey = sitekey
        self.output = “json”
    def get_data(self, data_type):
        click_api_url = “https://api.clicky.com/api/stats/4”
        payload = {“site_id”: self.site_id,
                   “sitekey”: self.sitekey,
                   “type”: data_type,
                   “output”: self.output}
        response = requests.get(click_api_url, params=payload)
        raw_stats = response.text
        return raw_stats
    def get_pages_data(self):
        data = self.get_data(“pages”)
        return json.loads(data)

from kafka import KafkaProducer
import json
class MyKafka(object):
    def __init__(self, kafka_brokers):
        self.producer = KafkaProducer(
            value_serializer=lambda v: json.dumps(v).encode(‘utf-8’),
            bootstrap_servers=kafka_brokers
        )
    def send_page_data(self, json_data):
        self.producer.send(‘admintome-pages’, json_data)

from kafka import KafkaProducer

class MyKafka(object):
    def __init__(self, kafka_brokers):

[ “broker:ip”, “broker:ip” ]

[ “mslave1.admintome.lab:31000” ]

self.producer = KafkaProducer(
            value_serializer=lambda v: json.dumps(v).encode(‘utf-8’),
            bootstrap_servers=kafka_brokers
        )

def send_page_data(self, json_data):
    self.producer.send(‘admintome-pages’, json_data)

from clicky import Clicky
from mykafka import MyKafka
import logging
import time
import os
from logging.config import dictConfig
class Main(object):
    def __init__(self):
        if ‘KAFKA_BROKERS’ in os.environ:
            kafka_brokers = os.environ[‘KAFKA_BROKERS’].split(‘,’)
        else:
            raise ValueError(‘KAFKA_BROKERS environment variable not set’)
        if ‘SITE_ID’ in os.environ:
            self.site_id = os.environ[‘SITE_ID’]
        else:
            raise ValueError(‘SITE_ID environment variable not set’)
        if ‘SITEKEY’ in os.environ:
            self.sitekey = os.environ[‘SITEKEY’]
        else:
            raise ValueError(‘SITEKEY environment variable not set’)
        logging_config = dict(
            version=1,
            formatters={
                ‘f’: {‘format’:
                      ‘%(asctime)s %(name)-12s %(levelname)-8s %(message)s’}
            },
            handlers={
                ‘h’: {‘class’: ‘logging.StreamHandler’,
                      ‘formatter’: ‘f’,
                      ‘level’: logging.DEBUG}
            },
            root={
                ‘handlers’: [‘h’],
                ‘level’: logging.DEBUG,
            },
        )
        self.logger = logging.getLogger()
        dictConfig(logging_config)
        self.logger.info(“Initializing Kafka Producer”)
        self.logger.info(“KAFKA_BROKERS={0}”.format(kafka_brokers))
        self.mykafka = MyKafka(kafka_brokers)
    def init_clicky(self):
        self.clicky = Clicky(self.site_id, self.sitekey)
        self.logger.info(“Clicky Stats Polling Initialized”)
    def run(self):
        self.init_clicky()
        starttime = time.time()
        while True:
            data = self.clicky.get_pages_data()
            self.logger.info(“Successfully polled Clicky pages data”)
            self.mykafka.send_page_data(data)
            self.logger.info(“Published page data to Kafka”)
            time.sleep(300.0 - ((time.time() - starttime) % 300.0))
if __name__ == “__main__”:
    logging.info(“Initializing Clicky Stats Polling”)
    main = Main()

    main.run()

 if ‘KAFKA_BROKERS’ in os.environ:
            kafka_brokers = os.environ[‘KAFKA_BROKERS’].split(‘,’)
        else:
            raise ValueError(‘KAFKA_BROKERS environment variable not set’)
        if ‘SITE_ID’ in os.environ:
            self.site_id = os.environ[‘SITE_ID’]
        else:
            raise ValueError(‘SITE_ID environment variable not set’)
        if ‘SITEKEY’ in os.environ:
            self.sitekey = os.environ[‘SITEKEY’]
        else:
            raise ValueError(‘SITEKEY environment variable not set’)

  def run(self):
        self.init_clicky()
        starttime = time.time()
        while True:
            data = self.clicky.get_pages_data()
            self.logger.info(“Successfully polled Clicky pages data”)
            self.mykafka.send_page_data(data)
            self.logger.info(“Published page data to Kafka”)
            time.sleep(300.0 - ((time.time() - starttime) % 300.0))

(intake) $ export KAFKA_BROKERS=”mslave1.admintome.lab:31000”
(intake) $ export SITE_ID=”{your site id}”
(intake) $ export SITEKEY=”{your sitekey}”
(intake) $ python main.py
2018-06-25 15:34:32,259 root INFO Initializing Kafka Producer
2018-06-25 15:34:32,259 root INFO KAFKA_BROKERS=[‘mslave1.admintome.lab:31000’]
2018-06-25 15:34:32,374 root INFO Clicky Stats Polling Initialized
2018-06-25 15:34:32,754 root INFO Successfully polled Clicky pages data
2018-06-25 15:34:32,755 root INFO Published page data to Kafka

       https://github.com/admintome/clicky-state-intake

FROM python:3
WORKDIR /usr/src/app
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD [ “python”, “./main.py” ]

$ docker build -t  {your docker hub username}site-stats-intake .

$ docker push -t admintome/site-stats-intake

$ docker pull admintome/site-stats-intake

{
  “id”: “site-stats-intake”,
  “cmd”: null,
  “cpus”: 1,
  “mem”: 128,
  “disk”: 0,
  “instances”: 1,
  “container”: {
    “docker”: {
      “image”: “admintome/site-stats-intake”
    },
    “type”: “DOCKER”
  },
  “networks”: [
    {
      “mode”: “host”
    }
  ],
  “env”: {
    “KAFKA_BROKERS”: “192.168.1.x:port”,
    “SITE_ID”: “{your site_id}”,
    “SITEKEY”: “{your sitekey}”
  }
}

import sys
from kafka import KafkaConsumer
consumer = KafkaConsumer(‘admintome-pages’, bootstrap_servers=”mslave1.admintome.lab:31000”,
                         auto_offset_reset=’earliest’)
try:
    for message in consumer:
        print(message.value)
except KeyboardInterrupt:
    sys.exit()

(intake) $ python consumer.py
b’[{“type”: “pages”, “dates”: [{“date”: “2018-06-25”, “items”: [{“value”: “145”, “value_percent”: “43.2”, “title”: “Kafka Tutorial for Fast Data Architecture - AdminTome Blog”, “stats_url”: “http://clicky.com/stats/visitors?site_id=101045340&date=2018-06-25&href=%2Fblog%2Fkafka-tutorial-for-fast-data-architecture%2F”, “url”: “http://www.admintome.com/blog/kafka-tutorial-for-fast-data-architecture/”},...