{"id":1306,"date":"2022-03-02T23:13:38","date_gmt":"2022-03-02T15:13:38","guid":{"rendered":"http:\/\/www.eait.co\/?p=1306"},"modified":"2022-03-03T10:36:15","modified_gmt":"2022-03-03T02:36:15","slug":"flume%e5%90%84%e7%a7%8d%e6%8b%a6%e6%88%aa%e5%99%a8%ef%bc%88%e4%ba%8c%ef%bc%89-%e6%ad%a3%e5%88%99%e5%92%8c%e8%87%aa%e5%ae%9a%e4%b9%89%e6%8b%a6%e6%88%aa%e5%99%a8","status":"publish","type":"post","link":"https:\/\/notes.coremix.net\/?p=1306","title":{"rendered":"Flume\u5404\u79cd\u62e6\u622a\u5668\uff08\u4e8c\uff09\u2014\u2014\u6b63\u5219\u548c\u81ea\u5b9a\u4e49\u62e6\u622a\u5668"},"content":{"rendered":"

\u67e5\u8be2\u66ff\u6362\u62e6\u622a\u5668<\/span><\/h4>\n

 <\/p>\n

search.conf<\/p>\n

\r\n#1 agent\r\na1.sources = r1\r\na1.sinks = k1\r\na1.channels = c1\r\n\r\n#2 source\r\na1.sources.r1.type = exec\r\na1.sources.r1.channels = c1\r\na1.sources.r1.command = tail -F \/opt\/Shalltest\r\na1.sources.r1.interceptors = i1\r\na1.sources.r1.interceptors.i1.type = search_replace\r\n\r\n#\u9047\u5230\u6570\u5b57\u6539\u6210shalltest\uff0cA123\u4f1a\u66ff\u6362\u4e3aAshalltest\r\na1.sources.r1.interceptors.i1.searchPattern = [0-9]+\r\na1.sources.r1.interceptors.i1.replaceString = shalltest\r\na1.sources.r1.interceptors.i1.charset = UTF-8\r\n\r\n#3 sink\r\na1.sinks.k1.type = logger\r\n\r\n#4 Chanel\r\na1.channels.c1.type = memory\r\na1.channels.c1.capacity = 1000\r\na1.channels.c1.transactionCapacity = 100\r\n\r\n#5 bind\r\na1.sources.r1.channels = c1\r\na1.sinks.k1.channel = c1\r\n<\/pre>\n
\r\nbin\/flume-ng agent -c conf\/ -f jobconf\/search.conf -n a1 -Dflume.root.logger=INFO,console\r\n<\/pre>\n
\u6b63\u5219\u8fc7\u6ee4\u62e6\u622a\u5668<\/span><\/h4>\n
filter.conf<\/p>\n
\r\n#1 agent\r\na1.sources = r1\r\na1.sinks = k1\r\na1.channels = c1\r\n\r\n#2 source\r\na1.sources.r1.type = exec\r\na1.sources.r1.channels = c1\r\na1.sources.r1.command = tail -F \/opt\/Shalltest\r\na1.sources.r1.interceptors = i1\r\na1.sources.r1.interceptors.i1.type = regex_filter\r\na1.sources.r1.interceptors.i1.regex = ^A.*\r\n#\u5982\u679cexcludeEvents\u8bbe\u4e3afalse,\u8868\u793a\u8fc7\u6ee4\u6389\u4e0d\u662f\u4ee5A\u5f00\u5934\u7684events\u3002\u5982\u679cexcludeEvents\u8bbe\u4e3atrue\uff0c\u5219\u8868\u793a\u8fc7\u6ee4\u6389\u4ee5A\u5f00\u5934\u7684events\u3002\r\na1.sources.r1.interceptors.i1.excludeEvents = true\r\n\r\na1.sinks.k1.type = logger\r\n\r\na1.channels.c1.type = memory\r\na1.channels.c1.capacity = 1000\r\na1.channels.c1.transactionCapacity = 100\r\n\r\na1.sources.r1.channels = c1\r\na1.sinks.k1.channel = c1\r\n\r\n<\/pre>\n
\r\nbin\/flume-ng agent -c conf\/ -f jobconf\/filter.conf -n a1 -Dflume.root.logger=INFO,console\r\n<\/pre>\n
\u6b63\u5219\u62bd\u53d6\u62e6\u622a\u5668<\/span><\/h4>\n
extractor.conf<\/p>\n
\r\n\r\n#1 agent\r\na1.sources = r1\r\na1.sinks = k1\r\na1.channels = c1\r\n\r\n#2 source\r\na1.sources.r1.type = exec\r\na1.sources.r1.channels = c1\r\na1.sources.r1.command = tail -F \/opt\/Shalltest\r\na1.sources.r1.interceptors = i1\r\na1.sources.r1.interceptors.i1.type = regex_extractor\r\na1.sources.r1.interceptors.i1.regex = hostname is (.*?) ip is (.*)\r\na1.sources.r1.interceptors.i1.serializers = s1 s2\r\na1.sources.r1.interceptors.i1.serializers.s1.name = hostname\r\na1.sources.r1.interceptors.i1.serializers.s2.name = ip\r\n\r\na1.sinks.k1.type = logger\r\n\r\na1.channels.c1.type = memory\r\na1.channels.c1.capacity = 1000\r\na1.channels.c1.transactionCapacity = 100\r\n\r\na1.sources.r1.channels = c1\r\na1.sinks.k1.channel = c1\r\n<\/pre>\n
\r\nbin\/flume-ng agent -c conf\/ -f jobconf\/extractor.conf -n a1 -Dflume.root.logger=INFO,console\r\n<\/pre>\n
\u81ea\u5b9a\u4e49\u62e6\u622a\u5668<\/span><\/h4>\n
maven\u914d\u7f6e\u6587\u4ef6<\/p>\n
\r\n<dependencies>\r\n        <!-- flume\u6838\u5fc3\u4f9d\u8d56 -->\r\n        <dependency>\r\n            <groupId>org.apache.flume<\/groupId>\r\n            <artifactId>flume-ng-core<\/artifactId>\r\n            <version>1.8.0<\/version>\r\n        <\/dependency>\r\n    <\/dependencies>\r\n    <build>\r\n    <plugins>\r\n    <!-- \u6253\u5305\u63d2\u4ef6 -->\r\n    <plugin>\r\n        <groupId>org.apache.maven.plugins<\/groupId>\r\n        <artifactId>maven-jar-plugin<\/artifactId>\r\n        <version>2.4<\/version>\r\n        <configuration>\r\n            <archive>\r\n                <manifest>\r\n                    <addClasspath>true<\/addClasspath>\r\n                    <classpathPrefix>lib\/<\/classpathPrefix>\r\n                    <mainClass><\/mainClass>\r\n                <\/manifest>\r\n            <\/archive>\r\n        <\/configuration>\r\n    <\/plugin>\r\n    <!-- \u7f16\u8bd1\u63d2\u4ef6 -->\r\n    <plugin>\r\n    <groupId>org.apache.maven.plugins<\/groupId>\r\n        <artifactId>maven-compiler-plugin<\/artifactId>\r\n        <configuration>\r\n            <source>1.8<\/source>\r\n            <target>1.8<\/target>\r\n            <encoding>utf-8<\/encoding>\r\n        <\/configuration>\r\n    <\/plugin>\r\n    <\/plugins>\r\n    <\/build>\r\n<\/pre>\n
Java\u4ee3\u7801\uff1a<\/strong><\/p>\n
\r\npackage ToUpCase;\r\n\r\nimport org.apache.flume.Context;\r\nimport org.apache.flume.Event;\r\nimport org.apache.flume.interceptor.Interceptor;\r\n\r\nimport java.util.ArrayList;\r\nimport java.util.List;\r\n\r\npublic class MyInterceptor implements Interceptor {\r\n    @Override\r\n    public void initialize() {\r\n\r\n    }\r\n\r\n    @Override\r\n    public Event intercept(Event event) {\r\n        byte[] body = event.getBody();\r\n        event.setBody(new String(body).toUpperCase().getBytes());\r\n        return event;\r\n    }\r\n\r\n    @Override\r\n    public List<Event> intercept(List<Event> list) {\r\n        List<Event> events = new ArrayList<>();\r\n        for (Event event : list) {\r\n            events.add(intercept(event));\r\n        }\r\n        return events;\r\n    }\r\n\r\n\r\n\r\n    @Override\r\n    public void close() {\r\n\r\n    }\r\n\r\n    public static class Builder implements Interceptor.Builder{\r\n\r\n        @Override\r\n        public Interceptor build() {\r\n            return new MyInterceptor();\r\n        }\r\n\r\n        @Override\r\n        public void configure(Context context) {\r\n\r\n        }\r\n    }\r\n\r\n}\r\n\r\n<\/pre>\n
ToUpCase.conf<\/p>\n
\r\n\r\n#1.agent \r\na1.sources = r1\r\na1.sinks =k1\r\na1.channels = c1\r\n\r\n\r\n# Describe\/configure the source\r\na1.sources.r1.type = exec\r\na1.sources.r1.command = tail -F \/opt\/Shalltest\r\na1.sources.r1.interceptors = i1\r\n#\u5168\u7c7b\u540d$Builder\r\na1.sources.r1.interceptors.i1.type = ToUpCase.MyInterceptor$Builder\r\n\r\n# Describe the sink\r\na1.sinks.k1.type = hdfs\r\na1.sinks.k1.hdfs.path = \/ToUpCase1\r\na1.sinks.k1.hdfs.filePrefix = events-\r\na1.sinks.k1.hdfs.round = true\r\na1.sinks.k1.hdfs.roundValue = 10\r\na1.sinks.k1.hdfs.roundUnit = minute\r\na1.sinks.k1.hdfs.rollInterval = 3\r\na1.sinks.k1.hdfs.rollSize = 20\r\na1.sinks.k1.hdfs.rollCount = 5\r\na1.sinks.k1.hdfs.batchSize = 1\r\na1.sinks.k1.hdfs.useLocalTimeStamp = true\r\n#\u751f\u6210\u7684\u6587\u4ef6\u7c7b\u578b\uff0c\u9ed8\u8ba4\u662f Sequencefile\uff0c\u53ef\u7528 DataStream\uff0c\u5219\u4e3a\u666e\u901a\u6587\u672c\r\na1.sinks.k1.hdfs.fileType = DataStream\r\n\r\n# Use a channel which buffers events in memory\r\na1.channels.c1.type = memory\r\na1.channels.c1.capacity = 1000\r\na1.channels.c1.transactionCapacity = 100\r\n\r\n# Bind the source and sink to the channel\r\na1.sources.r1.channels = c1\r\na1.sinks.k1.channel = c1\r\n<\/pre>\n
 <\/p>\n
\u542f\u52a8\uff0c\u5373\u53ef\u770b\u5230HDFS\u91cc\u9762\u6709\u4e86\u6587\u4ef6\u5939\uff0c\u5e76\u5df2\u7ecf\u8f6c\u5316\u6210\u5927\u5199<\/p>\n
\r\nbin\/flume-ng agent -c conf\/ -n a1 -f jar\/ToUpCase.conf -C jar\/vip.isok-1.0-SNAPSHOT.jar -Dflume.root.logger=DEBUG,console\r\n<\/pre>\n
 <\/p>\n
 <\/p>\n
 <\/p>\n
 <\/p>\n
 <\/p>\n
 <\/p>\n
 <\/p>\n","protected":false},"excerpt":{"rendered":"
\u67e5\u8be2\u66ff\u6362\u62e6\u622a\u5668   search.conf #1 agent a1.sources = r1 a1. […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-1306","post","type-post","status-publish","format-standard","hentry","category-uncategorized"],"blocksy_meta":[],"_links":{"self":[{"href":"https:\/\/notes.coremix.net\/index.php?rest_route=\/wp\/v2\/posts\/1306","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/notes.coremix.net\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/notes.coremix.net\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/notes.coremix.net\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/notes.coremix.net\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=1306"}],"version-history":[{"count":7,"href":"https:\/\/notes.coremix.net\/index.php?rest_route=\/wp\/v2\/posts\/1306\/revisions"}],"predecessor-version":[{"id":1313,"href":"https:\/\/notes.coremix.net\/index.php?rest_route=\/wp\/v2\/posts\/1306\/revisions\/1313"}],"wp:attachment":[{"href":"https:\/\/notes.coremix.net\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=1306"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/notes.coremix.net\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=1306"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/notes.coremix.net\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=1306"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}