storm与redis

Author: centmeng

src/com/msj/storm/redis/RedisOperations.java

@@ -0,0 +1,33 @@
+package com.msj.storm.redis;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import redis.clients.jedis.Jedis;
+
+import java.io.Serializable;
+import java.util.Map;
+
+
+/**
+ * @author Vincent.M [email protected]
+ * @date 2017/5/11 下午2:57
+ * @copyright ©2017 孟少杰 All Rights Reserved
+ * @desc Redis与storm-初始化Jedis
+ */
+public class RedisOperations implements Serializable {
+
+	private static final long serialVersionUID = 1L;
+	Jedis jedis = null;
+
+	public RedisOperations(String redisIP, int port) {
+		// Connecting to Redis on localhost，也可以使用集群
+		jedis = new Jedis(redisIP, port);
+	}
+
+	public void insert(Map<String, Object> record, String id) {
+		try {
+			jedis.set(id, new ObjectMapper().writeValueAsString(record));
+		} catch (Exception e) {
+			System.out.println("Record not persist into datastore : ");
+		}
+	}
+}

src/com/msj/storm/redis/SampleSpout.java

@@ -0,0 +1,64 @@
+package com.msj.storm.redis;
+
+import backtype.storm.spout.SpoutOutputCollector;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.topology.base.BaseRichSpout;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Values;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+
+public class SampleSpout extends BaseRichSpout {
+	private static final long serialVersionUID = 1L;
+	private SpoutOutputCollector spoutOutputCollector;
+	
+	private static final Map<Integer, String> FIRSTNAMEMAP = new HashMap<Integer, String>();
+	static {
+		FIRSTNAMEMAP.put(0, "john");
+		FIRSTNAMEMAP.put(1, "nick");
+		FIRSTNAMEMAP.put(2, "mick");
+		FIRSTNAMEMAP.put(3, "tom");
+		FIRSTNAMEMAP.put(4, "jerry");
+	}
+	
+	private static final Map<Integer, String> LASTNAME = new HashMap<Integer, String>();
+	static {
+		LASTNAME.put(0, "anderson");
+		LASTNAME.put(1, "watson");
+		LASTNAME.put(2, "ponting");
+		LASTNAME.put(3, "dravid");
+		LASTNAME.put(4, "lara");
+	}
+	
+	private static final Map<Integer, String> COMPANYNAME = new HashMap<Integer, String>();
+	static {
+		COMPANYNAME.put(0, "abc");
+		COMPANYNAME.put(1, "dfg");
+		COMPANYNAME.put(2, "pqr");
+		COMPANYNAME.put(3, "ecd");
+		COMPANYNAME.put(4, "awe");
+	}
+
+	public void open(Map conf, TopologyContext context,
+			SpoutOutputCollector spoutOutputCollector) {
+		// Open the spout
+		this.spoutOutputCollector = spoutOutputCollector;
+	}
+
+	public void nextTuple() {
+		// Storm cluster repeatedly call this method to emit the continuous //
+		// stream of tuples.
+		final Random rand = new Random();
+		// generate the random number from 0 to 4.
+		int randomNumber = rand.nextInt(5);
+		spoutOutputCollector.emit (new Values(FIRSTNAMEMAP.get(randomNumber),LASTNAME.get(randomNumber),COMPANYNAME.get(randomNumber)));
+	}
+
+	public void declareOutputFields(OutputFieldsDeclarer declarer) {
+		// emit the field site.
+		declarer.declare(new Fields("firstName","lastName","companyName"));
+	}
+}

src/com/msj/storm/redis/StormRedisBolt.java

@@ -0,0 +1,50 @@
+package com.msj.storm.redis;
+
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.BasicOutputCollector;
+import backtype.storm.topology.IBasicBolt;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.tuple.Tuple;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+public class StormRedisBolt implements IBasicBolt{
+	
+	private static final long serialVersionUID = 2L;
+	private RedisOperations redisOperations = null;
+	private String redisIP = null;
+	private int port;
+	public StormRedisBolt(String redisIP, int port) {
+		this.redisIP = redisIP;
+		this.port = port;
+	}
+	
+	public void execute(Tuple input, BasicOutputCollector collector) {
+		Map<String, Object> record = new HashMap<String, Object>();
+		//"firstName","lastName","companyName")
+		record.put("firstName", input.getValueByField("firstName"));
+		record.put("lastName", input.getValueByField("lastName"));
+		record.put("companyName", input.getValueByField("companyName"));
+		redisOperations.insert(record, UUID.randomUUID().toString());
+	}
+
+	public void declareOutputFields(OutputFieldsDeclarer declarer) {
+		
+	}
+
+	public Map<String, Object> getComponentConfiguration() {
+		return null;
+	}
+
+	public void prepare(Map stormConf, TopologyContext context) {
+		redisOperations = new RedisOperations(this.redisIP, this.port);
+	}
+
+	public void cleanup() {
+		
+	}
+
+}
+

src/com/msj/storm/redis/Topology.java

@@ -0,0 +1,53 @@
+package com.msj.storm.redis;
+
+import backtype.storm.Config;
+import backtype.storm.LocalCluster;
+import backtype.storm.generated.AlreadyAliveException;
+import backtype.storm.generated.InvalidTopologyException;
+import backtype.storm.topology.TopologyBuilder;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author Vincent.M [email protected]
+ * @date 2017/5/11 下午2:57
+ * @copyright ©2017 孟少杰 All Rights Reserved
+ * @desc Redis与storm
+ */
+public class Topology {
+	public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
+		
+		TopologyBuilder builder = new TopologyBuilder();
+		
+		List<String> zks = new ArrayList<String>();
+		zks.add("192.168.1.114");
+		
+		List<String> cFs = new ArrayList<String>();
+		cFs.add("personal");
+		cFs.add("company");
+		
+		// set the spout class
+		builder.setSpout("spout", new SampleSpout(), 2);
+		// set the bolt class
+		builder.setBolt("bolt", new StormRedisBolt("192.168.1.114",6379), 2).shuffleGrouping("spout");
+
+		Config conf = new Config();
+		conf.setDebug(true);
+		// create an instance of LocalCluster class for
+		// executing topology in local mode.
+		LocalCluster cluster = new LocalCluster();
+
+		// StormRedisTopology is the name of submitted topology.
+		cluster.submitTopology("StormRedisTopology", conf, builder.createTopology());
+		try {
+			Thread.sleep(10000);
+		} catch (Exception exception) {
+			System.out.println("Thread interrupted exception : " + exception);
+		}
+		// kill the StormRedisTopology
+		cluster.killTopology("StormRedisTopology");
+		// shutdown the storm test cluster
+		cluster.shutdown();
+}
+}