The realization process of Spout
This article mainly explains "the implementation process of Spout". Interested friends may wish to have a look. The method introduced in this paper is simple, fast and practical. Next, let the editor take you to learn "the implementation process of Spout"!
The implementation process of Spout:
Monitor file changes separately and monitor whether new log files are added in the directory.
After the data is described in the field, it is converted to tuple.
Declare the packet between Spout and Bolt and determine the path that tuple sends to Bolt.
The specific code of Spout is shown in Listing Three.
The logic of the open, nextTuple, and delcareOutputFields methods in Listing Three:Spout.
1. Public void open (Map conf, TopologyContext context,SpoutOutputCollector collector)
2. {
3. _ collector = collector
4. Try
5. {
6. FileReader = new BufferedReader (new FileReader (new File (file))
7.}
8. Catch (FileNotFoundException e)
9. {
10. System.exit (1)
11.}
12.}
13.
14. Public void nextTuple ()
15. {
16. Protected void ListenFile (File file)
17. {
18. Utils.sleep (2000)
19. RandomAccessFile access = null
20. String line = null
21. Try
twenty-two。 {
23. While (line = access.readLine ())! = null)
24. {
25. If (line! = null)
twenty-six。 {
twenty-seven。 String [] fields=null
twenty-eight。 If (tupleInfo.getDelimiter () .equals ("|") fields = line.split ("\" + tupleInfo.getDelimiter ())
twenty-nine。 Else
thirty。 Fields = line.split (tupleInfo.getDelimiter ())
thirty-one。 If (tupleInfo.getFieldList (). Size () = = fields.length) _ collector.emit (new Values (fields))
thirty-two。 }
thirty-three。 }
thirty-four。 }
thirty-five。 Catch (IOException ex) {}
thirty-six。 }
thirty-seven。 }
thirty-eight。
thirty-nine。 Public void declareOutputFields (OutputFieldsDeclarer declarer)
forty。 {
forty-one。 String [] fieldsArr = new String [tupleInfo.getFieldList () .size ()]
forty-two。 For (int item0; I
