[PROJECT IS NO LONGER MAINTAINED] Code examples that show to integrate Apache Kafka 0.8+ with Apache Storm 0.9+ and Apache Spark Streaming 1.1+, while using Apache Avro as the data serialization format.
Code examples that show how to integrate Apache Kafka 0.8+ with Apache Storm 0.9+ and Apache Spark 1.1+ while using Apache Avro as the data serialization format.
A great alternative to the examples in this repository, which require you to operate a Spark or Storm processing cluster: build elastic, distributed, fault-tolerant stream processing applications with Kafka's Streams API (read: no additional cluster required)
"Kafka Streams (source code), a component of open source Apache Kafka, is a powerful, easy-to-use library for building highly scalable, fault-tolerant, distributed stream processing applications on top of Apache Kafka. It builds upon important concepts for stream processing such as properly distinguishing between event-time and processing-time, handling of late-arriving data, and efficient management of application state."
Take a look at the Kafka Streams code examples at https://github.com/confluentinc/examples.
Table of Contents
$ ./sbt test
This command launches our test suite.
Notably it will run end-to-end tests of Kafka, Storm, and Kafka/Storm as well as Kafka/Spark Streaming integration. See this abridged version of the test output:
[...other tests removed...]
[info] KafkaSpec:
[info] Kafka
[info] - should synchronously send and receive a Tweet in Avro format
[info] + Given a ZooKeeper instance
[info] + And a Kafka broker instance
[info] + And some tweets
[info] + And a single-threaded Kafka consumer group
[info] + When I start a synchronous Kafka producer that sends the tweets in Avro binary format
[info] + Then the consumer app should receive the tweets
[info] - should asynchronously send and receive a Tweet in Avro format
[info] + Given a ZooKeeper instance
[info] + And a Kafka broker instance
[info] + And some tweets
[info] + And a single-threaded Kafka consumer group
[info] + When I start an asynchronous Kafka producer that sends the tweets in Avro binary format
[info] + Then the consumer app should receive the tweets
[info] StormSpec:
[info] Storm
[info] - should start a local cluster
[info] + Given no cluster
[info] + When I start a LocalCluster instance
[info] + Then the local cluster should start properly
[info] - should run a basic topology
[info] + Given a local cluster
[info] + And a wordcount topology
[info] + And the input words alice, bob, joe, alice
[info] + When I submit the topology
[info] + Then the topology should properly count the words
[info] KafkaStormSpec:
[info] As a user of Storm
[info] I want to read Avro-encoded data from Kafka
[info] so that I can quickly build Kafka<->Storm data flows
[info] Feature: AvroDecoderBolt[T]
[info] Scenario: User creates a Storm topology that uses AvroDecoderBolt
[info] Given a ZooKeeper instance
[info] And a Kafka broker instance
[info] And a Storm topology that uses AvroDecoderBolt and that reads tweets from topic testing-input} and writes them as-is to topic testing-output
[info] And some tweets
[info] And a synchronous Kafka producer app that writes to the topic testing-input
[info] And a single-threaded Kafka consumer app that reads from topic testing-output and Avro-decodes the incoming data
[info] And a Storm topology configuration that registers an Avro Kryo decorator for Tweet
[info] When I run the Storm topology
[info] And I Avro-encode the tweets and use the Kafka producer app to sent them to Kafka
[info] Synchronously sending Tweet {"username": "ANY_USER_1", "text": "ANY_TEXT_1", "timestamp": 1411993272} to topic Some(testing-input)
[info] Synchronously sending Tweet {"username": "ANY_USER_2", "text": "ANY_TEXT_2", "timestamp": 0} to topic Some(testing-input)
[info] Synchronously sending Tweet {"username": "ANY_USER_3", "text": "ANY_TEXT_3", "timestamp": 1234} to topic Some(testing-input)
[info] Then the Kafka consumer app should receive the original tweets from the Storm topology
[info] Feature: AvroScheme[T] for Kafka spout
[info] Scenario: User creates a Storm topology that uses AvroScheme in Kafka spout
[info] Given a ZooKeeper instance
[info] And a Kafka broker instance
[info] And a Storm topology that uses AvroScheme and that reads tweets from topic testing-input and writes them as-is to topic testing-output
[info] And some tweets
[info] And a synchronous Kafka producer app that writes to the topic testing-input
[info] And a single-threaded Kafka consumer app that reads from topic testing-output and Avro-decodes the incoming data
[info] And a Storm topology configuration that registers an Avro Kryo decorator for Tweet
[info] When I run the Storm topology
[info] And I Avro-encode the tweets and use the Kafka producer app to sent them to Kafka
[info] Synchronously sending Tweet {"username": "ANY_USER_1", "text": "ANY_TEXT_1", "timestamp": 1411993272} to topic Some(testing-input)
[info] Synchronously sending Tweet {"username": "ANY_USER_2", "text": "ANY_TEXT_2", "timestamp": 0} to topic Some(testing-input)
[info] Synchronously sending Tweet {"username": "ANY_USER_3", "text": "ANY_TEXT_3", "timestamp": 1234} to topic Some(testing-input)
[info] Then the Kafka consumer app should receive the original tweets from the Storm topology
[info] KafkaSparkStreamingSpec:
[info] As a user of Spark Streaming
[info] I want to read Avro-encoded data from Kafka
[info] so that I can quickly build Kafka<->Spark Streaming data flows
[info] Feature: Basic functionality
[info] Scenario: User creates a Spark Streaming job that reads from and writes to Kafka
[info] Given a ZooKeeper instance
[info] And a Kafka broker instance
[info] And some tweets
[info] And a synchronous Kafka producer app that writes to the topic KafkaTopic(testing-input,1,1,{})
[info] And a single-threaded Kafka consumer app that reads from topic KafkaTopic(testing-output,1,1,{}) and Avro-decodes the incoming data
[info] When I Avro-encode the tweets and use the Kafka producer app to sent them to Kafka
[info] And I run a streaming job that reads tweets from topic KafkaTopic(testing-input,1,1,{}) and writes them as-is to topic KafkaTopic(testing-output,1,1,{})
[info] Then the Spark Streaming job should consume all tweets from Kafka
[info] And the job should write back all tweets to Kafka
[info] And the Kafka consumer app should receive the original tweets from the Spark Streaming job
[info] Run completed in 45 seconds, 787 milliseconds.
[info] Total number of tests run: 27
[info] Suites: completed 9, aborted 0
[info] Tests: succeeded 27, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.
$ ./sbt run
This command launches KafkaStormDemo. This demo starts in-memory instances of ZooKeeper, Kafka, and Storm. It then runs a demo Storm topology that connects to and reads from the Kafka instance.
You will see output similar to the following (some parts removed to improve readability):
7031 [Thread-19] INFO backtype.storm.daemon.worker - Worker 3f7f1a51-5c9e-43a5-b431-e39a7272215e for storm kafka-storm-starter-1-1400839826 on daa60807-d440-4b45-94fc-8dd7798453d2:1027 has finished loading
7033 [Thread-29-kafka-spout] INFO storm.kafka.DynamicBrokersReader - Read partition info from zookeeper: GlobalPartitionInformation{partitionMap={0=127.0.0.1:9092}}
7050 [Thread-29-kafka-spout] INFO backtype.storm.daemon.executor - Opened spout kafka-spout:(1)
7051 [Thread-29-kafka-spout] INFO backtype.storm.daemon.executor - Activating spout kafka-spout:(1)
7051 [Thread-29-kafka-spout] INFO storm.kafka.ZkCoordinator - Refreshing partition manager connections
7065 [Thread-29-kafka-spout] INFO storm.kafka.DynamicBrokersReader - Read partition info from zookeeper: GlobalPartitionInformation{partitionMap={0=127.0.0.1:9092}}
7066 [Thread-29-kafka-spout] INFO storm.kafka.ZkCoordinator - Deleted partition managers: []
7066 [Thread-29-kafka-spout] INFO storm.kafka.ZkCoordinator - New partition managers: [Partition{host=127.0.0.1:9092, partition=0}]
7083 [Thread-29-kafka-spout] INFO storm.kafka.PartitionManager - Read partition information from: /kafka-spout/kafka-storm-starter/partition_0 --> null
7100 [Thread-29-kafka-spout] INFO storm.kafka.PartitionManager - No partition information found, using configuration to determine offset
7105 [Thread-29-kafka-spout] INFO storm.kafka.PartitionManager - Starting Kafka 127.0.0.1:0 from offset 18
7106 [Thread-29-kafka-spout] INFO storm.kafka.ZkCoordinator - Finished refreshing
At this point Storm is connected to Kafka (more precisely: to the testing
topic in Kafka). Not much will happen
afterwards because a) we are not sending any data to the Kafka topic and b) this demo Storm topology only reads from the
Kafka topic but it does nothing to the data that was read.
Note that this example will actually run two in-memory instances of ZooKeeper: the first (listening at
127.0.0.1:2181/tcp
) is used by the Kafka instance, the second (listening at 127.0.0.1:2000/tcp
) is automatically
started and used by the in-memory Storm cluster. This is because, when running in local aka in-memory mode, Storm
versions < 0.9.3 do not allow you to reconfigure or disable its own ZooKeeper instance (see the
Storm FAQ below for further information).
To stop the demo application you must kill or Ctrl-C
the process in the terminal.
You can use KafkaStormDemo as a starting point to create your own, "real" Storm topologies that read from a "real" Kafka, Storm, and ZooKeeper infrastructure. An easy way to get started with such an infrastructure is by deploying Kafka, Storm, and ZooKeeper via a tool such as Wirbelsturm.
What features do we showcase in kafka-storm-starter? Note that we focus on showcasing, and not necessarily on "production ready".
AvroDecoderBolt[T <: org.apache.avro.specific.SpecificRecordBase]
that can be parameterized with the type of
the Avro record T
it will deserialize its data to (i.e. no need to write another decoder bolt just because the
bolt needs to handle a different Avro schema).AvroScheme[T <: org.apache.avro.specific.SpecificRecordBase]
scheme, i.e. a custom
backtype.storm.spout.Scheme
to auto-deserialize a spout's incoming data. The scheme can be parameterized with
the type of the Avro record T
it will deserializes its data to (i.e. no need to write another scheme just
because the scheme needs to handle a different Avro schema).
AvroScheme
if you want to perform the Avro
decoding step directly in the spout instead of placing an AvroDecoderBolt
after the Kafka spout. You may
want to profile your topology which of the two approaches works best for your use case.backtype.storm.serialization.IKryoDecorator
, i.e. a custom
Kryo serializer for Storm.
AvroKryoDecorator[T]
variant yet.
(A "straight-forward" approach we tried -- similar to the other parameterized components -- compiled fine but
failed at runtime when running the tests). Code contributions are welcome!AvroKafkaSinkBolt[T <: org.apache.avro.specific.SpecificRecordBase]
that can be parameterized with the type
of the Avro record T
it will serialize its data to before sending the encoded data to Kafka (i.e. no
need to write another Kafka sink bolt just because the bolt needs to handle a different Avro schema).Tweet
from twitter.avsc.This project follows the git-flow approach. This means, for instance, that:
develop
is used for integration of the "next release".master
is used for bringing forth production releases.Follow the git-flow installation instructions for your development machine.
See git-flow and the introduction article Why aren't you using git-flow? for details.
Your development machine requires:
This project also needs Scala 2.10.4 and sbt 0.13.2, but these will be automatically downloaded and made available (locally/sandboxed) to the project as part of the build setup.
$ ./sbt clean compile
If you want to only (re)generate Java classes from Avro schemas:
$ ./sbt avro:generate
Generated Java sources are stored under target/scala-*/src_managed/main/compiled_avro/
.
$ ./sbt clean test
Here are some examples that demonstrate how you can run only a certain subset of tests:
# Use `-l` to exclude tests by tag:
# Run all tests WITH THE EXCEPTION of those tagged as integration tests
$ ./sbt "test-only * -- -l com.miguno.kafkastorm.integration.IntegrationTest"
# Use `-n` to include tests by tag (and skip all tests that lack the tag):
# Run ONLY tests tagged as integration tests
$ ./sbt "test-only * -- -n com.miguno.kafkastorm.integration.IntegrationTest"
# Run only the tests in suite AvroSchemeSpec:
$ ./sbt "test-only com.miguno.kafkastorm.storm.serialization.AvroSchemeSpec"
# You can also combine the examples above, of course.
Test reports in JUnit XML format are written to target/test-reports/junitxml/*.xml
. Make sure that your actual build
steps run the ./sbt test
task, otherwise the JUnit XML reports will not be generate (note that ./sbt scoverage:test
will not generate the JUnit XML reports unfortunately).
Integration with CI servers:
**/target/test-reports/junitxml/*.xml
.target/test-reports/junitxml/*.xml
Further details are available at:
We are using sbt-scoverage to create code coverage reports for unit tests.
Run the unit tests via:
$ ./sbt clean scoverage:test
target/scala-2.10/scoverage-report/index.html
../target/scala-2.10/coverage-report/cobertura.xml
./target/scala-2.10/scoverage-report/scoverage.xml
Integration with CI servers:
**/target/scala-2.10/coverage-report/cobertura.xml
.target/scala-2.10/scoverage-report/** => coberturareport/
.To create a normal ("slim") jar:
$ ./sbt clean package
>>> Generates `target/scala-2.10/kafka-storm-starter_2.10-0.2.0-SNAPSHOT.jar`
To create a fat jar, which includes any dependencies of kafka-storm-starter:
$ ./sbt assembly
>>> Generates `target/scala-2.10/kafka-storm-starter-assembly-0.2.0-SNAPSHOT.jar`
Note: By default, assembly
by itself will NOT run any tests. If you want to run tests before assembly, chain sbt
commands in sequence, e.g. ./sbt test assembly
. See assembly.sbt` for details why we do this.
To create a scaladoc/javadoc jar:
$ ./sbt packageDoc
>>> Generates `target/scala-2.10/kafka-storm-starter_2.10-0.2.0-SNAPSHOT-javadoc.jar`
To create a sources jar:
$ ./sbt packageSrc
>>> Generates `target/scala-2.10/kafka-storm-starter_2.10-0.2.0-SNAPSHOT-sources.jar`
To create API docs:
$ ./sbt doc
>>> Generates `target/scala-2.10/api/*` (HTML files)
kafka-storm-starter integrates the sbt-idea plugin. Use the following command to build IDEA project files:
$ ./sbt gen-idea
You can then open kafka-storm-starter as a project in IDEA via File > Open... and selecting the top-level directory of kafka-storm-starter.
Important note: There is a bug when using the sbt plugins for Avro and for IntelliJ IDEA in combination. The sbt
plugin for Avro reads the Avro *.avsc
schemas stored under src/main/avro
and generates the corresponding Java
classes, which it stores under target/scala-2.10/src_managed/main/compiled_avro
(in the case of kafka-storm-starter,
a Tweet.java
class will be generated from the Avro schema twitter.avsc). The latter
path must be added to IDEA's Source Folders setting, which will happen automatically for you. However the
aforementioned bug will add a second, incorrect path to Source Folders, too, which will cause IDEA to complain about
not being able to find the Avro-generated Java classes (here: the Tweet
class).
Until this bug is fixed upstream you can use the following workaround, which you must perform everytime you run
./sbt gen-idea
:
target/scala-2.10/src_managed/main/compiled_avro/com
entry from the Source Folders listing
(the source folders are colored in light-blue). Note the trailing .../com
, which comes from
com.miguno.avro.Tweet
in the twitter.avsc Avro schema.See also this screenshot (click to enlarge):
kafka-storm-starter integrates the sbt-eclipse plugin. Use the following command to build Eclipse project files:
$ ./sbt eclipse
Then use the Import Wizard in Eclipse to import Existing Projects into Workspace.
In short you can normally safely ignore those errors -- it's for a reason they are logged at INFO level and not at ERROR level.
As described in the mailing list thread Zookeeper exceptions:
"The reason you see those NoNode error code is the following. Every time we want to create a new [ZK] path, say
/brokers/ids/1
, we try to create it directly. If this fails because the parent path doesn't exist, we try to create
the parent path first. This will happen recursively. However, the NoNode
error should show up only once, not every
time a broker is started (assuming ZK data hasn't been cleaned up)."
A similar answer was given in the thread Clean up kafka environment:
"These info messages show up when Kafka tries to create new consumer groups. While trying to create the children of
/consumers/[group]
, if the parent path doesn't exist, the zookeeper server logs these messages. Kafka internally
handles these cases correctly by first creating the parent node."
LocalCluster
and ZooKeeperLocalCluster
starts an embedded ZooKeeper instance listening at localhost:2000/tcp
. If a different process is already bound to
2000/tcp
, then Storm will increment the embedded ZooKeeper's port until it finds a free port (2000
-> 2001
->
2002
, and so on). LocalCluster
then reads the Storm defaults and overrides some of Storm's configuration (see the
mk-local-storm-cluster
function in
testing.clj and
the mk-inprocess-zookeeper
function in
zookeeper.clj
for details):
STORM-CLUSTER-MODE "local"
STORM-ZOOKEEPER-PORT zk-port
STORM-ZOOKEEPER-SERVERS ["localhost"]}
where zk-port
is the final port chosen.
In Storm versions <= 0.9.2 it is not possible to launch a local Storm cluster via LocalCluster
without its own embedded
ZooKeeper. Likewise it is not possible to control on which port the embedded ZooKeeper process will listen -- it will
always follow the 2000/tcp
based algorithm above to set the port.
In Storm 0.9.3 and later you can configure LocalCluster
to use a custom ZooKeeper instance, thanks to
STORM-213.
This section lists known issues and limitations a) for the upstream projects such as Storm and Kafka, and b) for our own code.
Note: We squelch this message during test runs. See log4j.properties.
You may see the following exception when running the integration tests, which you can safely ignore:
[2014-03-07 11:56:59,250] WARN Failed to register with JMX (org.apache.zookeeper.server.ZooKeeperServer)
javax.management.InstanceAlreadyExistsException: org.apache.ZooKeeperService:name0=StandaloneServer_port-1
The root cause is that in-memory ZooKeeper instances have a hardcoded JMX setup. And because we cannot prevent Storm's
LocalCluster
to start its own ZooKeeper instance alongside "ours" (see FAQ section above), there will be two ZK
instances trying to use the same JMX setup. Since the JMX setup is not relevant for our testing the exception can be
safely ignored, albeit we'd prefer to come up with a proper fix, of course.
See also ZOOKEEPER-1350: Make JMX registration optional in LearnerZooKeeperServer,
which will make it possible to disable JMX registration when using Curator's TestServer
to run an in-memory ZooKeeper
instance (this patch will be included in ZooKeeper 3.5.0, see JIRA ticket above).
At the time of writing Kafka 0.8 is not officially compatible with ZooKeeper 3.4.x, which is the latest stable version of ZooKeeper. Instead the Kafka project recommends ZooKeeper 3.3.4.
So which version of ZooKeeper should you do pick, particularly if you are already running a ZooKeeper cluster for other parts of your infrastructure (such as an Hadoop cluster)?
The TL;DR version is: Try using ZooKeeper 3.4.5 for both Kafka and Storm, but see the caveats and workarounds below. In the worst case use separate ZooKeeper clusters/versions for Storm (3.4.5) and Kafka (3.3.4). Generally speaking though, the best 3.3.x version of ZooKeeper is 3.3.6, which is the latest stable 3.3.x version. This is because 3.3.6 fixed a number of serious bugs that could lead to data corruption.
Tip: You can verify the exact ZK version used in kafka-storm-starter by running ./sbt dependency-graph
.
Notes:
Thread.sleep()
in some tests instead of more intelligent approaches. To prevent transient failures we
may thus want to improve those tests. In Kafka's test suites, for instance, tests are using waitUntilTrue()
to
detect more reliably when to proceed (or fail/timeout) with the next step. See the related discussion in the
review request 19696 for KAFKA-1317.See CHANGELOG.
Code contributions, bug reports, feature requests etc. are all welcome.
If you are new to GitHub please read Contributing to a project for how to send patches and pull requests to kafka-storm-starter.
Copyright © 2014 Michael G. Noll
See LICENSE for licensing information.
Want to perform 1-click deployments of Kafka clusters and/or Storm clusters (with a Graphite instance, with Redis, with...)? Take a look at Wirbelsturm, with which you can deploy such environments locally and to Amazon AWS.
Kafka in general:
Unit testing:
Storm in general:
Unit testing:
backtype.storm.Testing
is apparently not well suited to test Trident topologies.
See Any Java example to write test cases for storm Transactional topology
(Mar 2013) for details.OutputCollector
for unit testing.LocalCluster
), and in-memory Kafka and Zookeeper
instances. For a number of reasons we opted not to use that code. We list it in this section in case someone else
may find it helpful.Kafka spout wurstmeister/storm-kafka-0.8-plus:
Kafka spout HolmesNL/kafka-spout, written by the Netherlands Forensics Institute:
Twitter Bijection:
Specific*
APIGeneric*
APIKafka: