Real-time Streams & Logs

Andrew Montalenti, CTO

Keith Bourgoin, Backend Lead

Agenda

• Parse.ly problem space
• Aggregating the stream (Storm)
• Organizing around logs (Kafka)

Admin

Our presentations and code:

http://parse.ly/code

This presentation's slides:

http://parse.ly/slides/logs

This presentation's notes:

http://parse.ly/slides/logs/notes

What is Parse.ly?

What is Parse.ly?

Web content analytics for digital storytellers.

Velocity

Average post has <48-hour shelf life.

Volume

Top publishers write 1000's of posts per day.

Time series data

Summary data

Ranked data

Benchmark data

Information radiators

Architecture evolution

Queues and workers

Queues: RabbitMQ => Redis => ZeroMQ

Workers: Cron Jobs => Celery

Workers and databases

Lots of moving parts

In short: it started to get messy

Introducing Storm

Storm is a distributed real-time computation system.

Hadoop provides a set of general primitives for doing batch processing.

Storm provides a set of general primitives for doing real-time computation.

Perfect as a replacement for ad-hoc workers-and-queues systems.

Storm features

• Speed
• Fault tolerance
• Parallelism
• Guaranteed Messages
• Easy Code Management
• Local Dev

Storm primitives

Streaming Data Set, typically from Kafka.

ZeroMQ used for inter-process communication.

Bolts & Spouts; Storm's Topology is a DAG.

Nimbus & Workers manage execution.

Tuneable parallelism + built-in fault tolerance.

Wired Topology

Tuple Tree

Tuple tree, anchoring, and retries.

Word Stream Spout (Storm)

;; spout configuration
{"word-spout" (shell-spout-spec
      ;; Python Spout implementation:
      ;; - fetches words (e.g. from Kafka)
        ["python" "words.py"]
      ;; - emits (word,) tuples
        ["word"]
      )
}

Word Stream Spout in Python

import itertools

from streamparse import storm

class WordSpout(storm.Spout):

    def initialize(self, conf, ctx):
        self.words = itertools.cycle(['dog', 'cat',
                                      'zebra', 'elephant'])

    def next_tuple(self):
        word = next(self.words)
        storm.emit([word])

WordSpout().run()

Word Count Bolt (Storm)

;; bolt configuration
{"count-bolt" (shell-bolt-spec
       ;; Bolt input: Spout and field grouping on word
         {"word-spout" ["word"]}
       ;; Python Bolt implementation:
       ;; - maintains a Counter of word
       ;; - increments as new words arrive
         ["python" "wordcount.py"]
       ;; Emits latest word count for most recent word
         ["word" "count"]
       ;; parallelism = 2
         :p 2
       )
}

Word Count Bolt in Python

from collections import Counter

from streamparse import storm

class WordCounter(storm.Bolt):

    def initialize(self, conf, ctx):
        self.counts = Counter()

    def process(self, tup):
        word = tup.values[0]
        self.counts[word] += 1
        storm.emit([word, self.counts[word]])
        storm.log('%s: %d' % (word, self.counts[word]))

WordCounter().run()

streamparse

sparse provides a CLI front-end to streamparse, a framework for creating Python projects for running, debugging, and submitting Storm topologies for data processing. (still in development)

After installing the lein (only dependency), you can run:

pip install streamparse

This will offer a command-line tool, sparse. Use:

sparse quickstart

Running and debugging

You can then run the local Storm topology using:

$ sparse run
Running wordcount topology...
Options: {:spec "topologies/wordcount.clj", ...}
#<StormTopology StormTopology(spouts:{word-spout=...
storm.daemon.nimbus - Starting Nimbus with conf {...
storm.daemon.supervisor - Starting supervisor with id 4960ac74...
storm.daemon.nimbus - Received topology submission with conf {...
... lots of output as topology runs...

Interested? Lightning talk!

Organizing around logs

Not all logs are application logs

A "log" could be any stream of structured data:

• Web logs
• Raw data waiting to be processed
• Partially processed data
• Database operations (e.g. mongo's oplog)

A series of timestamped facts about a given system.

LinkedIn's lattice problem

Enter the unified log

Log-centric is simpler

Parse.ly is log-centric, too

Introducing Apache Kafka

Log-centric messaging system developed at LinkedIn.

Designed for throughput; efficient resource use.

• Persists to disk; in-memory for recent data
• Little to no overhead for new consumers
• Scalable to 10,000's of messages per second

As of 0.8, full replication of topic data.

Kafka concepts

What's the catch?

Replication isn't perfect. Network partitions can cause problems.

No out-of-order acknowledgement:

• "Offset" is a marker of where consumer is in log; nothing more.
• On a restart, you know where to start reading, but not if individual messages before the stored offset was fully processed.
• In practice, not as much of a problem as it sounds.

Kafka is a "distributed log"

Topics are logs, not queues.

Consumers read into offsets of the log.

Logs are maintained for a configurable period of time.

Messages can be "replayed".

Consumers can share identical logs easily.

Multi-consumer

Even if Kafka's availability and scalability story isn't interesting to you, the multi-consumer story should be.

Queue problems, revisited

Traditional queues (e.g. RabbitMQ / Redis):

• not distributed / highly available at core
• not persistent ("overflows" easily)
• more consumers mean more queue server load

Kafka solves all of these problems.

Kafka + Storm

Good fit for at-least-once processing.

No need for out-of-order acks.

Community work is ongoing for at-most-once processing.

Able to keep up with Storm's high-throughput processing.

Great for handling backpressure during traffic spikes.

Kafka in Python (1)

python-kafka (0.8+)

• https://github.com/mumrah/kafka-python

from kafka.client import KafkaClient
from kafka.consumer import SimpleConsumer

kafka = KafkaClient('localhost:9092')
consumer = SimpleConsumer(kafka, 'test_consumer', 'raw_data')
start = time.time()
for msg in consumer:
    count += 1
    if count % 1000 == 0:
        dur = time.time() - start
        print 'Reading at {:.2f} messages/sec'.format(dur/1000)
        start = time.time()

Kafka in Python (2)

samsa (0.7x)

• https://github.com/getsamsa/samsa

import time
from kazoo.client import KazooClient
from samsa.cluster import Cluster

zk = KazooClient()
zk.start()
cluster = Cluster(zk)
queue = cluster.topics['raw_data'].subscribe('test_consumer')
start = time.time()
for msg in queue:
    count += 1
    if count % 1000 == 0:
        dur = time.time() - start
        print 'Reading at {:.2f} messages/sec'.format(dur/1000)
        queue.commit_offsets() # commit to zk every 1k msgs

Other Log-Centric Companies

Conclusion

What we've learned

• There is no silver bullet data processing technology.
• Log storage is very cheap, and getting cheaper.
• "Timestamped facts" is rawest form of data available.
• Storm and Kafka allow you to develop atop those facts.
• Organizing around real-time logs is a wise decision.

Questions?

Go forth and stream!

Parse.ly:

• http://parse.ly/code
• http://twitter.com/parsely

Andrew & Keith:

• http://twitter.com/amontalenti
• http://twitter.com/kbourgoin