adding simple bolt and spout and woring examples
This commit is contained in:
parent
473fbc8817
commit
223c9b07e2
|
@ -0,0 +1,19 @@
|
||||||
|
require 'red_storm'
|
||||||
|
require 'examples/simple/random_sentence_spout'
|
||||||
|
require 'examples/simple/split_sentence_bolt'
|
||||||
|
require 'examples/simple/word_count_bolt'
|
||||||
|
|
||||||
|
class ClusterWordCountTopology
|
||||||
|
def start(base_class_path)
|
||||||
|
builder = TopologyBuilder.new
|
||||||
|
builder.setSpout(1, JRubySpout.new(base_class_path, "RandomSentenceSpout"), 5)
|
||||||
|
builder.setBolt(2, JRubyBolt.new(base_class_path, "SplitSentenceBolt"), 4).shuffleGrouping(1)
|
||||||
|
builder.setBolt(3, JRubyBolt.new(base_class_path, "WordCountBolt"), 4).fieldsGrouping(2, Fields.new("word"))
|
||||||
|
|
||||||
|
conf = Config.new
|
||||||
|
conf.setDebug(true)
|
||||||
|
conf.setNumWorkers(20);
|
||||||
|
conf.setMaxSpoutPending(1000);
|
||||||
|
StormSubmitter.submitTopology("word-count", conf, builder.createTopology);
|
||||||
|
end
|
||||||
|
end
|
|
@ -0,0 +1,4 @@
|
||||||
|
class ExclamationBolt < RedStorm::SimpleBolt
|
||||||
|
output_fields :word
|
||||||
|
on_tuple(:ack => true, :anchor => true) {|tuple| tuple.getString(0) + "!!!"}
|
||||||
|
end
|
|
@ -0,0 +1,25 @@
|
||||||
|
java_import 'backtype.storm.testing.TestWordSpout'
|
||||||
|
|
||||||
|
require 'red_storm'
|
||||||
|
require 'examples/simple/exclamation_bolt'
|
||||||
|
|
||||||
|
# this example topology uses the Storm TestWordSpout and our own JRuby ExclamationBolt
|
||||||
|
|
||||||
|
class LocalExclamationTopology
|
||||||
|
def start(base_class_path)
|
||||||
|
builder = TopologyBuilder.new
|
||||||
|
|
||||||
|
builder.setSpout(1, TestWordSpout.new, 10)
|
||||||
|
builder.setBolt(2, JRubyBolt.new(base_class_path, "ExclamationBolt"), 3).shuffleGrouping(1)
|
||||||
|
builder.setBolt(3, JRubyBolt.new(base_class_path, "ExclamationBolt"), 2).shuffleGrouping(2)
|
||||||
|
|
||||||
|
conf = Config.new
|
||||||
|
conf.setDebug(true)
|
||||||
|
|
||||||
|
cluster = LocalCluster.new
|
||||||
|
cluster.submitTopology("test", conf, builder.createTopology)
|
||||||
|
sleep(5)
|
||||||
|
cluster.killTopology("test")
|
||||||
|
cluster.shutdown
|
||||||
|
end
|
||||||
|
end
|
|
@ -0,0 +1,29 @@
|
||||||
|
java_import 'backtype.storm.testing.TestWordSpout'
|
||||||
|
|
||||||
|
require 'red_storm'
|
||||||
|
|
||||||
|
class ExclamationBolt2 < RedStorm::SimpleBolt
|
||||||
|
output_fields :word
|
||||||
|
on_tuple(:ack => true, :anchor => true) {|tuple| tuple.getString(0) + "!!!"}
|
||||||
|
end
|
||||||
|
|
||||||
|
# this example topology uses the Storm TestWordSpout and our own JRuby ExclamationBolt
|
||||||
|
|
||||||
|
class LocalExclamationTopology2
|
||||||
|
def start(base_class_path)
|
||||||
|
builder = TopologyBuilder.new
|
||||||
|
|
||||||
|
builder.setSpout(1, TestWordSpout.new, 10)
|
||||||
|
builder.setBolt(2, JRubyBolt.new(base_class_path, "ExclamationBolt2"), 3).shuffleGrouping(1)
|
||||||
|
builder.setBolt(3, JRubyBolt.new(base_class_path, "ExclamationBolt2"), 2).shuffleGrouping(2)
|
||||||
|
|
||||||
|
conf = Config.new
|
||||||
|
conf.setDebug(true)
|
||||||
|
|
||||||
|
cluster = LocalCluster.new
|
||||||
|
cluster.submitTopology("test", conf, builder.createTopology)
|
||||||
|
sleep(5)
|
||||||
|
cluster.killTopology("test")
|
||||||
|
cluster.shutdown
|
||||||
|
end
|
||||||
|
end
|
|
@ -0,0 +1,52 @@
|
||||||
|
require 'redis'
|
||||||
|
require 'thread'
|
||||||
|
|
||||||
|
require 'red_storm'
|
||||||
|
require 'examples/simple/word_count_bolt'
|
||||||
|
|
||||||
|
# RedisWordSpout reads the Redis queue "test" on localhost:6379
|
||||||
|
# and emits each word items pop'ed from the queue.
|
||||||
|
|
||||||
|
class RedisWordSpout
|
||||||
|
output_fields :word
|
||||||
|
|
||||||
|
on_next_tuple {@q.pop if @q.size > 0}
|
||||||
|
|
||||||
|
on_init do
|
||||||
|
@q = Queue.new
|
||||||
|
@redis_reader = detach_redis_reader
|
||||||
|
end
|
||||||
|
|
||||||
|
private
|
||||||
|
|
||||||
|
def detach_redis_reader
|
||||||
|
Thread.new do
|
||||||
|
Thread.current.abort_on_exception = true
|
||||||
|
|
||||||
|
redis = Redis.new(:host => "localhost", :port => 6379)
|
||||||
|
loop do
|
||||||
|
if data = redis.blpop("test", 0)
|
||||||
|
@q << data[1]
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
|
||||||
|
class LocalRedisWordCountTopology
|
||||||
|
def start(base_class_path)
|
||||||
|
builder = TopologyBuilder.new
|
||||||
|
builder.setSpout(1, JRubySpout.new(base_class_path, "RedisWordSpout"), 1)
|
||||||
|
builder.setBolt(2, JRubyBolt.new(base_class_path, "WordCountBolt"), 3).fieldsGrouping(1, Fields.new("word"))
|
||||||
|
|
||||||
|
conf = Config.new
|
||||||
|
conf.setDebug(true)
|
||||||
|
conf.setMaxTaskParallelism(3)
|
||||||
|
|
||||||
|
cluster = LocalCluster.new
|
||||||
|
cluster.submitTopology("redis-word-count", conf, builder.createTopology)
|
||||||
|
sleep(600)
|
||||||
|
cluster.shutdown
|
||||||
|
end
|
||||||
|
end
|
|
@ -0,0 +1,22 @@
|
||||||
|
require 'red_storm'
|
||||||
|
require 'examples/simple/random_sentence_spout'
|
||||||
|
require 'examples/simple/split_sentence_bolt'
|
||||||
|
require 'examples/simple/word_count_bolt'
|
||||||
|
|
||||||
|
class LocalWordCountTopology
|
||||||
|
def start(base_class_path)
|
||||||
|
builder = TopologyBuilder.new
|
||||||
|
builder.setSpout(1, JRubySpout.new(base_class_path, "RandomSentenceSpout"), 5)
|
||||||
|
builder.setBolt(2, JRubyBolt.new(base_class_path, "SplitSentenceBolt"), 8).shuffleGrouping(1)
|
||||||
|
builder.setBolt(3, JRubyBolt.new(base_class_path, "WordCountBolt"), 12).fieldsGrouping(2, Fields.new("word"))
|
||||||
|
|
||||||
|
conf = Config.new
|
||||||
|
conf.setDebug(true)
|
||||||
|
conf.setMaxTaskParallelism(3)
|
||||||
|
|
||||||
|
cluster = LocalCluster.new
|
||||||
|
cluster.submitTopology("word-count", conf, builder.createTopology)
|
||||||
|
sleep(5)
|
||||||
|
cluster.shutdown
|
||||||
|
end
|
||||||
|
end
|
|
@ -0,0 +1,16 @@
|
||||||
|
class RandomSentenceSpout < RedStorm::SimpleSpout
|
||||||
|
set :is_distributed => true
|
||||||
|
output_fields :word
|
||||||
|
|
||||||
|
on_next_tuple {@sentences[rand(@sentences.length)]}
|
||||||
|
|
||||||
|
on_init do
|
||||||
|
@sentences = [
|
||||||
|
"the cow jumped over the moon",
|
||||||
|
"an apple a day keeps the doctor away",
|
||||||
|
"four score and seven years ago",
|
||||||
|
"snow white and the seven dwarfs",
|
||||||
|
"i am at two with nature"
|
||||||
|
]
|
||||||
|
end
|
||||||
|
end
|
|
@ -0,0 +1,4 @@
|
||||||
|
class SplitSentenceBolt < RedStorm::SimpleBolt
|
||||||
|
output_fields :word
|
||||||
|
on_tuple(:emit => false) {|tuple| tuple.getString(0).split(' ').each {|w| emit(w)}}
|
||||||
|
end
|
|
@ -0,0 +1,32 @@
|
||||||
|
class WordCountBolt < RedStorm::SimpleBolt
|
||||||
|
output_fields :word, :count
|
||||||
|
|
||||||
|
on_tuple do |tuple|
|
||||||
|
word = tuple.getString(0)
|
||||||
|
@counts[word] += 1
|
||||||
|
|
||||||
|
[word, @counts[word]]
|
||||||
|
end
|
||||||
|
|
||||||
|
def initialize
|
||||||
|
@counts = Hash.new{|h, k| h[k] = 0}
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
|
||||||
|
# class WordCountBolt < RedStorm::SimpleBolt
|
||||||
|
|
||||||
|
# output_fields :word, :count
|
||||||
|
# on_tuple :count_word, :ack => true, :anchor => true
|
||||||
|
|
||||||
|
# def count_word(tuple)
|
||||||
|
# word = tuple.getString(0)
|
||||||
|
# @counts[word] += 1
|
||||||
|
|
||||||
|
# [word, @counts[word]]
|
||||||
|
# end
|
||||||
|
|
||||||
|
# def initialize
|
||||||
|
# @counts = Hash.new{|h, k| h[k] = 0}
|
||||||
|
# end
|
||||||
|
# end
|
|
@ -4,3 +4,5 @@ end
|
||||||
|
|
||||||
require 'red_storm/version'
|
require 'red_storm/version'
|
||||||
require 'red_storm/application'
|
require 'red_storm/application'
|
||||||
|
require 'red_storm/simple_bolt'
|
||||||
|
require 'red_storm/simple_spout'
|
||||||
|
|
|
@ -1,19 +1,22 @@
|
||||||
require 'rake'
|
require 'rake'
|
||||||
|
|
||||||
class RedStorm::Application
|
module RedStorm
|
||||||
|
|
||||||
|
class Application
|
||||||
|
|
||||||
def run(args)
|
def run(args)
|
||||||
if args.size == 1 && File.exist?(args.first)
|
if args.size == 1 && File.exist?(args.first)
|
||||||
load("#{RedStorm::REDSTORM_HOME}/Rakefile")
|
|
||||||
Rake::Task['launch'].invoke(args)
|
|
||||||
else
|
|
||||||
task = args.shift
|
|
||||||
if ["install", "examples", "jar"].include?(task)
|
|
||||||
load("#{RedStorm::REDSTORM_HOME}/Rakefile")
|
load("#{RedStorm::REDSTORM_HOME}/Rakefile")
|
||||||
Rake::Task[task].invoke(args)
|
Rake::Task['launch'].invoke(args)
|
||||||
else
|
else
|
||||||
puts("\nUsage: redstorm install|examples|jar|topology_class_file_name")
|
task = args.shift
|
||||||
exit(1)
|
if ["install", "examples", "jar"].include?(task)
|
||||||
|
load("#{RedStorm::REDSTORM_HOME}/Rakefile")
|
||||||
|
Rake::Task[task].invoke(args)
|
||||||
|
else
|
||||||
|
puts("\nUsage: redstorm install|examples|jar|topology_class_file_name")
|
||||||
|
exit(1)
|
||||||
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
|
@ -0,0 +1,81 @@
|
||||||
|
module RedStorm
|
||||||
|
|
||||||
|
class SimpleBolt
|
||||||
|
|
||||||
|
# DSL class mthods
|
||||||
|
|
||||||
|
def self.output_fields(*fields)
|
||||||
|
@fields = fields.map(&:to_s)
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.on_tuple(method_name = nil, options = {}, &execute_block)
|
||||||
|
self.execute_options.merge!(options)
|
||||||
|
@execute_block = block_given? ? execute_block : lambda {|tuple| self.send(method_name, tuple)}
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.on_init(method_name = nil, &init_block)
|
||||||
|
@init_block = block_given? ? init_block : lambda {self.send(method_name)}
|
||||||
|
end
|
||||||
|
|
||||||
|
def emit(*values)
|
||||||
|
@collector.emit(Values.new(*values))
|
||||||
|
end
|
||||||
|
|
||||||
|
def ack(tuple)
|
||||||
|
@collector.ack(tuple)
|
||||||
|
end
|
||||||
|
|
||||||
|
# Bolt interface
|
||||||
|
|
||||||
|
def execute(tuple)
|
||||||
|
if (output = instance_exec(tuple, &self.class.execute_block)) && self.class.emit?
|
||||||
|
values = [output].flatten
|
||||||
|
self.class.anchor? ? @collector.emit(tuple, Values.new(*values)) : @collector.emit(Values.new(*values))
|
||||||
|
@collector.ack(tuple) if self.class.ack?
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
def prepare(config, context, collector)
|
||||||
|
@collector = collector
|
||||||
|
@context = context
|
||||||
|
@config = config
|
||||||
|
instance_exec(&self.class.init_block)
|
||||||
|
end
|
||||||
|
|
||||||
|
def declare_output_fields(declarer)
|
||||||
|
declarer.declare(Fields.new(self.class.fields))
|
||||||
|
end
|
||||||
|
|
||||||
|
private
|
||||||
|
|
||||||
|
def self.fields
|
||||||
|
@fields
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.execute_block
|
||||||
|
@execute_block ||= lambda {}
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.init_block
|
||||||
|
@init_block ||= lambda {}
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.emit?
|
||||||
|
!!self.execute_options[:emit]
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.ack?
|
||||||
|
!!self.execute_options[:ack]
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.anchor?
|
||||||
|
!!self.execute_options[:anchor]
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.execute_options
|
||||||
|
@execute_options ||= {:emit => true, :ack => false, :anchor => false}
|
||||||
|
end
|
||||||
|
|
||||||
|
end
|
||||||
|
|
||||||
|
end
|
|
@ -0,0 +1,113 @@
|
||||||
|
module RedStorm
|
||||||
|
|
||||||
|
class SimpleSpout
|
||||||
|
|
||||||
|
|
||||||
|
# DSL class methods
|
||||||
|
|
||||||
|
def self.output_fields(*fields)
|
||||||
|
@fields = fields.map(&:to_s)
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.on_next_tuple(method_name = nil, options = {}, &execute_block)
|
||||||
|
self.execute_options.merge!(options)
|
||||||
|
@execute_block = block_given? ? execute_block : lambda {self.send(method_name)}
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.on_init(method_name = nil, &init_block)
|
||||||
|
@init_block = block_given? ? init_block : lambda {self.send(method_name)}
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.on_ack(method_name = nil, &ack_block)
|
||||||
|
@ack_block = block_given? ? ack_block : lambda {|msg_id| self.send(method_name, msg_id)}
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.on_fail(method_name = nil, &fail_block)
|
||||||
|
@fail_block = block_given? ? fail_block : lambda {|msg_id| self.send(method_name, msg_id)}
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.set(options = {})
|
||||||
|
self.global_options.merge!(options)
|
||||||
|
end
|
||||||
|
|
||||||
|
def emit(*values)
|
||||||
|
@collector.emit(Values.new(*values))
|
||||||
|
end
|
||||||
|
|
||||||
|
# Spout interface
|
||||||
|
|
||||||
|
def next_tuple
|
||||||
|
output = instance_exec(&self.class.execute_block)
|
||||||
|
if self.class.emit?
|
||||||
|
if output
|
||||||
|
values = [output].flatten
|
||||||
|
@collector.emit(Values.new(*values))
|
||||||
|
else
|
||||||
|
sleep(0.1)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
def open(config, context, collector)
|
||||||
|
@collector = collector
|
||||||
|
@context = context
|
||||||
|
@config = config
|
||||||
|
instance_exec(&self.class.init_block)
|
||||||
|
end
|
||||||
|
|
||||||
|
def declare_output_fields(declarer)
|
||||||
|
declarer.declare(Fields.new(self.class.fields))
|
||||||
|
end
|
||||||
|
|
||||||
|
def is_distributed
|
||||||
|
self.class.is_distributed?
|
||||||
|
end
|
||||||
|
|
||||||
|
def ack(msg_id)
|
||||||
|
instance_exec(msg_id, &self.class.ack_block)
|
||||||
|
end
|
||||||
|
|
||||||
|
def fail(msg_id)
|
||||||
|
instance_exec(msg_id, &self.class.fail_block)
|
||||||
|
end
|
||||||
|
|
||||||
|
private
|
||||||
|
|
||||||
|
def self.fields
|
||||||
|
@fields
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.execute_block
|
||||||
|
@execute_block ||= lambda {}
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.init_block
|
||||||
|
@init_block ||= lambda {}
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.ack_block
|
||||||
|
@ack_block ||= lambda {}
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.fail_block
|
||||||
|
@fail_block ||= lambda {}
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.is_distributed?
|
||||||
|
!!@global_options[:is_distributed]
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.execute_options
|
||||||
|
@execute_options ||= {:emit => true}
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.global_options
|
||||||
|
@global_options ||= {:is_distributed => false}
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.emit?
|
||||||
|
!!self.execute_options[:emit]
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
end
|
Loading…
Reference in New Issue