不透明分区事务IOpaquePartitionedTransactional实例

1、spout

public class MyOpaquePtTxSpout implements IOpaquePartitionedTransactionalSpout<MyMata> {

	/**
	 * 将Coordinator发射数据动作下放到emitPartitionBatch类执行。
	 */
	private static final long serialVersionUID = 1L;
	public static int BATCH_NUM = 10;
	public Map<Integer, Map<Long, String>> PT_DATA_MP = new HashMap<Integer, Map<Long, String>>();

	public MyOpaquePtTxSpout() {
		Random random = new Random();
		String[] hosts = { "www.taobao.com" };
		String[] session_id = { "ABYH6Y4V4SCVXTG6DPB4VH9U123", "XXYH6YCGFJYERTT834R52FDXV9U34",
				"BBYH61456FGHHJ7JL89RG5VV9UYU7", "CYYH6Y2345GHI899OFG4V9U567", "VVVYH6Y4V4SFXZ56JIPDPB4V678" };
		String[] time = { "2014-01-07 08:40:50", "2014-01-07 08:40:51", "2014-01-07 08:40:52", "2014-01-07 08:40:53",
				"2014-01-07 09:40:49", "2014-01-07 10:40:49", "2014-01-07 11:40:49", "2014-01-07 12:40:49" };

		for (int j = 0; j < 5; j++) {
			HashMap<Long, String> dbMap = new HashMap<Long, String>();
			for (long i = 0; i < 100; i++) {
				dbMap.put(i, hosts[0] + "\t" + session_id[random.nextInt(5)] + "\t" + time[random.nextInt(8)]);
			}
			PT_DATA_MP.put(j, dbMap);
		}
	}

	public org.apache.storm.transactional.partitioned.IOpaquePartitionedTransactionalSpout.Coordinator getCoordinator(
			Map conf, TopologyContext context) {
		return new MyCoordinator();
	}

	public org.apache.storm.transactional.partitioned.IOpaquePartitionedTransactionalSpout.Emitter<MyMata> getEmitter(
			Map conf, TopologyContext context) {
		return new MyEmitter();
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("tx", "log"));
	}

	public Map<String, Object> getComponentConfiguration() {
		return null;
	}

	public class MyCoordinator implements IOpaquePartitionedTransactionalSpout.Coordinator {

		public void close() {

		}

		public boolean isReady() {
			Utils.sleep(1000);
			return true;
		}

	}

	public class MyEmitter implements IOpaquePartitionedTransactionalSpout.Emitter<MyMata> {

		public void close() {

		}

		public MyMata emitPartitionBatch(TransactionAttempt tx, BatchOutputCollector collector, int partition,
				MyMata lastPartitionMeta) {
			System.err.println("emitPartitionBatch partition:" + partition);
			long beginPoint = 0;
			if (lastPartitionMeta == null) {
				beginPoint = 0;
			} else {
				beginPoint = lastPartitionMeta.getBeginPoint() + lastPartitionMeta.getNum();
			}

			MyMata mata = new MyMata();
			mata.setBeginPoint(beginPoint);
			mata.setNum(BATCH_NUM);
			System.err.println("启动一个事务：" + mata.toString());
			//获取某一个分区数据
			Map<Long, String> batchMap = PT_DATA_MP.get(partition);
			for (Long i = mata.getBeginPoint(); i < mata.getBeginPoint() + mata.getNum(); i++) {
				if (batchMap.size() <= i) {
					break;
				}
				collector.emit(new Values(tx, batchMap.get(i)));
			}
			//返回一个mata给当前方法，当作下次循环的参数
			return mata;
		}

		public int numPartitions() {
			return 5;
		}

	}
}

2、处理类不作改变

public class MyDailyBatchBolt implements IBatchBolt<TransactionAttempt> {

	/**
	 * 不作改变
	 */
	private static final long serialVersionUID = 1L;
	Map<String, Integer> countMap = new HashMap<String, Integer>();
	BatchOutputCollector collector ;
	Integer count = null;
	String today = null;
	TransactionAttempt tx = null;
	
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		String log = tuple.getString(1);
		tx = (TransactionAttempt)tuple.getValue(0);
		if (log != null && log.split("\\t").length >=3 ) {
			today = DateFmt.getCountDate(log.split("\\t")[2], DateFmt.date_short) ;
			count = countMap.get(today);
			if(count == null)
			{
				count = 0;
			}
			count ++ ;
			
			countMap.put(today, count);
		}
	}

	
	public void finishBatch() {
		System.err.println(tx+"--"+today+"--"+count);
		collector.emit(new Values(tx,today,count));
	}

	
	public void prepare(Map conf, TopologyContext context,
			BatchOutputCollector collector, TransactionAttempt id) {
		this.collector = collector;
		
	}

	
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("tx","date","count"));
	}

	
	public Map<String, Object> getComponentConfiguration() {
		return null;
	}

}

3、汇总及批量提交

public class MyDailyCommitterBolt extends BaseTransactionalBolt implements ICommitter {

	/**
	 * finishBatch增加上个事务批次处理后的结果
	 */
	private static final long serialVersionUID = 1L;
	public static Map<String, DbValue> dbMap = new HashMap<String, DbValue>();

	Map<String, Integer> countMap = new HashMap<String, Integer>();
	TransactionAttempt id;
	BatchOutputCollector collector;
	String today = null;

	public void execute(Tuple tuple) {
		today = tuple.getString(1);
		Integer count = tuple.getInteger(2);
		id = (TransactionAttempt) tuple.getValue(0);

		if (today != null && count != null) {
			Integer batchCount = countMap.get(today);
			if (batchCount == null) {
				batchCount = 0;
			}
			batchCount += count;
			countMap.put(today, batchCount);
		}
	}

	public void finishBatch() {
		// TODO Auto-generated method stub
		if (countMap.size() > 0) {
			DbValue value = dbMap.get(today);
			DbValue newValue;
			if (value == null || !value.txid.equals(id.getTransactionId())) {
				// 更新数据库
				newValue = new DbValue();
				newValue.txid = id.getTransactionId();
				newValue.dateStr = today;
				if (value == null) {
					newValue.count = countMap.get(today);
					newValue.pre_count = 0;
				} else {
					newValue.pre_count = value.count;// 上个事务批次处理后的结果
					newValue.count = value.count + countMap.get("2014-01-07");
				}
				dbMap.put(today, newValue);
			} else {
				newValue = value;
			}
			System.out.println("total==========================:" + dbMap.get(today).count);
		}
	}

	public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector, TransactionAttempt id) {
		this.id = id;
		this.collector = collector;
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {

	}

	// 模拟一个数据表
	public static class  DbValue {
		String dateStr; // 按日期汇总
		int count = 0; // 汇总数
		BigInteger txid; // 事务ID
		int pre_count; // 上个事务批次处理后的结果
	}

}

4、topo类

public class MyDailyTopo {

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub

		TransactionalTopologyBuilder builder = new TransactionalTopologyBuilder("ttbId","spoutid",new MyOpaquePtTxSpout(),1);
		builder.setBolt("bolt1", new MyDailyBatchBolt(),3).shuffleGrouping("spoutid");
		builder.setBolt("committer", new MyDailyCommitterBolt(),1).shuffleGrouping("bolt1") ;
		
		Config conf = new Config() ;
		conf.setDebug(false);

		if (args.length > 0) {
			try {
				StormSubmitter.submitTopology(args[0], conf, builder.buildTopology());
			} catch (AlreadyAliveException e) {
				e.printStackTrace();
			} catch (InvalidTopologyException e) {
				e.printStackTrace();
			} catch (AuthorizationException e) {
				e.printStackTrace();
			}
		}else {
			LocalCluster localCluster = new LocalCluster();
			localCluster.submitTopology("mytopology", conf, builder.buildTopology());
		}
		
		
		
		
	}

}

6、测试结果

引用

emitPartitionBatch partition:0
启动一个事务：0----10
emitPartitionBatch partition:1
启动一个事务：0----10
emitPartitionBatch partition:2
启动一个事务：0----10
emitPartitionBatch partition:3
启动一个事务：0----10
emitPartitionBatch partition:4
启动一个事务：0----10
1:-1542691397832167833--2014-01-07--16
1:-1542691397832167833--2014-01-07--22
1:-1542691397832167833--2014-01-07--12
total==========================:50
null--null--null
null--null--null
null--null--null
emitPartitionBatch partition:0
启动一个事务：10----10
emitPartitionBatch partition:1
启动一个事务：10----10
emitPartitionBatch partition:2
启动一个事务：10----10
emitPartitionBatch partition:3
启动一个事务：10----10
emitPartitionBatch partition:4
启动一个事务：10----10
2:9191261929288144194--2014-01-07--22
2:9191261929288144194--2014-01-07--16
2:9191261929288144194--2014-01-07--12
total==========================:100