感謝各位的幫助,不管有沒有解決問題....
使用版本:1.7.0
大體業務:一個Agent(a1)負責監聽nc,并將接收到的訊息送往其它兩個Agent(a2和a3),這兩個Agent(監聽的)使用的是虛擬機
預期結果:a1把監聽到的資料發到hadoop102(a2負責接收字母)/hadoop103(a3負責接收數字)
當前結果:兩個負責接收的沒有顯示各自的,反而有什么就發什么
flume組態檔詳細
[a1] -> 監聽并發送監聽聽到的訊息
# Name
a1.sources = s1
a1.channels = c1 c2
a1.sinks = k1 k2
# Sources
a1.sources.s1.type = netcat
a1.sources.s1.bind = localhost
a1.sources.s1.port = 44444
# Channels
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.channels.c2.type = memory
a1.channels.c2.capacity = 1000
a1.channels.c2.transactionCapacity = 100
# Channel Interceptors
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = dell.flume.test.test1$Builder
# Channel Selector
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = type
a1.sources.r1.selector.mapping.letter = c1
a1.sources.r1.selector.mapping.number = c2
# Sinks
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop102
a1.sinks.k1.port = 4142
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = hadoop103
a1.sinks.k2.port = 4141
# Bind
a1.sources.s1.channels = c1 c2
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2
[a2] -> 接收字母
# Name
a2.sources = s1
a2.channels = c1
a2.sinks = k1
# Sources
a2.sources.s1.type = avro
a2.sources.s1.bind = hadoop102
a2.sources.s1.port = 4142
# Channels
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100
# Sinks
a2.sinks.k1.type = logger
# Bind
a2.sources.s1.channels = c1
a2.sinks.k1.channel = c1
[a3] -> 接收數字
# Name
a3.sources = s1
a3.channels = c1
a3.sinks = k1
# Sources
a3.sources.s1.type = avro
a3.sources.s1.bind = hadoop103
a3.sources.s1.port = 4141
# Channels
a3.channels.c1.type = memory
a3.channels.c1.capacity = 1000
a3.channels.c1.transactionCapacity = 100
# Sinks
a3.sinks.k1.type = logger
# Bind
a3.sources.s1.channels = c1
a3.sinks.k1.channel = c1
攔截器代碼一覽
package dell.flume.test;
import java.util.List;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
public class test1 implements Interceptor
{
public void initialize()
{
}
public Event intercept(Event event)
{
byte[] body = event.getBody();
if (body[0] < 'z' && body[0] > 'a')
{
event.getHeaders().put("type", "letter");
} else if (body[0] > '0' && body[0] < '9')
event.getHeaders().put("type", "number");
return event;
}
public List<Event> intercept(List<Event> events)
{
for (Event event : events) {
intercept(event);
}
return events;
}
public void close()
{
}
public static class Builder implements Interceptor.Builder
{
public Interceptor build()
{
return new test1();
}
public void configure(Context context)
{
}
}
}
運行時的截圖
為什么我的Headers里什么都沒有?我明明已經把我用Maven生成的Jar放到了flume/lib目錄下啊
以下是我所生成的Jar檔案截圖
以下是我引入的POM包
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>test.flume</groupId>
<artifactId>dell.flume.test</artifactId>
<version>0.0.1-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.flume</groupId>
<artifactId>flume-ng-core</artifactId>
<version>1.7.0</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.8.2</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.8.2</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>2.7.2</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.7.2</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>2.7.2</version>
</dependency>
<dependency>
<groupId>jdk.tools</groupId>
<artifactId>jdk.tools</artifactId>
<version>1.8</version>
<scope>system</scope>
<systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>
</dependency>
</dependencies>
</project>
版本與集群里的都是與之對應的
請問是我哪里做錯了么?
我把我生成Jar包的方式也寫出來吧
首先是對專案右鍵
然后先運行的是Maven clean
然后運行的是Maven build
再其次是生成Jar包Maven install
我拷貝到集群的jar包就是使用的Maven install
flume的是日志檔案沒有什么例外出現,我還是把他貼上吧
03 ??<8c>?<9c><88> 2020 18:22:26,861 INFO [agent-shutdown-hook] (org.apache.flume.instrumentation.MonitoredCounterGroup.stop:177) - Shutdown Metric for type: SINK, name: k1. sink.event.drain.attempt == 3
03 ??<8c>?<9c><88> 2020 18:22:26,862 INFO [agent-shutdown-hook] (org.apache.flume.instrumentation.MonitoredCounterGroup.stop:177) - Shutdown Metric for type: SINK, name: k1. sink.event.drain.sucess == 3
03 ??<8c>?<9c><88> 2020 18:22:26,862 INFO [agent-shutdown-hook] (org.apache.flume.sink.AbstractRpcSink.stop:320) - Rpc sink k1 stopped. Metrics: SINK:k1{sink.batch.underflow=3, sink.connection.failed.count=0, sink.connection.closed.count=1, sink.event.drain.attempt=3, sink.batch.complete=0, sink.event.drain.sucess=3, sink.connection.creation.count=1, sink.batch.empty=114}
亂碼不影響,因為沒有出現Java Error之類的錯誤
總之就這些了,預期的目標就是a1通過攔截器,將字母與數字分別列印到不同的虛擬機(Agent)上
問題比較長,但我希望描述的足夠清楚,望有緣人能給出個實際有效能解決的辦法,謝謝
哦對了,我嘗試過使用-C 指定jar包的方式了,無濟于事
uj5u.com熱心網友回復：
=================a1.sources = s1
=================
【上面宣告的是 s1， 下面使用的是 r1 】
===========================================
# Channel Interceptors
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = dell.flume.test.test1$Builder
# Channel Selector
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = type
a1.sources.r1.selector.mapping.letter = c1
a1.sources.r1.selector.mapping.number = c
===========================================
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/269290.html
標籤：Java EE