flume 自定义source，sink，channel，拦截器

￡神魔★判官ぃ 2022-06-04 00:08 291阅读 0赞

按照以往的惯例，还是需求驱动学习，有位网友在我的flume学习五中留言提了一个问题如下：

我想实现一个功能，就在读一个文件的时候，将文件的名字和文件生成的日期作为event的header传到hdfs上时，不同的event存到不同的目录下，如一个文件是a.log.2014-07-25在hdfs上是存到/a/2014-07-25目录下，a.log.2014-07-26存到/a/2014-07-26目录下，就是每个文件对应自己的目录，这个要怎么实现。

带着这个问题，我又重新翻看了官方的文档，发现一个spooling directory source跟这个需求稍微有点吻合：它监视指定的文件夹下面有没有写入新的文件，有的话，就会把该文件内容传递给sink,然后将该文件后缀标示为.complete，表示已处理。提供了参数可以将文件名和文件全路径名添加到event的header中去。

现有的功能不能满足我们的需求，但是至少提供了一个方向：它能将文件名放入header！

当时就在祈祷源代码不要太复杂，这样我们在这个地方稍微修改修改，把文件名拆分一下，然后再放入header，这样就完成了我们想要的功能了。

于是就打开了源代码，果然不复杂，代码结构非常清晰，按照我的思路，稍微改了一下，就实现了这个功能，主要修改了与spooling directory source代码相关的三个类，分别是：ReliableSpoolingFileEventExtReader，SpoolDirectorySourceConfigurationExtConstants，SpoolDirectoryExtSource（在原类名的基础上增加了：Ext）代码如下：

首先，要根据flume ng提供的接口来实现自定义source，需要我们依赖flume ng的配置，我们引入两个配置flume-ng-core和flume-ng-configuration，具体的maven配置如下：

<dependency>
 <groupId>org.apache.flume</groupId>
 <artifactId>flume-ng-core</artifactId>
 <version>1.6.0</version>
 </dependency>
 <dependency>
 <groupId>org.apache.flume</groupId>
 <artifactId>flume-ng-configuration</artifactId>
 <version>1.6.0</version>
 </dependency>

\[java\]  [view plain][]  [copy][view plain]

1. /\* 
2.  \* Licensed to the Apache Software Foundation (ASF) under one 
3.  \* or more contributor license agreements.  See the NOTICE file 
4.  \* distributed with this work for additional information 
5.  \* regarding copyright ownership.  The ASF licenses this file 
6.  \* to you under the Apache License, Version 2.0 (the 
7.  \* "License"); you may not use this file except in compliance 
8.  \* with the License.  You may obtain a copy of the License at 
9.  \* 
10.  \* http://www.apache.org/licenses/LICENSE-2.0 
11.  \* 
12.  \* Unless required by applicable law or agreed to in writing, 
13.  \* software distributed under the License is distributed on an 
14.  \* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 
15.  \* KIND, either express or implied.  See the License for the 
16.  \* specific language governing permissions and limitations 
17.  \* under the License. 
18.  \*/  
19.   
20. **package** com.besttone.flume;  
21.   
22. **import** java.io.File;  
23. **import** java.io.FileFilter;  
24. **import** java.io.FileNotFoundException;  
25. **import** java.io.IOException;  
26. **import** java.nio.charset.Charset;  
27. **import** java.util.Arrays;  
28. **import** java.util.Collections;  
29. **import** java.util.Comparator;  
30. **import** java.util.List;  
31. **import** java.util.regex.Matcher;  
32. **import** java.util.regex.Pattern;  
33.   
34. **import** org.apache.flume.Context;  
35. **import** org.apache.flume.Event;  
36. **import** org.apache.flume.FlumeException;  
37. **import** org.apache.flume.annotations.InterfaceAudience;  
38. **import** org.apache.flume.annotations.InterfaceStability;  
39. **import** org.apache.flume.client.avro.ReliableEventReader;  
40. **import** org.apache.flume.serialization.DecodeErrorPolicy;  
41. **import** org.apache.flume.serialization.DurablePositionTracker;  
42. **import** org.apache.flume.serialization.EventDeserializer;  
43. **import** org.apache.flume.serialization.EventDeserializerFactory;  
44. **import** org.apache.flume.serialization.PositionTracker;  
45. **import** org.apache.flume.serialization.ResettableFileInputStream;  
46. **import** org.apache.flume.serialization.ResettableInputStream;  
47. **import** org.apache.flume.tools.PlatformDetect;  
48. **import** org.joda.time.DateTime;  
49. **import** org.joda.time.format.DateTimeFormat;  
50. **import** org.joda.time.format.DateTimeFormatter;  
51. **import** org.slf4j.Logger;  
52. **import** org.slf4j.LoggerFactory;  
53.   
54. **import** com.google.common.base.Charsets;  
55. **import** com.google.common.base.Optional;  
56. **import** com.google.common.base.Preconditions;  
57. **import** com.google.common.io.Files;  
58.   
59. /\*\* 
60.  \*  
61.  \* A \{@link ReliableEventReader\} which reads log data from files stored in a 
62.  \* spooling directory and renames each file once all of its data has been read 
63.  \* (through \{@link EventDeserializer\#readEvent()\} calls). The user must 
64.  \* \{@link \#commit()\} each read, to indicate that the lines have been fully 
65.  \* processed. 
66.  \*  
67.  \* Read calls will return no data if there are no files left to read. This 
68.  \* class, in general, is not thread safe. 
69.  \*  
70.  \*  
71.  \* This reader assumes that files with unique file names are left in the 
72.  \* spooling directory and not modified once they are placed there. Any user 
73.  \* behavior which violates these assumptions, when detected, will result in a 
74.  \* FlumeException being thrown. 
75.  \*  
76.  \*  
77.  \* This class makes the following guarantees, if above assumptions are met: 
78.  \* <ul> 
79.  \* <li>Once a log file has been renamed with the \{@link \#completedSuffix\}, all 
80.  \* of its records have been read through the 
81.  \* \{@link EventDeserializer\#readEvent()\} function and \{@link \#commit()\}ed at 
82.  \* least once. 
83.  \* <li>All files in the spooling directory will eventually be opened and 
84.  \* delivered to a \{@link \#readEvents(int)\} caller. 
85.  \* </ul> 
86.  \*/  
87. @InterfaceAudience.Private  
88. @InterfaceStability.Evolving  
89. **public** **class** ReliableSpoolingFileEventExtReader **implements** ReliableEventReader \{  
90.   
91.     **private** **static** **final** Logger logger = LoggerFactory  
92.             .getLogger(ReliableSpoolingFileEventExtReader.**class**);  
93.   
94.     **static** **final** String metaFileName = ".flumespool-main.meta";  
95.   
96.     **private** **final** File spoolDirectory;  
97.     **private** **final** String completedSuffix;  
98.     **private** **final** String deserializerType;  
99.     **private** **final** Context deserializerContext;  
100.     **private** **final** Pattern ignorePattern;  
101.     **private** **final** File metaFile;  
102.     **private** **final** **boolean** annotateFileName;  
103.     **private** **final** **boolean** annotateBaseName;  
104.     **private** **final** String fileNameHeader;  
105.     **private** **final** String baseNameHeader;  
106.   
107.     // 添加参数开始  
108.     **private** **final** **boolean** annotateFileNameExtractor;  
109.     **private** **final** String fileNameExtractorHeader;  
110.     **private** **final** Pattern fileNameExtractorPattern;  
111.     **private** **final** **boolean** convertToTimestamp;  
112.     **private** **final** String dateTimeFormat;  
113.   
114.     **private** **final** **boolean** splitFileName;  
115.     **private** **final** String splitBy;  
116.     **private** **final** String splitBaseNameHeader;  
117.     // 添加参数结束  
118.   
119.     **private** **final** String deletePolicy;  
120.     **private** **final** Charset inputCharset;  
121.     **private** **final** DecodeErrorPolicy decodeErrorPolicy;  
122.   
123.     **private** Optional<FileInfo> currentFile = Optional.absent();  
124.     /\*\* Always contains the last file from which lines have been read. \*\*/  
125.     **private** Optional<FileInfo> lastFileRead = Optional.absent();  
126.     **private** **boolean** committed = **true**;  
127.   
128.     /\*\* 
129.      \* Create a ReliableSpoolingFileEventReader to watch the given directory. 
130.      \*/  
131.     **private** ReliableSpoolingFileEventExtReader(File spoolDirectory,  
132.             String completedSuffix, String ignorePattern,  
133.             String trackerDirPath, **boolean** annotateFileName,  
134.             String fileNameHeader, **boolean** annotateBaseName,  
135.             String baseNameHeader, String deserializerType,  
136.             Context deserializerContext, String deletePolicy,  
137.             String inputCharset, DecodeErrorPolicy decodeErrorPolicy,  
138.             **boolean** annotateFileNameExtractor, String fileNameExtractorHeader,  
139.             String fileNameExtractorPattern, **boolean** convertToTimestamp,  
140.             String dateTimeFormat, **boolean** splitFileName, String splitBy,  
141.             String splitBaseNameHeader) **throws** IOException \{  
142.   
143.         // Sanity checks  
144.         Preconditions.checkNotNull(spoolDirectory);  
145.         Preconditions.checkNotNull(completedSuffix);  
146.         Preconditions.checkNotNull(ignorePattern);  
147.         Preconditions.checkNotNull(trackerDirPath);  
148.         Preconditions.checkNotNull(deserializerType);  
149.         Preconditions.checkNotNull(deserializerContext);  
150.         Preconditions.checkNotNull(deletePolicy);  
151.         Preconditions.checkNotNull(inputCharset);  
152.   
153.         // validate delete policy  
154.         **if** (!deletePolicy.equalsIgnoreCase(DeletePolicy.NEVER.name())  
155.                 && !deletePolicy  
156.                         .equalsIgnoreCase(DeletePolicy.IMMEDIATE.name())) \{  
157.             **throw** **new** IllegalArgumentException("Delete policies other than "  
158.                     + "NEVER and IMMEDIATE are not yet supported");  
159.         \}  
160.   
161.         **if** (logger.isDebugEnabled()) \{  
162.             logger.debug("Initializing \{\} with directory=\{\}, metaDir=\{\}, "  
163.                     + "deserializer=\{\}", **new** Object\[\] \{  
164.                     ReliableSpoolingFileEventExtReader.**class**.getSimpleName(),  
165.                     spoolDirectory, trackerDirPath, deserializerType \});  
166.         \}  
167.   
168.         // Verify directory exists and is readable/writable  
169.         Preconditions  
170.                 .checkState(  
171.                         spoolDirectory.exists(),  
172.                         "Directory does not exist: "  
173.                                 + spoolDirectory.getAbsolutePath());  
174.         Preconditions.checkState(spoolDirectory.isDirectory(),  
175.                 "Path is not a directory: " + spoolDirectory.getAbsolutePath());  
176.   
177.         // Do a canary test to make sure we have access to spooling directory  
178.         **try** \{  
179.             File canary = File.createTempFile("flume-spooldir-perm-check-",  
180.                     ".canary", spoolDirectory);  
181.             Files.write("testing flume file permissions\\n", canary,  
182.                     Charsets.UTF\_8);  
183.             List<String> lines = Files.readLines(canary, Charsets.UTF\_8);  
184.             Preconditions.checkState(!lines.isEmpty(), "Empty canary file %s",  
185.                     canary);  
186.             **if** (!canary.delete()) \{  
187.                 **throw** **new** IOException("Unable to delete canary file " + canary);  
188.             \}  
189.             logger.debug("Successfully created and deleted canary file: \{\}",  
190.                     canary);  
191.         \} **catch** (IOException e) \{  
192.             **throw** **new** FlumeException("Unable to read and modify files"  
193.                     + " in the spooling directory: " + spoolDirectory, e);  
194.         \}  
195.   
196.         **this**.spoolDirectory = spoolDirectory;  
197.         **this**.completedSuffix = completedSuffix;  
198.         **this**.deserializerType = deserializerType;  
199.         **this**.deserializerContext = deserializerContext;  
200.         **this**.annotateFileName = annotateFileName;  
201.         **this**.fileNameHeader = fileNameHeader;  
202.         **this**.annotateBaseName = annotateBaseName;  
203.         **this**.baseNameHeader = baseNameHeader;  
204.         **this**.ignorePattern = Pattern.compile(ignorePattern);  
205.         **this**.deletePolicy = deletePolicy;  
206.         **this**.inputCharset = Charset.forName(inputCharset);  
207.         **this**.decodeErrorPolicy = Preconditions.checkNotNull(decodeErrorPolicy);  
208.   
209.         // 增加代码开始  
210.         **this**.annotateFileNameExtractor = annotateFileNameExtractor;  
211.         **this**.fileNameExtractorHeader = fileNameExtractorHeader;  
212.         **this**.fileNameExtractorPattern = Pattern  
213.                 .compile(fileNameExtractorPattern);  
214.         **this**.convertToTimestamp = convertToTimestamp;  
215.         **this**.dateTimeFormat = dateTimeFormat;  
216.   
217.         **this**.splitFileName = splitFileName;  
218.         **this**.splitBy = splitBy;  
219.         **this**.splitBaseNameHeader = splitBaseNameHeader;  
220.         // 增加代码结束  
221.   
222.         File trackerDirectory = **new** File(trackerDirPath);  
223.   
224.         // if relative path, treat as relative to spool directory  
225.         **if** (!trackerDirectory.isAbsolute()) \{  
226.             trackerDirectory = **new** File(spoolDirectory, trackerDirPath);  
227.         \}  
228.   
229.         // ensure that meta directory exists  
230.         **if** (!trackerDirectory.exists()) \{  
231.             **if** (!trackerDirectory.mkdir()) \{  
232.                 **throw** **new** IOException(  
233.                         "Unable to mkdir nonexistent meta directory "  
234.                                 + trackerDirectory);  
235.             \}  
236.         \}  
237.   
238.         // ensure that the meta directory is a directory  
239.         **if** (!trackerDirectory.isDirectory()) \{  
240.             **throw** **new** IOException("Specified meta directory is not a directory"  
241.                     + trackerDirectory);  
242.         \}  
243.   
244.         **this**.metaFile = **new** File(trackerDirectory, metaFileName);  
245.     \}  
246.   
247.     /\*\* 
248.      \* Return the filename which generated the data from the last successful 
249.      \* \{@link \#readEvents(int)\} call. Returns null if called before any file 
250.      \* contents are read. 
251.      \*/  
252.     **public** String getLastFileRead() \{  
253.         **if** (!lastFileRead.isPresent()) \{  
254.             **return** **null**;  
255.         \}  
256.         **return** lastFileRead.get().getFile().getAbsolutePath();  
257.     \}  
258.   
259.     // public interface  
260.     **public** Event readEvent() **throws** IOException \{  
261.         List<Event> events = readEvents(1);  
262.         **if** (!events.isEmpty()) \{  
263.             **return** events.get(0);  
264.         \} **else** \{  
265.             **return** **null**;  
266.         \}  
267.     \}  
268.   
269.     **public** List<Event> readEvents(**int** numEvents) **throws** IOException \{  
270.         **if** (!committed) \{  
271.             **if** (!currentFile.isPresent()) \{  
272.                 **throw** **new** IllegalStateException("File should not roll when "  
273.                         + "commit is outstanding.");  
274.             \}  
275.             logger.info("Last read was never committed - resetting mark position.");  
276.             currentFile.get().getDeserializer().reset();  
277.         \} **else** \{  
278.             // Check if new files have arrived since last call  
279.             **if** (!currentFile.isPresent()) \{  
280.                 currentFile = getNextFile();  
281.             \}  
282.             // Return empty list if no new files  
283.             **if** (!currentFile.isPresent()) \{  
284.                 **return** Collections.emptyList();  
285.             \}  
286.         \}  
287.   
288.         EventDeserializer des = currentFile.get().getDeserializer();  
289.         List<Event> events = des.readEvents(numEvents);  
290.   
291.         /\* 
292.          \* It's possible that the last read took us just up to a file boundary. 
293.          \* If so, try to roll to the next file, if there is one. 
294.          \*/  
295.         **if** (events.isEmpty()) \{  
296.             retireCurrentFile();  
297.             currentFile = getNextFile();  
298.             **if** (!currentFile.isPresent()) \{  
299.                 **return** Collections.emptyList();  
300.             \}  
301.             events = currentFile.get().getDeserializer().readEvents(numEvents);  
302.         \}  
303.   
304.         **if** (annotateFileName) \{  
305.             String filename = currentFile.get().getFile().getAbsolutePath();  
306.             **for** (Event event : events) \{  
307.                 event.getHeaders().put(fileNameHeader, filename);  
308.             \}  
309.         \}  
310.   
311.         **if** (annotateBaseName) \{  
312.             String basename = currentFile.get().getFile().getName();  
313.             **for** (Event event : events) \{  
314.                 event.getHeaders().put(baseNameHeader, basename);  
315.             \}  
316.         \}  
317.   
318.         // 增加代码开始  
319.   
320.         // 按正则抽取文件名的内容  
321.         **if** (annotateFileNameExtractor) \{  
322.   
323.             Matcher matcher = fileNameExtractorPattern.matcher(currentFile  
324.                     .get().getFile().getName());  
325.   
326.             **if** (matcher.find()) \{  
327.                 String value = matcher.group();  
328.                 **if** (convertToTimestamp) \{  
329.                     DateTimeFormatter formatter = DateTimeFormat  
330.                             .forPattern(dateTimeFormat);  
331.                     DateTime dateTime = formatter.parseDateTime(value);  
332.   
333.                     value = Long.toString(dateTime.getMillis());  
334.                 \}  
335.   
336.                 **for** (Event event : events) \{  
337.                     event.getHeaders().put(fileNameExtractorHeader, value);  
338.                 \}  
339.             \}  
340.   
341.         \}  
342.   
343.         // 按分隔符拆分文件名  
344.         **if** (splitFileName) \{  
345.             String\[\] splits = currentFile.get().getFile().getName()  
346.                     .split(splitBy);  
347.   
348.             **for** (Event event : events) \{  
349.                 **for** (**int** i = 0; i < splits.length; i++) \{  
350.                     event.getHeaders().put(splitBaseNameHeader + i, splits\[i\]);  
351.                 \}  
352.   
353.             \}  
354.   
355.         \}  
356.   
357.         // 增加代码结束  
358.   
359.         committed = **false**;  
360.         lastFileRead = currentFile;  
361.         **return** events;  
362.     \}  
363.   
364.     @Override  
365.     **public** **void** close() **throws** IOException \{  
366.         **if** (currentFile.isPresent()) \{  
367.             currentFile.get().getDeserializer().close();  
368.             currentFile = Optional.absent();  
369.         \}  
370.     \}  
371.   
372.     /\*\* Commit the last lines which were read. \*/  
373.     @Override  
374.     **public** **void** commit() **throws** IOException \{  
375.         **if** (!committed && currentFile.isPresent()) \{  
376.             currentFile.get().getDeserializer().mark();  
377.             committed = **true**;  
378.         \}  
379.     \}  
380.   
381.     /\*\* 
382.      \* Closes currentFile and attempt to rename it. 
383.      \*  
384.      \* If these operations fail in a way that may cause duplicate log entries, 
385.      \* an error is logged but no exceptions are thrown. If these operations fail 
386.      \* in a way that indicates potential misuse of the spooling directory, a 
387.      \* FlumeException will be thrown. 
388.      \*  
389.      \* @throws FlumeException 
390.      \*             if files do not conform to spooling assumptions 
391.      \*/  
392.     **private** **void** retireCurrentFile() **throws** IOException \{  
393.         Preconditions.checkState(currentFile.isPresent());  
394.   
395.         File fileToRoll = **new** File(currentFile.get().getFile()  
396.                 .getAbsolutePath());  
397.   
398.         currentFile.get().getDeserializer().close();  
399.   
400.         // Verify that spooling assumptions hold  
401.         **if** (fileToRoll.lastModified() != currentFile.get().getLastModified()) \{  
402.             String message = "File has been modified since being read: "  
403.                     + fileToRoll;  
404.             **throw** **new** IllegalStateException(message);  
405.         \}  
406.         **if** (fileToRoll.length() != currentFile.get().getLength()) \{  
407.             String message = "File has changed size since being read: "  
408.                     + fileToRoll;  
409.             **throw** **new** IllegalStateException(message);  
410.         \}  
411.   
412.         **if** (deletePolicy.equalsIgnoreCase(DeletePolicy.NEVER.name())) \{  
413.             rollCurrentFile(fileToRoll);  
414.         \} **else** **if** (deletePolicy.equalsIgnoreCase(DeletePolicy.IMMEDIATE.name())) \{  
415.             deleteCurrentFile(fileToRoll);  
416.         \} **else** \{  
417.             // TODO: implement delay in the future  
418.             **throw** **new** IllegalArgumentException("Unsupported delete policy: "  
419.                     + deletePolicy);  
420.         \}  
421.     \}  
422.   
423.     /\*\* 
424.      \* Rename the given spooled file 
425.      \*  
426.      \* @param fileToRoll 
427.      \* @throws IOException 
428.      \*/  
429.     **private** **void** rollCurrentFile(File fileToRoll) **throws** IOException \{  
430.   
431.         File dest = **new** File(fileToRoll.getPath() + completedSuffix);  
432.         logger.info("Preparing to move file \{\} to \{\}", fileToRoll, dest);  
433.   
434.         // Before renaming, check whether destination file name exists  
435.         **if** (dest.exists() && PlatformDetect.isWindows()) \{  
436.             /\* 
437.              \* If we are here, it means the completed file already exists. In 
438.              \* almost every case this means the user is violating an assumption 
439.              \* of Flume (that log files are placed in the spooling directory 
440.              \* with unique names). However, there is a corner case on Windows 
441.              \* systems where the file was already rolled but the rename was not 
442.              \* atomic. If that seems likely, we let it pass with only a warning. 
443.              \*/  
444.             **if** (Files.equal(currentFile.get().getFile(), dest)) \{  
445.                 logger.warn("Completed file " + dest  
446.                         + " already exists, but files match, so continuing.");  
447.                 **boolean** deleted = fileToRoll.delete();  
448.                 **if** (!deleted) \{  
449.                     logger.error("Unable to delete file "  
450.                             + fileToRoll.getAbsolutePath()  
451.                             + ". It will likely be ingested another time.");  
452.                 \}  
453.             \} **else** \{  
454.                 String message = "File name has been re-used with different"  
455.                         + " files. Spooling assumptions violated for " + dest;  
456.                 **throw** **new** IllegalStateException(message);  
457.             \}  
458.   
459.             // Dest file exists and not on windows  
460.         \} **else** **if** (dest.exists()) \{  
461.             String message = "File name has been re-used with different"  
462.                     + " files. Spooling assumptions violated for " + dest;  
463.             **throw** **new** IllegalStateException(message);  
464.   
465.             // Destination file does not already exist. We are good to go!  
466.         \} **else** \{  
467.             **boolean** renamed = fileToRoll.renameTo(dest);  
468.             **if** (renamed) \{  
469.                 logger.debug("Successfully rolled file \{\} to \{\}", fileToRoll,  
470.                         dest);  
471.   
472.                 // now we no longer need the meta file  
473.                 deleteMetaFile();  
474.             \} **else** \{  
475.                 /\* 
476.                  \* If we are here then the file cannot be renamed for a reason 
477.                  \* other than that the destination file exists (actually, that 
478.                  \* remains possible w/ small probability due to TOC-TOU 
479.                  \* conditions). 
480.                  \*/  
481.                 String message = "Unable to move "  
482.                         + fileToRoll  
483.                         + " to "  
484.                         + dest  
485.                         + ". This will likely cause duplicate events. Please verify that "  
486.                         + "flume has sufficient permissions to perform these operations.";  
487.                 **throw** **new** FlumeException(message);  
488.             \}  
489.         \}  
490.     \}  
491.   
492.     /\*\* 
493.      \* Delete the given spooled file 
494.      \*  
495.      \* @param fileToDelete 
496.      \* @throws IOException 
497.      \*/  
498.     **private** **void** deleteCurrentFile(File fileToDelete) **throws** IOException \{  
499.         logger.info("Preparing to delete file \{\}", fileToDelete);  
500.         **if** (!fileToDelete.exists()) \{  
501.             logger.warn("Unable to delete nonexistent file: \{\}", fileToDelete);  
502.             **return**;  
503.         \}  
504.         **if** (!fileToDelete.delete()) \{  
505.             **throw** **new** IOException("Unable to delete spool file: "  
506.                     + fileToDelete);  
507.         \}  
508.         // now we no longer need the meta file  
509.         deleteMetaFile();  
510.     \}  
511.   
512.     /\*\* 
513.      \* Find and open the oldest file in the chosen directory. If two or more 
514.      \* files are equally old, the file name with lower lexicographical value is 
515.      \* returned. If the directory is empty, this will return an absent option. 
516.      \*/  
517.     **private** Optional<FileInfo> getNextFile() \{  
518.         /\* Filter to exclude finished or hidden files \*/  
519.         FileFilter filter = **new** FileFilter() \{  
520.             **public** **boolean** accept(File candidate) \{  
521.                 String fileName = candidate.getName();  
522.                 **if** ((candidate.isDirectory())  
523.                         || (fileName.endsWith(completedSuffix))  
524.                         || (fileName.startsWith("."))  
525.                         || ignorePattern.matcher(fileName).matches()) \{  
526.                     **return** **false**;  
527.                 \}  
528.                 **return** **true**;  
529.             \}  
530.         \};  
531.         List<File> candidateFiles = Arrays.asList(spoolDirectory  
532.                 .listFiles(filter));  
533.         **if** (candidateFiles.isEmpty()) \{  
534.             **return** Optional.absent();  
535.         \} **else** \{  
536.             Collections.sort(candidateFiles, **new** Comparator<File>() \{  
537.                 **public** **int** compare(File a, File b) \{  
538.                     **int** timeComparison = **new** Long(a.lastModified())  
539.                             .compareTo(**new** Long(b.lastModified()));  
540.                     **if** (timeComparison != 0) \{  
541.                         **return** timeComparison;  
542.                     \} **else** \{  
543.                         **return** a.getName().compareTo(b.getName());  
544.                     \}  
545.                 \}  
546.             \});  
547.             File nextFile = candidateFiles.get(0);  
548.             **try** \{  
549.                 // roll the meta file, if needed  
550.                 String nextPath = nextFile.getPath();  
551.                 PositionTracker tracker = DurablePositionTracker.getInstance(  
552.                         metaFile, nextPath);  
553.                 **if** (!tracker.getTarget().equals(nextPath)) \{  
554.                     tracker.close();  
555.                     deleteMetaFile();  
556.                     tracker = DurablePositionTracker.getInstance(metaFile,  
557.                             nextPath);  
558.                 \}  
559.   
560.                 // sanity check  
561.                 Preconditions  
562.                         .checkState(  
563.                                 tracker.getTarget().equals(nextPath),  
564.                                 "Tracker target %s does not equal expected filename %s",  
565.                                 tracker.getTarget(), nextPath);  
566.   
567.                 ResettableInputStream in = **new** ResettableFileInputStream(  
568.                         nextFile, tracker,  
569.                         ResettableFileInputStream.DEFAULT\_BUF\_SIZE,  
570.                         inputCharset, decodeErrorPolicy);  
571.                 EventDeserializer deserializer = EventDeserializerFactory  
572.                         .getInstance(deserializerType, deserializerContext, in);  
573.   
574.                 **return** Optional.of(**new** FileInfo(nextFile, deserializer));  
575.             \} **catch** (FileNotFoundException e) \{  
576.                 // File could have been deleted in the interim  
577.                 logger.warn("Could not find file: " + nextFile, e);  
578.                 **return** Optional.absent();  
579.             \} **catch** (IOException e) \{  
580.                 logger.error("Exception opening file: " + nextFile, e);  
581.                 **return** Optional.absent();  
582.             \}  
583.         \}  
584.     \}  
585.   
586.     **private** **void** deleteMetaFile() **throws** IOException \{  
587.         **if** (metaFile.exists() && !metaFile.delete()) \{  
588.             **throw** **new** IOException("Unable to delete old meta file " + metaFile);  
589.         \}  
590.     \}  
591.   
592.     /\*\* An immutable class with information about a file being processed. \*/  
593.     **private** **static** **class** FileInfo \{  
594.         **private** **final** File file;  
595.         **private** **final** **long** length;  
596.         **private** **final** **long** lastModified;  
597.         **private** **final** EventDeserializer deserializer;  
598.   
599.         **public** FileInfo(File file, EventDeserializer deserializer) \{  
600.             **this**.file = file;  
601.             **this**.length = file.length();  
602.             **this**.lastModified = file.lastModified();  
603.             **this**.deserializer = deserializer;  
604.         \}  
605.   
606.         **public** **long** getLength() \{  
607.             **return** length;  
608.         \}  
609.   
610.         **public** **long** getLastModified() \{  
611.             **return** lastModified;  
612.         \}  
613.   
614.         **public** EventDeserializer getDeserializer() \{  
615.             **return** deserializer;  
616.         \}  
617.   
618.         **public** File getFile() \{  
619.             **return** file;  
620.         \}  
621.     \}  
622.   
623.     @InterfaceAudience.Private  
624.     @InterfaceStability.Unstable  
625.     **static** **enum** DeletePolicy \{  
626.         NEVER, IMMEDIATE, DELAY  
627.     \}  
628.   
629.     /\*\* 
630.      \* Special builder class for ReliableSpoolingFileEventReader 
631.      \*/  
632.     **public** **static** **class** Builder \{  
633.         **private** File spoolDirectory;  
634.         **private** String completedSuffix = SpoolDirectorySourceConfigurationExtConstants.SPOOLED\_FILE\_SUFFIX;  
635.         **private** String ignorePattern = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_IGNORE\_PAT;  
636.         **private** String trackerDirPath = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_TRACKER\_DIR;  
637.         **private** Boolean annotateFileName = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_FILE\_HEADER;  
638.         **private** String fileNameHeader = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_FILENAME\_HEADER\_KEY;  
639.         **private** Boolean annotateBaseName = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_BASENAME\_HEADER;  
640.         **private** String baseNameHeader = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_BASENAME\_HEADER\_KEY;  
641.         **private** String deserializerType = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_DESERIALIZER;  
642.         **private** Context deserializerContext = **new** Context();  
643.         **private** String deletePolicy = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_DELETE\_POLICY;  
644.         **private** String inputCharset = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_INPUT\_CHARSET;  
645.         **private** DecodeErrorPolicy decodeErrorPolicy = DecodeErrorPolicy  
646.                 .valueOf(SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_DECODE\_ERROR\_POLICY  
647.                         .toUpperCase());  
648.   
649.         // 增加代码开始  
650.   
651.         **private** Boolean annotateFileNameExtractor = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_FILENAME\_EXTRACTOR;  
652.         **private** String fileNameExtractorHeader = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_FILENAME\_EXTRACTOR\_HEADER\_KEY;  
653.         **private** String fileNameExtractorPattern = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_FILENAME\_EXTRACTOR\_PATTERN;  
654.         **private** Boolean convertToTimestamp = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_FILENAME\_EXTRACTOR\_CONVERT\_TO\_TIMESTAMP;  
655.   
656.         **private** String dateTimeFormat = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_FILENAME\_EXTRACTOR\_DATETIME\_FORMAT;  
657.   
658.         **private** Boolean splitFileName = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_SPLIT\_FILENAME;  
659.         **private** String splitBy = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_SPLITY\_BY;  
660.         **private** String splitBaseNameHeader = SpoolDirectorySourceConfigurationExtConstants.DEFAULT\_SPLIT\_BASENAME\_HEADER;  
661.   
662.         **public** Builder annotateFileNameExtractor(  
663.                 Boolean annotateFileNameExtractor) \{  
664.             **this**.annotateFileNameExtractor = annotateFileNameExtractor;  
665.             **return** **this**;  
666.         \}  
667.   
668.         **public** Builder fileNameExtractorHeader(String fileNameExtractorHeader) \{  
669.             **this**.fileNameExtractorHeader = fileNameExtractorHeader;  
670.             **return** **this**;  
671.         \}  
672.   
673.         **public** Builder fileNameExtractorPattern(String fileNameExtractorPattern) \{  
674.             **this**.fileNameExtractorPattern = fileNameExtractorPattern;  
675.             **return** **this**;  
676.         \}  
677.   
678.         **public** Builder convertToTimestamp(Boolean convertToTimestamp) \{  
679.             **this**.convertToTimestamp = convertToTimestamp;  
680.             **return** **this**;  
681.         \}  
682.   
683.         **public** Builder dateTimeFormat(String dateTimeFormat) \{  
684.             **this**.dateTimeFormat = dateTimeFormat;  
685.             **return** **this**;  
686.         \}  
687.   
688.         **public** Builder splitFileName(Boolean splitFileName) \{  
689.             **this**.splitFileName = splitFileName;  
690.             **return** **this**;  
691.         \}  
692.   
693.         **public** Builder splitBy(String splitBy) \{  
694.             **this**.splitBy = splitBy;  
695.             **return** **this**;  
696.         \}  
697.   
698.         **public** Builder splitBaseNameHeader(String splitBaseNameHeader) \{  
699.             **this**.splitBaseNameHeader = splitBaseNameHeader;  
700.             **return** **this**;  
701.         \}  
702.   
703.         // 增加代码结束  
704.   
705.         **public** Builder spoolDirectory(File directory) \{  
706.             **this**.spoolDirectory = directory;  
707.             **return** **this**;  
708.         \}  
709.   
710.         **public** Builder completedSuffix(String completedSuffix) \{  
711.             **this**.completedSuffix = completedSuffix;  
712.             **return** **this**;  
713.         \}  
714.   
715.         **public** Builder ignorePattern(String ignorePattern) \{  
716.             **this**.ignorePattern = ignorePattern;  
717.             **return** **this**;  
718.         \}  
719.   
720.         **public** Builder trackerDirPath(String trackerDirPath) \{  
721.             **this**.trackerDirPath = trackerDirPath;  
722.             **return** **this**;  
723.         \}  
724.   
725.         **public** Builder annotateFileName(Boolean annotateFileName) \{  
726.             **this**.annotateFileName = annotateFileName;  
727.             **return** **this**;  
728.         \}  
729.   
730.         **public** Builder fileNameHeader(String fileNameHeader) \{  
731.             **this**.fileNameHeader = fileNameHeader;  
732.             **return** **this**;  
733.         \}  
734.   
735.         **public** Builder annotateBaseName(Boolean annotateBaseName) \{  
736.             **this**.annotateBaseName = annotateBaseName;  
737.             **return** **this**;  
738.         \}  
739.   
740.         **public** Builder baseNameHeader(String baseNameHeader) \{  
741.             **this**.baseNameHeader = baseNameHeader;  
742.             **return** **this**;  
743.         \}  
744.   
745.         **public** Builder deserializerType(String deserializerType) \{  
746.             **this**.deserializerType = deserializerType;  
747.             **return** **this**;  
748.         \}  
749.   
750.         **public** Builder deserializerContext(Context deserializerContext) \{  
751.             **this**.deserializerContext = deserializerContext;  
752.             **return** **this**;  
753.         \}  
754.   
755.         **public** Builder deletePolicy(String deletePolicy) \{  
756.             **this**.deletePolicy = deletePolicy;  
757.             **return** **this**;  
758.         \}  
759.   
760.         **public** Builder inputCharset(String inputCharset) \{  
761.             **this**.inputCharset = inputCharset;  
762.             **return** **this**;  
763.         \}  
764.   
765.         **public** Builder decodeErrorPolicy(DecodeErrorPolicy decodeErrorPolicy) \{  
766.             **this**.decodeErrorPolicy = decodeErrorPolicy;  
767.             **return** **this**;  
768.         \}  
769.   
770.         **public** ReliableSpoolingFileEventExtReader build() **throws** IOException \{  
771.             **return** **new** ReliableSpoolingFileEventExtReader(spoolDirectory,  
772.                     completedSuffix, ignorePattern, trackerDirPath,  
773.                     annotateFileName, fileNameHeader, annotateBaseName,  
774.                     baseNameHeader, deserializerType, deserializerContext,  
775.                     deletePolicy, inputCharset, decodeErrorPolicy,  
776.                     annotateFileNameExtractor, fileNameExtractorHeader,  
777.                     fileNameExtractorPattern, convertToTimestamp,  
778.                     dateTimeFormat, splitFileName, splitBy, splitBaseNameHeader);  
779.         \}  
780.     \}  
781.   
782. \}

\[java\]  [view plain][]  [copy][view plain]

1.  /\* 
2.   \* Licensed to the Apache Software Foundation (ASF) under one or more 
3.   \* contributor license agreements. See the NOTICE file distributed with this 
4.   \* work for additional information regarding copyright ownership. The ASF 
5.   \* licenses this file to you under the Apache License, Version 2.0 (the 
6.   \* "License"); you may not use this file except in compliance with the License. 
7.   \* You may obtain a copy of the License at 
8.   \* 
9.   \* http://www.apache.org/licenses/LICENSE-2.0 
10.  \* 
11.  \* Unless required by applicable law or agreed to in writing, software 
12.  \* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 
13.  \* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 
14.  \* License for the specific language governing permissions and limitations under 
15.  \* the License. 
16.  \*/  
17.   
18. **package** com.besttone.flume;  
19.   
20. **import** org.apache.flume.serialization.DecodeErrorPolicy;  
21.   
22. **public** **class** SpoolDirectorySourceConfigurationExtConstants \{  
23.     /\*\* Directory where files are deposited. \*/  
24.     **public** **static** **final** String SPOOL\_DIRECTORY = "spoolDir";  
25.   
26.     /\*\* Suffix appended to files when they are finished being sent. \*/  
27.     **public** **static** **final** String SPOOLED\_FILE\_SUFFIX = "fileSuffix";  
28.     **public** **static** **final** String DEFAULT\_SPOOLED\_FILE\_SUFFIX = ".COMPLETED";  
29.   
30.     /\*\* Header in which to put absolute path filename. \*/  
31.     **public** **static** **final** String FILENAME\_HEADER\_KEY = "fileHeaderKey";  
32.     **public** **static** **final** String DEFAULT\_FILENAME\_HEADER\_KEY = "file";  
33.   
34.     /\*\* Whether to include absolute path filename in a header. \*/  
35.     **public** **static** **final** String FILENAME\_HEADER = "fileHeader";  
36.     **public** **static** **final** **boolean** DEFAULT\_FILE\_HEADER = **false**;  
37.   
38.     /\*\* Header in which to put the basename of file. \*/  
39.     **public** **static** **final** String BASENAME\_HEADER\_KEY = "basenameHeaderKey";  
40.     **public** **static** **final** String DEFAULT\_BASENAME\_HEADER\_KEY = "basename";  
41.   
42.     /\*\* Whether to include the basename of a file in a header. \*/  
43.     **public** **static** **final** String BASENAME\_HEADER = "basenameHeader";  
44.     **public** **static** **final** **boolean** DEFAULT\_BASENAME\_HEADER = **false**;  
45.   
46.     /\*\* What size to batch with before sending to ChannelProcessor. \*/  
47.     **public** **static** **final** String BATCH\_SIZE = "batchSize";  
48.     **public** **static** **final** **int** DEFAULT\_BATCH\_SIZE = 100;  
49.   
50.     /\*\* Maximum number of lines to buffer between commits. \*/  
51.     @Deprecated  
52.     **public** **static** **final** String BUFFER\_MAX\_LINES = "bufferMaxLines";  
53.     @Deprecated  
54.     **public** **static** **final** **int** DEFAULT\_BUFFER\_MAX\_LINES = 100;  
55.   
56.     /\*\* Maximum length of line (in characters) in buffer between commits. \*/  
57.     @Deprecated  
58.     **public** **static** **final** String BUFFER\_MAX\_LINE\_LENGTH = "bufferMaxLineLength";  
59.     @Deprecated  
60.     **public** **static** **final** **int** DEFAULT\_BUFFER\_MAX\_LINE\_LENGTH = 5000;  
61.   
62.     /\*\* Pattern of files to ignore \*/  
63.     **public** **static** **final** String IGNORE\_PAT = "ignorePattern";  
64.     **public** **static** **final** String DEFAULT\_IGNORE\_PAT = "^$"; // no effect  
65.   
66.     /\*\* Directory to store metadata about files being processed \*/  
67.     **public** **static** **final** String TRACKER\_DIR = "trackerDir";  
68.     **public** **static** **final** String DEFAULT\_TRACKER\_DIR = ".flumespool";  
69.   
70.     /\*\* Deserializer to use to parse the file data into Flume Events \*/  
71.     **public** **static** **final** String DESERIALIZER = "deserializer";  
72.     **public** **static** **final** String DEFAULT\_DESERIALIZER = "LINE";  
73.   
74.     **public** **static** **final** String DELETE\_POLICY = "deletePolicy";  
75.     **public** **static** **final** String DEFAULT\_DELETE\_POLICY = "never";  
76.   
77.     /\*\* Character set used when reading the input. \*/  
78.     **public** **static** **final** String INPUT\_CHARSET = "inputCharset";  
79.     **public** **static** **final** String DEFAULT\_INPUT\_CHARSET = "UTF-8";  
80.   
81.     /\*\* What to do when there is a character set decoding error. \*/  
82.     **public** **static** **final** String DECODE\_ERROR\_POLICY = "decodeErrorPolicy";  
83.     **public** **static** **final** String DEFAULT\_DECODE\_ERROR\_POLICY = DecodeErrorPolicy.FAIL  
84.             .name();  
85.   
86.     **public** **static** **final** String MAX\_BACKOFF = "maxBackoff";  
87.   
88.     **public** **static** **final** Integer DEFAULT\_MAX\_BACKOFF = 4000;  
89.   
90.     // 增加代码开始  
91.     **public** **static** **final** String FILENAME\_EXTRACTOR = "fileExtractor";  
92.     **public** **static** **final** **boolean** DEFAULT\_FILENAME\_EXTRACTOR = **false**;  
93.   
94.     **public** **static** **final** String FILENAME\_EXTRACTOR\_HEADER\_KEY = "fileExtractorHeaderKey";  
95.     **public** **static** **final** String DEFAULT\_FILENAME\_EXTRACTOR\_HEADER\_KEY = "fileExtractorHeader";  
96.   
97.     **public** **static** **final** String FILENAME\_EXTRACTOR\_PATTERN = "fileExtractorPattern";  
98.     **public** **static** **final** String DEFAULT\_FILENAME\_EXTRACTOR\_PATTERN = "(.)\*";  
99.   
100.     **public** **static** **final** String FILENAME\_EXTRACTOR\_CONVERT\_TO\_TIMESTAMP = "convertToTimestamp";  
101.     **public** **static** **final** **boolean** DEFAULT\_FILENAME\_EXTRACTOR\_CONVERT\_TO\_TIMESTAMP = **false**;  
102.   
103.     **public** **static** **final** String FILENAME\_EXTRACTOR\_DATETIME\_FORMAT = "dateTimeFormat";  
104.     **public** **static** **final** String DEFAULT\_FILENAME\_EXTRACTOR\_DATETIME\_FORMAT = "yyyy-MM-dd";  
105.   
106.   
107.     **public** **static** **final** String SPLIT\_FILENAME = "splitFileName";  
108.     **public** **static** **final** **boolean** DEFAULT\_SPLIT\_FILENAME = **false**;  
109.   
110.     **public** **static** **final** String SPLITY\_BY = "splitBy";  
111.     **public** **static** **final** String DEFAULT\_SPLITY\_BY = "\\\\.";  
112.   
113.     **public** **static** **final** String SPLIT\_BASENAME\_HEADER = "splitBaseNameHeader";  
114.     **public** **static** **final** String DEFAULT\_SPLIT\_BASENAME\_HEADER = "fileNameSplit";  
115.     // 增加代码结束  
116.   
117. \}

\[java\]  [view plain][]  [copy][view plain]

1. **package** com.besttone.flume;  
2.   
3. **import** **static** com.besttone.flume.SpoolDirectorySourceConfigurationExtConstants.\*;  
4.   
5.   
6. **import** java.io.File;  
7. **import** java.io.IOException;  
8. **import** java.util.List;  
9. **import** java.util.concurrent.Executors;  
10. **import** java.util.concurrent.ScheduledExecutorService;  
11. **import** java.util.concurrent.TimeUnit;  
12.   
13. **import** org.apache.flume.ChannelException;  
14. **import** org.apache.flume.Context;  
15. **import** org.apache.flume.Event;  
16. **import** org.apache.flume.EventDrivenSource;  
17. **import** org.apache.flume.FlumeException;  
18. **import** org.apache.flume.conf.Configurable;  
19. **import** org.apache.flume.instrumentation.SourceCounter;  
20. **import** org.apache.flume.serialization.DecodeErrorPolicy;  
21. **import** org.apache.flume.serialization.LineDeserializer;  
22. **import** org.apache.flume.source.AbstractSource;  
23. **import** org.slf4j.Logger;  
24. **import** org.slf4j.LoggerFactory;  
25.   
26. **import** com.google.common.annotations.VisibleForTesting;  
27. **import** com.google.common.base.Preconditions;  
28. **import** com.google.common.base.Throwables;  
29.   
30. **public** **class** SpoolDirectoryExtSource **extends** AbstractSource **implements**  
31.         Configurable, EventDrivenSource \{  
32.   
33.     **private** **static** **final** Logger logger = LoggerFactory  
34.             .getLogger(SpoolDirectoryExtSource.**class**);  
35.   
36.     // Delay used when polling for new files  
37.     **private** **static** **final** **int** POLL\_DELAY\_MS = 500;  
38.   
39.     /\* Config options \*/  
40.     **private** String completedSuffix;  
41.     **private** String spoolDirectory;  
42.     **private** **boolean** fileHeader;  
43.     **private** String fileHeaderKey;  
44.     **private** **boolean** basenameHeader;  
45.     **private** String basenameHeaderKey;  
46.     **private** **int** batchSize;  
47.     **private** String ignorePattern;  
48.     **private** String trackerDirPath;  
49.     **private** String deserializerType;  
50.     **private** Context deserializerContext;  
51.     **private** String deletePolicy;  
52.     **private** String inputCharset;  
53.     **private** DecodeErrorPolicy decodeErrorPolicy;  
54.     **private** **volatile** **boolean** hasFatalError = **false**;  
55.   
56.     **private** SourceCounter sourceCounter;  
57.     ReliableSpoolingFileEventExtReader reader;  
58.     **private** ScheduledExecutorService executor;  
59.     **private** **boolean** backoff = **true**;  
60.     **private** **boolean** hitChannelException = **false**;  
61.     **private** **int** maxBackoff;  
62.   
63.     // 增加代码开始  
64.   
65.     **private** Boolean annotateFileNameExtractor;  
66.     **private** String fileNameExtractorHeader;  
67.     **private** String fileNameExtractorPattern;  
68.     **private** Boolean convertToTimestamp;  
69.   
70.     **private** String dateTimeFormat;  
71.       
72.     **private** **boolean** splitFileName;  
73.     **private**  String splitBy;  
74.     **private**  String splitBaseNameHeader;  
75.   
76.     // 增加代码结束  
77.   
78.     @Override  
79.     **public** **synchronized** **void** start() \{  
80.         logger.info("SpoolDirectorySource source starting with directory: \{\}",  
81.                 spoolDirectory);  
82.   
83.         executor = Executors.newSingleThreadScheduledExecutor();  
84.   
85.         File directory = **new** File(spoolDirectory);  
86.         **try** \{  
87.             reader = **new** ReliableSpoolingFileEventExtReader.Builder()  
88.                     .spoolDirectory(directory).completedSuffix(completedSuffix)  
89.                     .ignorePattern(ignorePattern)  
90.                     .trackerDirPath(trackerDirPath)  
91.                     .annotateFileName(fileHeader).fileNameHeader(fileHeaderKey)  
92.                     .annotateBaseName(basenameHeader)  
93.                     .baseNameHeader(basenameHeaderKey)  
94.                     .deserializerType(deserializerType)  
95.                     .deserializerContext(deserializerContext)  
96.                     .deletePolicy(deletePolicy).inputCharset(inputCharset)  
97.                     .decodeErrorPolicy(decodeErrorPolicy)  
98.                     .annotateFileNameExtractor(annotateFileNameExtractor)  
99.                     .fileNameExtractorHeader(fileNameExtractorHeader)  
100.                     .fileNameExtractorPattern(fileNameExtractorPattern)  
101.                     .convertToTimestamp(convertToTimestamp)  
102.                     .dateTimeFormat(dateTimeFormat)  
103.                     .splitFileName(splitFileName).splitBy(splitBy)  
104.                     .splitBaseNameHeader(splitBaseNameHeader).build();  
105.         \} **catch** (IOException ioe) \{  
106.             **throw** **new** FlumeException(  
107.                     "Error instantiating spooling event parser", ioe);  
108.         \}  
109.   
110.         Runnable runner = **new** SpoolDirectoryRunnable(reader, sourceCounter);  
111.         executor.scheduleWithFixedDelay(runner, 0, POLL\_DELAY\_MS,  
112.                 TimeUnit.MILLISECONDS);  
113.   
114.         **super**.start();  
115.         logger.debug("SpoolDirectorySource source started");  
116.         sourceCounter.start();  
117.     \}  
118.   
119.     @Override  
120.     **public** **synchronized** **void** stop() \{  
121.         executor.shutdown();  
122.         **try** \{  
123.             executor.awaitTermination(10L, TimeUnit.SECONDS);  
124.         \} **catch** (InterruptedException ex) \{  
125.             logger.info("Interrupted while awaiting termination", ex);  
126.         \}  
127.         executor.shutdownNow();  
128.   
129.         **super**.stop();  
130.         sourceCounter.stop();  
131.         logger.info("SpoolDir source \{\} stopped. Metrics: \{\}", getName(),  
132.                 sourceCounter);  
133.     \}  
134.   
135.     @Override  
136.     **public** String toString() \{  
137.         **return** "Spool Directory source " + getName() + ": \{ spoolDir: "  
138.                 + spoolDirectory + " \}";  
139.     \}  
140.   
141.     @Override  
142.     **public** **synchronized** **void** configure(Context context) \{  
143.         spoolDirectory = context.getString(SPOOL\_DIRECTORY);  
144.         Preconditions.checkState(spoolDirectory != **null**,  
145.                 "Configuration must specify a spooling directory");  
146.   
147.         completedSuffix = context.getString(SPOOLED\_FILE\_SUFFIX,  
148.                 DEFAULT\_SPOOLED\_FILE\_SUFFIX);  
149.         deletePolicy = context.getString(DELETE\_POLICY, DEFAULT\_DELETE\_POLICY);  
150.         fileHeader = context.getBoolean(FILENAME\_HEADER, DEFAULT\_FILE\_HEADER);  
151.         fileHeaderKey = context.getString(FILENAME\_HEADER\_KEY,  
152.                 DEFAULT\_FILENAME\_HEADER\_KEY);  
153.         basenameHeader = context.getBoolean(BASENAME\_HEADER,  
154.                 DEFAULT\_BASENAME\_HEADER);  
155.         basenameHeaderKey = context.getString(BASENAME\_HEADER\_KEY,  
156.                 DEFAULT\_BASENAME\_HEADER\_KEY);  
157.         batchSize = context.getInteger(BATCH\_SIZE, DEFAULT\_BATCH\_SIZE);  
158.         inputCharset = context.getString(INPUT\_CHARSET, DEFAULT\_INPUT\_CHARSET);  
159.         decodeErrorPolicy = DecodeErrorPolicy  
160.                 .valueOf(context.getString(DECODE\_ERROR\_POLICY,  
161.                         DEFAULT\_DECODE\_ERROR\_POLICY).toUpperCase());  
162.   
163.         ignorePattern = context.getString(IGNORE\_PAT, DEFAULT\_IGNORE\_PAT);  
164.         trackerDirPath = context.getString(TRACKER\_DIR, DEFAULT\_TRACKER\_DIR);  
165.   
166.         deserializerType = context  
167.                 .getString(DESERIALIZER, DEFAULT\_DESERIALIZER);  
168.         deserializerContext = **new** Context(context.getSubProperties(DESERIALIZER  
169.                 + "."));  
170.   
171.         // "Hack" to support backwards compatibility with previous generation of  
172.         // spooling directory source, which did not support deserializers  
173.         Integer bufferMaxLineLength = context  
174.                 .getInteger(BUFFER\_MAX\_LINE\_LENGTH);  
175.         **if** (bufferMaxLineLength != **null** && deserializerType != **null**  
176.                 && deserializerType.equalsIgnoreCase(DEFAULT\_DESERIALIZER)) \{  
177.             deserializerContext.put(LineDeserializer.MAXLINE\_KEY,  
178.                     bufferMaxLineLength.toString());  
179.         \}  
180.   
181.         maxBackoff = context.getInteger(MAX\_BACKOFF, DEFAULT\_MAX\_BACKOFF);  
182.         **if** (sourceCounter == **null**) \{  
183.             sourceCounter = **new** SourceCounter(getName());  
184.         \}  
185.           
186.         //增加代码开始  
187.   
188.         annotateFileNameExtractor=context.getBoolean(FILENAME\_EXTRACTOR, DEFAULT\_FILENAME\_EXTRACTOR);  
189.         fileNameExtractorHeader=context.getString(FILENAME\_EXTRACTOR\_HEADER\_KEY, DEFAULT\_FILENAME\_EXTRACTOR\_HEADER\_KEY);  
190.         fileNameExtractorPattern=context.getString(FILENAME\_EXTRACTOR\_PATTERN,DEFAULT\_FILENAME\_EXTRACTOR\_PATTERN);  
191.         convertToTimestamp=context.getBoolean(FILENAME\_EXTRACTOR\_CONVERT\_TO\_TIMESTAMP, DEFAULT\_FILENAME\_EXTRACTOR\_CONVERT\_TO\_TIMESTAMP);  
192.         dateTimeFormat=context.getString(FILENAME\_EXTRACTOR\_DATETIME\_FORMAT, DEFAULT\_FILENAME\_EXTRACTOR\_DATETIME\_FORMAT);  
193.           
194.         splitFileName=context.getBoolean(SPLIT\_FILENAME, DEFAULT\_SPLIT\_FILENAME);  
195.         splitBy=context.getString(SPLITY\_BY, DEFAULT\_SPLITY\_BY);  
196.         splitBaseNameHeader=context.getString(SPLIT\_BASENAME\_HEADER, DEFAULT\_SPLIT\_BASENAME\_HEADER);  
197.           
198.           
199.           
200.         //增加代码结束   
201.     \}  
202.   
203.     @VisibleForTesting  
204.     **protected** **boolean** hasFatalError() \{  
205.         **return** hasFatalError;  
206.     \}  
207.   
208.     /\*\* 
209.      \* The class always backs off, this exists only so that we can test without 
210.      \* taking a really long time. 
211.      \*  
212.      \* @param backoff 
213.      \*            - whether the source should backoff if the channel is full 
214.      \*/  
215.     @VisibleForTesting  
216.     **protected** **void** setBackOff(**boolean** backoff) \{  
217.         **this**.backoff = backoff;  
218.     \}  
219.   
220.     @VisibleForTesting  
221.     **protected** **boolean** hitChannelException() \{  
222.         **return** hitChannelException;  
223.     \}  
224.   
225.     @VisibleForTesting  
226.     **protected** SourceCounter getSourceCounter() \{  
227.         **return** sourceCounter;  
228.     \}  
229.   
230.     **private** **class** SpoolDirectoryRunnable **implements** Runnable \{  
231.         **private** ReliableSpoolingFileEventExtReader reader;  
232.         **private** SourceCounter sourceCounter;  
233.   
234.         **public** SpoolDirectoryRunnable(  
235.                 ReliableSpoolingFileEventExtReader reader,  
236.                 SourceCounter sourceCounter) \{  
237.             **this**.reader = reader;  
238.             **this**.sourceCounter = sourceCounter;  
239.         \}  
240.   
241.         @Override  
242.         **public** **void** run() \{  
243.             **int** backoffInterval = 250;  
244.             **try** \{  
245.                 **while** (!Thread.interrupted()) \{  
246.                     List<Event> events = reader.readEvents(batchSize);  
247.                     **if** (events.isEmpty()) \{  
248.                         **break**;  
249.                     \}  
250.                     sourceCounter.addToEventReceivedCount(events.size());  
251.                     sourceCounter.incrementAppendBatchReceivedCount();  
252.   
253.                     **try** \{  
254.                         getChannelProcessor().processEventBatch(events);  
255.                         reader.commit();  
256.                     \} **catch** (ChannelException ex) \{  
257.                         logger.warn("The channel is full, and cannot write data now. The "  
258.                                 + "source will try again after "  
259.                                 + String.valueOf(backoffInterval)  
260.                                 + " milliseconds");  
261.                         hitChannelException = **true**;  
262.                         **if** (backoff) \{  
263.                             TimeUnit.MILLISECONDS.sleep(backoffInterval);  
264.                             backoffInterval = backoffInterval << 1;  
265.                             backoffInterval = backoffInterval >= maxBackoff ? maxBackoff  
266.                                     : backoffInterval;  
267.                         \}  
268.                         **continue**;  
269.                     \}  
270.                     backoffInterval = 250;  
271.                     sourceCounter.addToEventAcceptedCount(events.size());  
272.                     sourceCounter.incrementAppendBatchAcceptedCount();  
273.                 \}  
274.                 logger.info("Spooling Directory Source runner has shutdown.");  
275.             \} **catch** (Throwable t) \{  
276.                 logger.error(  
277.                         "FATAL: "  
278.                                 + SpoolDirectoryExtSource.**this**.toString()  
279.                                 + ": "  
280.                                 + "Uncaught exception in SpoolDirectorySource thread. "  
281.                                 + "Restart or reconfigure Flume to continue processing.",  
282.                         t);  
283.                 hasFatalError = **true**;  
284.                 Throwables.propagate(t);  
285.             \}  
286.         \}  
287.     \}  
288. \}

主要提供了如下几个设置参数：

tier1.sources.source1.type=com.besttone.flume.SpoolDirectoryExtSource   \#写类的全路径名  
tier1.sources.source1.spoolDir=/opt/logs   \#监控的目录  
tier1.sources.source1.splitFileName=true   \#是否分隔文件名，并把分割后的内容添加到header中，默认false  
tier1.sources.source1.splitBy=\\\\.                   \#以什么符号分隔，默认是"."分隔  
tier1.sources.source1.splitBaseNameHeader=fileNameSplit  \#分割后写入header的key的前缀，比如a.log.2014-07-31,按“."分隔，

则header中有fileNameSplit0=a,fileNameSplit1=log,fileNameSplit2=2014-07-31

（其中还有扩展一个通过正则表达式抽取文件名的功能也在里面，我们这里不用到，就不介绍了）

扩展了这个source之后，前面的那个需求就很容易实现了，只需要：

tier1.sinks.sink1.hdfs.path=hdfs://master68:8020/flume/events/%\{fileNameSplit0\}/%\{fileNameSplit2\}

a.log.2014-07-31这个文件的内容就会保存到hdfs://master68:8020/flume/events/a/2014-07-31目录下面去了。

接下来我们说说如何部署这个我们扩展的自定义的spooling directory source（基于CM的设置）。

首先，我们把上面三个类打成JAR包：SpoolDirectoryExtSource.jar  
CM的flume插件目录为：/var/lib/flume-ng/plugins.d

然后再你需要使用这个source的agent上的/var/lib/flume-ng/plugins.d目录下面创建SpoolDirectoryExtSource目录以及子目录lib,libext,native,lib是放插件JAR的目录，libext是放插件的依赖JAR的目录，native放使用到的原生库（如果有用到的话）。

我们这里没有使用到其他的依赖，于是就把SpoolDirectoryExtSource.jar放到lib目录下就好了，最终的目录结构：

plugins.d/
    plugins.d/SpoolDirectoryExtSource/
    plugins.d/SpoolDirectoryExtSource/lib/SpoolDirectoryExtSource.jar
    plugins.d/SpoolDirectoryExtSource/libext/
    plugins.d/SpoolDirectoryExtSource/native/

重新启动flume agent,flume就会自动装载我们的插件，这样在flume.conf中就可以使用全路径类名配置type属性了。

最终flume.conf配置如下：

\[plain\]  [view plain][]  [copy][view plain]

1.  tier1.sources=source1  
2.  tier1.channels=channel1  
3.  tier1.sinks=sink1  
4.  tier1.sources.source1.type=com.besttone.flume.SpoolDirectoryExtSource  
5.  tier1.sources.source1.spoolDir=/opt/logs  
6.  tier1.sources.source1.splitFileName=true  
7.  tier1.sources.source1.splitBy=\\\\.  
8.  tier1.sources.source1.splitBaseNameHeader=fileNameSplit  
9.  tier1.sources.source1.channels=channel1  
10. tier1.sinks.sink1.type=hdfs  
11. tier1.sinks.sink1.channel=channel1  
12. tier1.sinks.sink1.hdfs.path=hdfs://master68:8020/flume/events/%\{fileNameSplit0\}/%\{fileNameSplit2\}  
13. tier1.sinks.sink1.hdfs.round=true  
14. tier1.sinks.sink1.hdfs.roundValue=10  
15. tier1.sinks.sink1.hdfs.roundUnit=minute  
16. tier1.sinks.sink1.hdfs.fileType=DataStream  
17. tier1.sinks.sink1.hdfs.writeFormat=Text  
18. tier1.sinks.sink1.hdfs.rollInterval=0  
19. tier1.sinks.sink1.hdfs.rollSize=10240  
20. tier1.sinks.sink1.hdfs.rollCount=0  
21. tier1.sinks.sink1.hdfs.idleTimeout=60  
22. tier1.channels.channel1.type=memory  
23. tier1.channels.channel1.capacity=10000  
24. tier1.channels.channel1.transactionCapacity=1000  
25. tier1.channels.channel1.keep-alive=30

附上一张用logger作为sink的查看日志文件的截图：![Center][]

问题导读：  
1.如何实现flume端自定一个sink，来按照我们的规则来保存日志？  
2.想从flume的配置文件中获取rootPath的值，该如何配置？  
  
![4.gif][]  
  
  
  
  
最近需要利用flume来做收集远端日志，所以学习一些flume最基本的用法。这里仅作记录。  
  
远端日志收集的整体思路是远端自定义实现log4j的appender把消息发送到flume端，flume端自定义实现一个sink来按照我们的规则保存日志。  
  
自定义Sink代码：

1.  public class LocalFileLogSink extends AbstractSink implements Configurable \{  
    
2.       private static final Logger logger = LoggerFactory  
    
3.                . getLogger(LocalFileLogSink .class );  
    
4.              private static final String PROP\_KEY\_ROOTPATH = "rootPath";  
    
5.        private String rootPath;  
    
6.       @Override  
    
7.       public void configure(Context context) \{  
    
8.            String rootPath = context.getString(PROP\_KEY\_ROOTPATH );  
    
9.            setRootPath(rootPath);  
    
10.      \}  
    
11.              
    
12.           @Override  
    
13.           public Status process() throws EventDeliveryException \{  
    
14.            logger .debug("Do process" );  
    
15.        ｝  
    
16. \}

复制代码

实现Configurable接口，即可在初始化时，通过configure方法从context中获取配置的参数的值。这里，我们是想从flume的配置文件中获取rootPath的值，也就是日志保存的根路径。在flume-conf.properties中配置如下：

1.  agent.sinks = loggerSink  
    
2.  agent.sinks.loggerSink.rootPath = ./logs

复制代码

loggerSink是自定义sink的名称，我们取值时的key，只需要loggerSink后面的部分即可，即这里的rootPath。  
  
实际业务逻辑的执行，是通过继承复写AbstractSink中的process方法实现。从基类的getChannel方法中获取信道，从中取出Event处理即可。

1.  Channel ch = getChannel();  
    
2.             Transaction txn = ch.getTransaction();  
    
3.           txn.begin();  
    
4.            try \{  
    
5.                logger .debug("Get event." );  
    
6.               Event event = ch.take();  
    
7.               txn.commit();  
    
8.               status = Status. READY ;  
    
9.               return status;  
    
10.                    ｝finally \{  
    
11.              Log. info( "trx close.");  
    
12.              txn.close();  
    
13.          \}

以下是我的自定义kafka sink插件的pom文件，编译成jar包丢到flume的lib下即可使用

\[html\]  [view plain][view plain 1]  [copy][view plain 1]

1. **<?****xml** version="1.0" encoding="UTF-8"**?>**  
2.   
3.   
4. **<****project** xmlns="http://maven.apache.org/POM/4.0.0"  
5.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
6.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"**>**  
7.   **<****modelVersion****>**4.0.0**</****modelVersion****>**  
8.     
9.   **<****groupId****>**flume-sinks**</****groupId****>**  
10.   **<****artifactId****>**cmcc-kafka-sink**</****artifactId****>**  
11.   **<****name****>**Flume Kafka Sink**</****name****>**  
12.   **<****version****>**1.0.0**</****version****>**  
13.   **<****build****>**  
14.     **<****plugins****>**  
15.       **<****plugin****>**  
16.         **<****groupId****>**org.apache.maven.plugins**</****groupId****>**  
17.         **<****artifactId****>**maven-jar-plugin**</****artifactId****>**  
18.       **</****plugin****>**  
19.     **</****plugins****>**  
20.   **</****build****>**  
21.   
22.   
23.   **<****dependencies****>**  
24.     **<****dependency****>**  
25.       **<****groupId****>**org.apache.flume**</****groupId****>**  
26.       **<****artifactId****>**flume-ng-sdk**</****artifactId****>**  
27.       **<****version****>**1.5.2**</****version****>**  
28.     **</****dependency****>**  
29.   
30.   
31.     **<****dependency****>**  
32.       **<****groupId****>**org.apache.flume**</****groupId****>**  
33.       **<****artifactId****>**flume-ng-core**</****artifactId****>**  
34.       **<****version****>**1.5.2**</****version****>**  
35.     **</****dependency****>**  
36.   
37.   
38.     **<****dependency****>**  
39.       **<****groupId****>**org.apache.flume**</****groupId****>**  
40.       **<****artifactId****>**flume-ng-configuration**</****artifactId****>**  
41.       **<****version****>**1.5.2**</****version****>**  
42.     **</****dependency****>**  
43.   
44.   
45.     **<****dependency****>**  
46.       **<****groupId****>**org.slf4j**</****groupId****>**  
47.       **<****artifactId****>**slf4j-api**</****artifactId****>**  
48.       **<****version****>**1.6.1**</****version****>**  
49.     **</****dependency****>**  
50.   
51.   
52.     **<****dependency****>**  
53.       **<****groupId****>**junit**</****groupId****>**  
54.       **<****artifactId****>**junit**</****artifactId****>**  
55.       **<****version****>**4.10**</****version****>**  
56.       **<****scope****>**test**</****scope****>**  
57.     **</****dependency****>**  
58.   
59.   
60.     **<****dependency****>**  
61.       **<****groupId****>**org.apache.kafka**</****groupId****>**  
62.       **<****artifactId****>**kafka\_2.10**</****artifactId****>**  
63.       **<****version****>**0.8.1.1**</****version****>**  
64.     **</****dependency****>**  
65.   **</****dependencies****>**  
66.   
67.   
68. **</****project****>**

这里取出了parent，也取出了rat plugin，这样就避免了编译时出现的常见错误 [https://issues.apache.org/jira/browse/FLUME-1372][https_issues.apache.org_jira_browse_FLUME-1372]

定义了几个变量

\[java\]  [view plain][view plain 1]  [copy][view plain 1]

1.  **public** **static** **final** String BATCH\_SIZE = "batchSize";  
2.      **public** **static** **final** **int** DEFAULT\_BATCH\_SIZE = 100;  
3.      **public** **static** **final** String PARTITION\_KEY\_NAME = "cmcc.partition.key";  
4.      **public** **static** **final** String ENCODING\_KEY\_NAME = "cmcc.encoding";  
5.      **public** **static** **final** String DEFAULT\_ENCODING = "UTF-8";  
6.      **public** **static** **final** String CUSTOME\_TOPIC\_KEY\_NAME = "cmcc.topic.name";  
7.      **public** **static** **final** String DEFAULT\_TOPIC\_NAME="CMCC";

自定义sink实现需要继承AbstractSink和实现接口Configurable，并重写部分方法，如下：

\[java\]  [view plain][view plain 1]  [copy][view plain 1]

1. **package** org.apache.flume.cmcc.kafka;  
2.   
3. **import** java.util.ArrayList;  
4. **import** java.util.List;  
5. **import** java.util.Properties;  
6.   
7. **import** kafka.javaapi.producer.Producer;  
8. **import** kafka.producer.KeyedMessage;  
9. **import** kafka.producer.ProducerConfig;  
10.   
11. **import** org.apache.commons.lang.StringUtils;  
12. **import** org.apache.flume.Channel;  
13. **import** org.apache.flume.Context;  
14. **import** org.apache.flume.Event;  
15. **import** org.apache.flume.Transaction;  
16. **import** org.apache.flume.conf.Configurable;  
17. **import** org.apache.flume.instrumentation.SinkCounter;  
18. **import** org.apache.flume.sink.AbstractSink;  
19. **import** org.slf4j.Logger;  
20. **import** org.slf4j.LoggerFactory;  
21.   
22. **import** com.google.common.base.Throwables;  
23. **import** com.google.common.collect.ImmutableMap;  
24.   
25. **public** **class** CmccKafkaSink **extends** AbstractSink **implements** Configurable \{  
26.   
27.     **private** **static** **final** Logger log = LoggerFactory  
28.             .getLogger(CmccKafkaSink.**class**);  
29.   
30.     **private** Properties parameters;  
31.     **private** Producer<String, String> producer;  
32.     // private Context context;  
33.     **private** **int** batchSize;// 一次事务的event数量，整体提交  
34.     **private** List<KeyedMessage<String, String>> messageList;  
35.     **private** SinkCounter sinkCounter;  
36.   
37.     @Override  
38.     **public** Status process() \{  
39.         // TODO Auto-generated method stub  
40.         Status result = Status.READY;  
41.         Channel channel = getChannel();  
42.         Transaction transaction = **null**;  
43.         Event event = **null**;  
44.         **try** \{  
45.             **long** processedEvent = 0;  
46.             transaction = channel.getTransaction();  
47.             transaction.begin();// 事务开始  
48.             messageList.clear();  
49.             **for** (; processedEvent < batchSize; processedEvent++) \{  
50.                 event = channel.take();// 从channel取出一个事件  
51.                 **if** (event == **null**) \{  
52.                     result = Status.BACKOFF;  
53.                     **break**;  
54.                 \}  
55.                 sinkCounter.incrementEventDrainAttemptCount();  
56.                 // Map<String, String> headers = event.getHeaders();  
57.                 String partitionKey = parameters  
58.                         .getProperty(Constants.PARTITION\_KEY\_NAME);  
59.                 String topic = StringUtils.defaultIfEmpty(parameters  
60.                         .getProperty(Constants.CUSTOME\_TOPIC\_KEY\_NAME),  
61.                         Constants.DEFAULT\_TOPIC\_NAME);  
62.                 String encoding = StringUtils.defaultIfEmpty(  
63.                         parameters.getProperty(Constants.ENCODING\_KEY\_NAME),  
64.                         Constants.DEFAULT\_ENCODING);  
65.                 **byte**\[\] eventBody = event.getBody();  
66.                 String eventData = **new** String(eventBody, encoding);  
67.                 KeyedMessage<String, String> data = **null**;  
68.                 **if** (StringUtils.isEmpty(partitionKey)) \{  
69.                     data = **new** KeyedMessage<String, String>(topic, eventData);  
70.                 \} **else** \{  
71.                     data = **new** KeyedMessage<String, String>(topic,  
72.                             partitionKey, eventData);  
73.                 \}  
74.                 messageList.add(data);  
75.                 log.debug("Add data \[" + eventData  
76.                         + "\] into messageList,position:" + processedEvent);  
77.             \}  
78.   
79.             **if** (processedEvent == 0) \{  
80.                 sinkCounter.incrementBatchEmptyCount();  
81.                 result = Status.BACKOFF;  
82.             \} **else** \{  
83.                 **if** (processedEvent < batchSize) \{  
84.                     sinkCounter.incrementBatchUnderflowCount();  
85.                 \} **else** \{  
86.                     sinkCounter.incrementBatchCompleteCount();  
87.                 \}  
88.                 sinkCounter.addToEventDrainAttemptCount(processedEvent);  
89.                 producer.send(messageList);  
90.                 log.debug("Send MessageList to Kafka: \[ message List size = "  
91.                         + messageList.size() + ",processedEvent number = "  
92.                         + processedEvent + "\] ");  
93.             \}  
94.             transaction.commit();// batchSize个事件处理完成，一次事务提交  
95.             sinkCounter.addToEventDrainSuccessCount(processedEvent);  
96.             result = Status.READY;  
97.         \} **catch** (Exception e) \{  
98.             String errorMsg = "Failed to publish events !";  
99.             log.error(errorMsg, e);  
100.             e.printStackTrace();  
101.             result = Status.BACKOFF;  
102.             **if** (transaction != **null**) \{  
103.                 **try** \{  
104.                     transaction.rollback();  
105.                     log.debug("transaction rollback success !");  
106.                 \} **catch** (Exception ex) \{  
107.                     log.error(errorMsg, ex);  
108.                     **throw** Throwables.propagate(ex);  
109.                 \}  
110.             \}  
111.             // throw new EventDeliveryException(errorMsg, e);  
112.         \} **finally** \{  
113.             **if** (transaction != **null**) \{  
114.                 transaction.close();  
115.             \}  
116.         \}  
117.         **return** result;  
118.     \}  
119.   
120.     @Override  
121.     **public** **synchronized** **void** start() \{  
122.         // TODO Auto-generated method stub  
123.         log.info("Starting \{\}...", **this**);  
124.         sinkCounter.start();  
125.         **super**.start();  
126.         ProducerConfig config = **new** ProducerConfig(**this**.parameters);  
127.         **this**.producer = **new** Producer<String, String>(config);  
128.         sinkCounter.incrementConnectionCreatedCount();  
129.     \}  
130.   
131.     @Override  
132.     **public** **synchronized** **void** stop() \{  
133.         // TODO Auto-generated method stub  
134.         log.debug("Cmcc Kafka sink \{\} stopping...", getName());  
135.         sinkCounter.stop();  
136.         producer.close();  
137.         sinkCounter.incrementConnectionClosedCount();  
138.     \}  
139.   
140.     @Override  
141.     **public** **void** configure(Context context) \{  
142.         // TODO Auto-generated method stub  
143.         ImmutableMap<String, String> props = context.getParameters();  
144.         batchSize = context.getInteger(Constants.BATCH\_SIZE,  
145.                 Constants.DEFAULT\_BATCH\_SIZE);  
146.         messageList = **new** ArrayList<KeyedMessage<String, String>>(batchSize);  
147.         parameters = **new** Properties();  
148.         **for** (String key : props.keySet()) \{  
149.             String value = props.get(key);  
150.             **this**.parameters.put(key, value);  
151.         \}  
152.         **if** (sinkCounter == **null**) \{  
153.             sinkCounter = **new** SinkCounter(getName());  
154.         \}  
155.     \}  
156.   
157. \}

以上sink同时支持了flume的内部监控

这里为了提高性能，添加了batchSize的概念，也就减少了事务提交的次数

当然当通道中已经没有event了，这时候就将之前处理的event都提交了

下面看配置

\[plain\]  [view plain][view plain 1]  [copy][view plain 1]

1.  a1.sinks.k1.type=org.apache.flume.cmcc.kafka.CmccKafkaSink  
2.  a1.sinks.k1.metadata.broker.list=192.168.11.174:9092  
3.  a1.sinks.k1.partition.key=0  
4.  a1.sinks.k1.partitioner.class=org.apache.flume.cmcc.kafka.CmccPartition  
5.  a1.sinks.k1.serializer.class=kafka.serializer.StringEncoder  
6.  a1.sinks.k1.request.required.acks=0  
7.  a1.sinks.k1.max.message.size=1000000  
8.  a1.sinks.k1.cmcc.encoding=UTF-8  
9.  a1.sinks.k1.cmcc.topic.name=CMCC  
10. a1.sinks.k1.producer.type=sync  
11. a1.sinks.k1.serializer.class=kafka.serializer.StringEncoder  
12. a1.sinks.k1.batchSize=100

这里我们看到，有些属性，我们在Constants中并没有定义，这是如何读取的呢，我们来看下kafka的源码就知道了：

\[java\]  [view plain][view plain 1]  [copy][view plain 1]

1.  **private** ProducerConfig(VerifiableProperties props)  
2.     \{  
3.         **this**.props = props;  
4.         **super**();  
5.         kafka.producer.async.AsyncProducerConfig.**class**.$init$(**this**);  
6.         SyncProducerConfigShared.**class**.$init$(**this**);  
7.         brokerList = props.getString("metadata.broker.list");  
8.         partitionerClass = props.getString("partitioner.class", "kafka.producer.DefaultPartitioner");  
9.         producerType = props.getString("producer.type", "sync");  
10.        String prop;  
11.        compressionCodec = liftedTree1$1(prop = props.getString("compression.codec", NoCompressionCodec$.MODULE$.name()));  
12.        Object \_tmp = **null**;  
13.        compressedTopics = Utils$.MODULE$.parseCsvList(props.getString("compressed.topics", **null**));  
14.        messageSendMaxRetries = props.getInt("message.send.max.retries", 3);  
15.        retryBackoffMs = props.getInt("retry.backoff.ms", 100);  
16.        topicMetadataRefreshIntervalMs = props.getInt("topic.metadata.refresh.interval.ms", 600000);  
17.        ProducerConfig$.MODULE$.validate(**this**);  
18.    \}

kafka的源码在实例化ProducerConfig的时候会读取配置文件中的kafka配置信息的。

## 四、自定义Channel开发 ##

官网说待定。

下面是美团网的自定义Channel 开发，下面是链接

http://tech.meituan.com/mt-log-system-optimization.html

……

Flume本身提供了MemoryChannel和FileChannel。MemoryChannel处理速度快，但缓存大小有限，且没有持久化；FileChannel则刚好相反。我们希望利用两者的优势，在Sink处理速度够快，Channel没有缓存过多日志的时候，就使用MemoryChannel，当Sink处理速度跟不上，又需要Channel能够缓存下应用端发送过来的日志时，就使用FileChannel，由此我们开发了DualChannel，能够智能的在两个Channel之间切换。

其具体的逻辑如下：

\[java\]  [view plain][view plain 2]  [copy][view plain 2]

1.  /\*\*\* 
2.   \* putToMemChannel indicate put event to memChannel or fileChannel 
3.   \* takeFromMemChannel indicate take event from memChannel or fileChannel 
4.   \* \*/  
5.  **private** AtomicBoolean putToMemChannel = **new** AtomicBoolean(**true**);  
6.  **private** AtomicBoolean takeFromMemChannel = **new** AtomicBoolean(**true**);  
7.    
8.  **void** doPut(Event event) \{  
9.          **if** (switchon && putToMemChannel.get()) \{  
10.               //往memChannel中写数据  
11.               memTransaction.put(event);  
12.   
13.               **if** ( memChannel.isFull() || fileChannel.getQueueSize() > 100) \{  
14.                 putToMemChannel.set(**false**);  
15.               \}  
16.         \} **else** \{  
17.               //往fileChannel中写数据  
18.               fileTransaction.put(event);  
19.         \}  
20.   \}  
21.   
22. Event doTake() \{  
23.     Event event = **null**;  
24.     **if** ( takeFromMemChannel.get() ) \{  
25.         //从memChannel中取数据  
26.         event = memTransaction.take();  
27.         **if** (event == **null**) \{  
28.             takeFromMemChannel.set(**false**);  
29.         \}   
30.     \} **else** \{  
31.         //从fileChannel中取数据  
32.         event = fileTransaction.take();  
33.         **if** (event == **null**) \{  
34.             takeFromMemChannel.set(**true**);  
35.   
36.             putToMemChannel.set(**true**);  
37.         \}   
38.     \}  
39.     **return** event;  
40. \}

这里要说明下，官网是建议使用file channel，虽然它的效率比较低，但是它能保证数据完整性，而memory channel效率高，但是只能对数据丢失和重复不太敏感的业务使用

还是针对学习八中的那个需求，我们现在换一种实现方式，采用拦截器来实现。

先回想一下，spooldir source可以将文件名作为header中的key:basename写入到event的header当中去。试想一下，如果有一个拦截器可以拦截这个event,然后抽取header中这个key的值，将其拆分成3段，每一段都放入到header中，这样就可以实现那个需求了。

遗憾的是，flume没有提供可以拦截header的拦截器。不过有一个抽取body内容的拦截器：RegexExtractorInterceptor，看起来也很强大，以下是一个官方文档的示例：

If the Flume event body contained 1:2:3.4foobar5 and the following configuration was used  
  
  
a1.sources.r1.interceptors.i1.regex = (\\\\d):(\\\\d):(\\\\d)  
a1.sources.r1.interceptors.i1.serializers = s1 s2 s3  
a1.sources.r1.interceptors.i1.serializers.s1.name = one  
a1.sources.r1.interceptors.i1.serializers.s2.name = two  
a1.sources.r1.interceptors.i1.serializers.s3.name = three  
The extracted event will contain the same body but the following headers will have been added one=>1, two=>2, three=>3

大概意思就是，通过这样的配置，event body中如果有1:2:3.4foobar5 这样的内容，这会通过正则的规则抽取具体部分的内容，然后设置到header当中去。

于是决定打这个拦截器的主义，觉得只要把代码稍微改改，从拦截body改为拦截header中的具体key，就OK了。翻开源码，哎呀，很工整，改起来没难度，以下是我新增的一个拦截器：RegexExtractorExtInterceptor：

\[java\]  [view plain][view plain 3]  [copy][view plain 3]

1. **package** com.besttone.flume;  
2.   
3. **import** java.util.List;  
4. **import** java.util.Map;  
5. **import** java.util.regex.Matcher;  
6. **import** java.util.regex.Pattern;  
7.   
8. **import** org.apache.commons.lang.StringUtils;  
9. **import** org.apache.flume.Context;  
10. **import** org.apache.flume.Event;  
11. **import** org.apache.flume.interceptor.Interceptor;  
12. **import** org.apache.flume.interceptor.RegexExtractorInterceptorPassThroughSerializer;  
13. **import** org.apache.flume.interceptor.RegexExtractorInterceptorSerializer;  
14. **import** org.slf4j.Logger;  
15. **import** org.slf4j.LoggerFactory;  
16.   
17. **import** com.google.common.base.Charsets;  
18. **import** com.google.common.base.Preconditions;  
19. **import** com.google.common.base.Throwables;  
20. **import** com.google.common.collect.Lists;  
21.   
22. /\*\* 
23.  \* Interceptor that extracts matches using a specified regular expression and 
24.  \* appends the matches to the event headers using the specified serializers 
25.  \* Note that all regular expression matching occurs through Java's built in 
26.  \* java.util.regex package. Properties: 
27.  \*  
28.  \* regex: The regex to use 
29.  \*  
30.  \* serializers: Specifies the group the serializer will be applied to, and the 
31.  \* name of the header that will be added. If no serializer is specified for a 
32.  \* group the default \{@link RegexExtractorInterceptorPassThroughSerializer\} will 
33.  \* be used 
34.  \*  
35.  \* Sample config: 
36.  \*  
37.  \* agent.sources.r1.channels = c1 
38.  \*  
39.  \* agent.sources.r1.type = SEQ 
40.  \*  
41.  \* agent.sources.r1.interceptors = i1 
42.  \*  
43.  \* agent.sources.r1.interceptors.i1.type = REGEX\_EXTRACTOR 
44.  \*  
45.  \* agent.sources.r1.interceptors.i1.regex = (WARNING)|(ERROR)|(FATAL) 
46.  \*  
47.  \* agent.sources.r1.interceptors.i1.serializers = s1 s2 
48.  \* agent.sources.r1.interceptors.i1.serializers.s1.type = 
49.  \* com.blah.SomeSerializer agent.sources.r1.interceptors.i1.serializers.s1.name 
50.  \* = warning agent.sources.r1.interceptors.i1.serializers.s2.type = 
51.  \* org.apache.flume.interceptor.RegexExtractorInterceptorTimestampSerializer 
52.  \* agent.sources.r1.interceptors.i1.serializers.s2.name = error 
53.  \* agent.sources.r1.interceptors.i1.serializers.s2.dateFormat = yyyy-MM-dd 
54.  \* </code> 
55.  \*  
56.  \*  
57.  \* <pre> 
58.  \* Example 1: 
59.  \*  
60.  \* EventBody: 1:2:3.4foobar5 Configuration: 
61.  \* agent.sources.r1.interceptors.i1.regex = (\\\\d):(\\\\d):(\\\\d) 
62.  \*  
63.  \* agent.sources.r1.interceptors.i1.serializers = s1 s2 s3 
64.  \* agent.sources.r1.interceptors.i1.serializers.s1.name = one 
65.  \* agent.sources.r1.interceptors.i1.serializers.s2.name = two 
66.  \* agent.sources.r1.interceptors.i1.serializers.s3.name = three 
67.  \*  
68.  \* results in an event with the the following 
69.  \*  
70.  \* body: 1:2:3.4foobar5 headers: one=>1, two=>2, three=3 
71.  \*  
72.  \* Example 2: 
73.  \*  
74.  \* EventBody: 1:2:3.4foobar5 
75.  \*  
76.  \* Configuration: agent.sources.r1.interceptors.i1.regex = (\\\\d):(\\\\d):(\\\\d) 
77.  \*  
78.  \* agent.sources.r1.interceptors.i1.serializers = s1 s2 
79.  \* agent.sources.r1.interceptors.i1.serializers.s1.name = one 
80.  \* agent.sources.r1.interceptors.i1.serializers.s2.name = two 
81.  \*  
82.  \*  
83.  \* results in an event with the the following 
84.  \*  
85.  \* body: 1:2:3.4foobar5 headers: one=>1, two=>2 
86.  \* </pre> 
87.  \*/  
88. **public** **class** RegexExtractorExtInterceptor **implements** Interceptor \{  
89.   
90.     **static** **final** String REGEX = "regex";  
91.     **static** **final** String SERIALIZERS = "serializers";  
92.   
93.     // 增加代码开始  
94.   
95.     **static** **final** String EXTRACTOR\_HEADER = "extractorHeader";  
96.     **static** **final** **boolean** DEFAULT\_EXTRACTOR\_HEADER = **false**;  
97.     **static** **final** String EXTRACTOR\_HEADER\_KEY = "extractorHeaderKey";  
98.   
99.     // 增加代码结束  
100.   
101.     **private** **static** **final** Logger logger = LoggerFactory  
102.             .getLogger(RegexExtractorExtInterceptor.**class**);  
103.   
104.     **private** **final** Pattern regex;  
105.     **private** **final** List<NameAndSerializer> serializers;  
106.   
107.     // 增加代码开始  
108.   
109.     **private** **final** **boolean** extractorHeader;  
110.     **private** **final** String extractorHeaderKey;  
111.   
112.     // 增加代码结束  
113.   
114.     **private** RegexExtractorExtInterceptor(Pattern regex,  
115.             List<NameAndSerializer> serializers, **boolean** extractorHeader,  
116.             String extractorHeaderKey) \{  
117.         **this**.regex = regex;  
118.         **this**.serializers = serializers;  
119.         **this**.extractorHeader = extractorHeader;  
120.         **this**.extractorHeaderKey = extractorHeaderKey;  
121.     \}  
122.   
123.     @Override  
124.     **public** **void** initialize() \{  
125.         // NO-OP...  
126.     \}  
127.   
128.     @Override  
129.     **public** **void** close() \{  
130.         // NO-OP...  
131.     \}  
132.   
133.     @Override  
134.     **public** Event intercept(Event event) \{  
135.         String tmpStr;  
136.         **if**(extractorHeader)  
137.         \{  
138.             tmpStr = event.getHeaders().get(extractorHeaderKey);  
139.         \}  
140.         **else**  
141.         \{  
142.             tmpStr=**new** String(event.getBody(),  
143.                     Charsets.UTF\_8);  
144.         \}  
145.           
146.         Matcher matcher = regex.matcher(tmpStr);  
147.         Map<String, String> headers = event.getHeaders();  
148.         **if** (matcher.find()) \{  
149.             **for** (**int** group = 0, count = matcher.groupCount(); group < count; group++) \{  
150.                 **int** groupIndex = group + 1;  
151.                 **if** (groupIndex > serializers.size()) \{  
152.                     **if** (logger.isDebugEnabled()) \{  
153.                         logger.debug(  
154.                                 "Skipping group \{\} to \{\} due to missing serializer",  
155.                                 group, count);  
156.                     \}  
157.                     **break**;  
158.                 \}  
159.                 NameAndSerializer serializer = serializers.get(group);  
160.                 **if** (logger.isDebugEnabled()) \{  
161.                     logger.debug("Serializing \{\} using \{\}",  
162.                             serializer.headerName, serializer.serializer);  
163.                 \}  
164.                 headers.put(serializer.headerName, serializer.serializer  
165.                         .serialize(matcher.group(groupIndex)));  
166.             \}  
167.         \}  
168.         **return** event;  
169.     \}  
170.   
171.     @Override  
172.     **public** List<Event> intercept(List<Event> events) \{  
173.         List<Event> intercepted = Lists.newArrayListWithCapacity(events.size());  
174.         **for** (Event event : events) \{  
175.             Event interceptedEvent = intercept(event);  
176.             **if** (interceptedEvent != **null**) \{  
177.                 intercepted.add(interceptedEvent);  
178.             \}  
179.         \}  
180.         **return** intercepted;  
181.     \}  
182.   
183.     **public** **static** **class** Builder **implements** Interceptor.Builder \{  
184.   
185.         **private** Pattern regex;  
186.         **private** List<NameAndSerializer> serializerList;  
187.   
188.         // 增加代码开始  
189.   
190.         **private** **boolean** extractorHeader;  
191.         **private** String extractorHeaderKey;  
192.   
193.         // 增加代码结束  
194.   
195.         **private** **final** RegexExtractorInterceptorSerializer defaultSerializer = **new** RegexExtractorInterceptorPassThroughSerializer();  
196.   
197.         @Override  
198.         **public** **void** configure(Context context) \{  
199.             String regexString = context.getString(REGEX);  
200.             Preconditions.checkArgument(!StringUtils.isEmpty(regexString),  
201.                     "Must supply a valid regex string");  
202.   
203.             regex = Pattern.compile(regexString);  
204.             regex.pattern();  
205.             regex.matcher("").groupCount();  
206.             configureSerializers(context);  
207.   
208.             // 增加代码开始  
209.             extractorHeader = context.getBoolean(EXTRACTOR\_HEADER,  
210.                     DEFAULT\_EXTRACTOR\_HEADER);  
211.   
212.             **if** (extractorHeader) \{  
213.                 extractorHeaderKey = context.getString(EXTRACTOR\_HEADER\_KEY);  
214.                 Preconditions.checkArgument(  
215.                         !StringUtils.isEmpty(extractorHeaderKey),  
216.                         "必须指定要抽取内容的header key");  
217.             \}  
218.             // 增加代码结束  
219.         \}  
220.   
221.         **private** **void** configureSerializers(Context context) \{  
222.             String serializerListStr = context.getString(SERIALIZERS);  
223.             Preconditions.checkArgument(  
224.                     !StringUtils.isEmpty(serializerListStr),  
225.                     "Must supply at least one name and serializer");  
226.   
227.             String\[\] serializerNames = serializerListStr.split("\\\\s+");  
228.   
229.             Context serializerContexts = **new** Context(  
230.                     context.getSubProperties(SERIALIZERS + "."));  
231.   
232.             serializerList = Lists  
233.                     .newArrayListWithCapacity(serializerNames.length);  
234.             **for** (String serializerName : serializerNames) \{  
235.                 Context serializerContext = **new** Context(  
236.                         serializerContexts.getSubProperties(serializerName  
237.                                 + "."));  
238.                 String type = serializerContext.getString("type", "DEFAULT");  
239.                 String name = serializerContext.getString("name");  
240.                 Preconditions.checkArgument(!StringUtils.isEmpty(name),  
241.                         "Supplied name cannot be empty.");  
242.   
243.                 **if** ("DEFAULT".equals(type)) \{  
244.                     serializerList.add(**new** NameAndSerializer(name,  
245.                             defaultSerializer));  
246.                 \} **else** \{  
247.                     serializerList.add(**new** NameAndSerializer(name,  
248.                             getCustomSerializer(type, serializerContext)));  
249.                 \}  
250.             \}  
251.         \}  
252.   
253.         **private** RegexExtractorInterceptorSerializer getCustomSerializer(  
254.                 String clazzName, Context context) \{  
255.             **try** \{  
256.                 RegexExtractorInterceptorSerializer serializer = (RegexExtractorInterceptorSerializer) Class  
257.                         .forName(clazzName).newInstance();  
258.                 serializer.configure(context);  
259.                 **return** serializer;  
260.             \} **catch** (Exception e) \{  
261.                 logger.error("Could not instantiate event serializer.", e);  
262.                 Throwables.propagate(e);  
263.             \}  
264.             **return** defaultSerializer;  
265.         \}  
266.   
267.         @Override  
268.         **public** Interceptor build() \{  
269.             Preconditions.checkArgument(regex != **null**,  
270.                     "Regex pattern was misconfigured");  
271.             Preconditions.checkArgument(serializerList.size() > 0,  
272.                     "Must supply a valid group match id list");  
273.             **return** **new** RegexExtractorExtInterceptor(regex, serializerList,  
274.                     extractorHeader, extractorHeaderKey);  
275.         \}  
276.     \}  
277.   
278.     **static** **class** NameAndSerializer \{  
279.         **private** **final** String headerName;  
280.         **private** **final** RegexExtractorInterceptorSerializer serializer;  
281.   
282.         **public** NameAndSerializer(String headerName,  
283.                 RegexExtractorInterceptorSerializer serializer) \{  
284.             **this**.headerName = headerName;  
285.             **this**.serializer = serializer;  
286.         \}  
287.     \}  
288. \}

简单说明一下改动的内容：

增加了两个配置参数：

extractorHeader   是否抽取的是header部分，默认为false,即和原始的拦截器功能一致，抽取的是event body的内容

extractorHeaderKey 抽取的header的指定的key的内容，当extractorHeader为true时，必须指定该参数。

按照第八讲的方法，我们将该类打成jar包，作为flume的插件放到了/var/lib/flume-ng/plugins.d/RegexExtractorExtInterceptor/lib目录下，重新启动flume，将该拦截器加载到classpath中。

最终的flume.conf如下：

\[plain\]  [view plain][view plain 3]  [copy][view plain 3]

1.  tier1.sources=source1  
2.  tier1.channels=channel1  
3.  tier1.sinks=sink1  
4.  tier1.sources.source1.type=spooldir  
5.  tier1.sources.source1.spoolDir=/opt/logs  
6.  tier1.sources.source1.fileHeader=true  
7.  tier1.sources.source1.basenameHeader=true  
8.  tier1.sources.source1.interceptors=i1  
9.  tier1.sources.source1.interceptors.i1.type=com.besttone.flume.RegexExtractorExtInterceptor$Builder  
10. tier1.sources.source1.interceptors.i1.regex=(.\*)\\\\.(.\*)\\\\.(.\*)  
11. tier1.sources.source1.interceptors.i1.extractorHeader=true  
12. tier1.sources.source1.interceptors.i1.extractorHeaderKey=basename  
13. tier1.sources.source1.interceptors.i1.serializers=s1 s2 s3  
14. tier1.sources.source1.interceptors.i1.serializers.s1.name=one  
15. tier1.sources.source1.interceptors.i1.serializers.s2.name=two  
16. tier1.sources.source1.interceptors.i1.serializers.s3.name=three  
17. tier1.sources.source1.channels=channel1  
18. tier1.sinks.sink1.type=hdfs  
19. tier1.sinks.sink1.channel=channel1  
20. tier1.sinks.sink1.hdfs.path=hdfs://master68:8020/flume/events/%\{one\}/%\{three\}  
21. tier1.sinks.sink1.hdfs.round=true  
22. tier1.sinks.sink1.hdfs.roundValue=10  
23. tier1.sinks.sink1.hdfs.roundUnit=minute  
24. tier1.sinks.sink1.hdfs.fileType=DataStream  
25. tier1.sinks.sink1.hdfs.writeFormat=Text  
26. tier1.sinks.sink1.hdfs.rollInterval=0  
27. tier1.sinks.sink1.hdfs.rollSize=10240  
28. tier1.sinks.sink1.hdfs.rollCount=0  
29. tier1.sinks.sink1.hdfs.idleTimeout=60  
30. tier1.channels.channel1.type=memory  
31. tier1.channels.channel1.capacity=10000  
32. tier1.channels.channel1.transactionCapacity=1000  
33. tier1.channels.channel1.keep-alive=30

我把source type改回了内置的spooldir，而不是上一讲自定义的source,然后添加了一个拦截器i1,type是自定义的拦截器：com.besttone.flume.RegexExtractorExtInterceptor$Builder,正则表达式按“.”分隔抽取三部分，分别放到header中的key:one,two,three当中去，即a.log.2014-07-31,通过拦截器后，在header当中就会增加三个key: one=a,two=log,three=2014-07-31。这时候我们在tier1.sinks.sink1.hdfs.path=hdfs://master68:8020/flume/events/%\{one\}/%\{three\}。

就实现了和前面第八讲一模一样的需求。

也可以看到，自定义拦截器的改动成本非常小，比自定义source小多了，我们这就增加了一个类，就实现了该功能。

[view plain]: http://blog.csdn.net/xiao_jun_0820/article/details/38312091#
[Center]: /images/20220604/f66908e890c34fefb8b59dd0331b80a0.png
[4.gif]: /images/20220604/5658b010232c4848ba2f60d344da9adb.png
[view plain 1]: http://blog.csdn.net/simonchi/article/details/42489885#
[https_issues.apache.org_jira_browse_FLUME-1372]: https://issues.apache.org/jira/browse/FLUME-1372
[view plain 2]: http://blog.csdn.net/looklook5/article/details/40655935#
[view plain 3]: http://blog.csdn.net/xiao_jun_0820/article/details/38333171#