How to use IBatchSpout API
This article mainly introduces "how to use IBatchSpout API". In daily operation, I believe many people have doubts about how to use IBatchSpout API. The editor consulted all kinds of materials and sorted out simple and easy-to-use operation methods. I hope it will be helpful to answer the doubts about "how to use IBatchSpout API". Next, please follow the editor to study!
IBatchSpout is a kind of Spout that can be launched in batch by storm trident. Non-transactional, basic spout
1:Map getComponentConfiguration (); defines the configuration, which can be done with backtype.storm.Config.
2:void open (Map conf, TopologyContext context); initialization method of Spout. Parameter conf is the configuration defined by getComponentConfiguration.
3:Fields getOutputFields (); declare the output fields
4:void emitBatch (long batchId, TridentCollector collector); batch launch tuple. The batch number of this time is batchId.
5:void ack (long batchId); the data with batch number batchId was processed successfully
6: void close ()
An example
Package storm.projectA;import java.io.BufferedReader;import java.io.FileNotFoundException;import java.io.FileReader;import java.io.IOException;import java.util.ArrayList;import java.util.HashMap;import java.util.List;import java.util.Map;import storm.trident.operation.TridentCollector;import storm.trident.spout.IBatchSpout;import backtype.storm.Config;import backtype.storm.task.TopologyContext;import backtype.storm.tuple.Fields;import backtype.storm.tuple.Values Public class MySpout implements IBatchSpout {/ * / private static final long serialVersionUID = 1L; maximum number of private long maxBatchSize;// per batch private BufferedReader br;// source file stream HashMap batches = new HashMap (); / / Save all data sent so that the / * @ param conf configuration * @ param context * / @ Override public void open (Map conf, TopologyContext context) {String filePath = (String) conf.get ("filePath") MaxBatchSize = (Long) conf.get ("maxBatchSize"); try {br = new BufferedReader (new FileReader (filePath));} catch (FileNotFoundException e) {e.printStackTrace ();}} / * spout delivery method * @ param batchId batch id * @ param collector batch launcher * / @ Override public void emitBatch (long batchId, TridentCollector collector) {List batch = batches.get (batchId); if (batch = null) {batch = new ArrayList () For (int I = 0; I < maxBatchSize; iTunes +) {try {String line = br.readLine (); if (line = = null) {break;} batch.add (new Values (line));} catch (IOException e) {e.printStackTrace ();}} for (List list: batch) {collector.emit (list) } @ Override public void ack (long batchId) {batches.remove (batchId);} / * close method * / @ Override public void close () {if (brushed null) {try {br.close ();} catch (IOException e) {e.printStackTrace ();} @ Override public Map getComponentConfiguration () {Config conf = new Config () / / maximum parallelism local mode is set to 1 conf.setMaxTaskParallelism (1); conf.put ("filePath", "D:\\ aaa.txt"); conf.put ("maxBatchSize", 2); fileds * / @ Override public Fields getOutputFields () {return new Fields ("sentence") of return conf;} / * * output }} at this point, the study on "how to use IBatchSpout API" is over. I hope to be able to solve your doubts. The collocation of theory and practice can better help you learn, go and try it! If you want to continue to learn more related knowledge, please continue to follow the website, the editor will continue to work hard to bring you more practical articles!
