flume自定义source
package me;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Random;
import org.apache.flume.Context;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;
public class MySource extends AbstractSource implements Configurable,
PollableSource {
@Override
public long getBackOffSleepIncrement()
{
//
TODO Auto-generated method stub
return 0;
}
@Override
public long getMaxBackOffSleepInterval()
{
//
TODO Auto-generated method stub
return 0;
}
@Override
public Status
process() throws EventDeliveryException
{
try {
while (true)
{
int max
= 20;
int min
= 10;
Random
random = new Random();
int s
= random.nextInt(max) % (max - min + 1)
+ min;
HashMap<String,
String> header = new HashMap<String,
String>();
header.put("id",
Integer.toString(s));
this.getChannelProcessor()
.processEvent(EventBuilder.withBody(Integer.toString(s),
Charset.forName("UTF-8"),
header));
Thread.sleep(1000);
}
} catch (InterruptedException
e) {
e.printStackTrace();
}
return null;
}
@Override
public void configure(Context
arg0) {
}
}
|
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
agent.channels
= ch-1
agent.sources
= src-1
agent.sinks
= sk1
agent.sources.src-1.type
= me.MySource
agent.channels.ch-1.type
= memory
agent.channels.ch-1.capacity
= 10000
agent.channels.ch-1.transactionCapacity
= 10000
agent.channels.ch-1.byteCapacityBufferPercentage
= 20
agent.channels.ch-1.byteCapacity
= 800000
agent.sinks.sk1.type
= logger
agent.sinks.sk1.channel
= ch-1
agent.sources.src-1.channels=ch-1
|
|
1
2
|
lihudeMacBook-Pro:~
SunAndLi$ cd hadoop-2.7.2/flume/
lihudeMacBook-Pro:flume
SunAndLi$ bin/flume-ng agent -c conf -f conf/mysql-source --name agent -Dflume.root.logger=INFO,console
|