标签：Flume flume sinks kafka a1 k1 c1 KAFKA Kafka
1 简单实现
1）配置flume
# define a1.sources = r1 a1.sinks = k1 a1.channels = c1 # source a1.sources.r1.type = exec a1.sources.r1.command = tail -F /opt/module/data/flume.log # sink a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092 a1.sinks.k1.kafka.topic = first a1.sinks.k1.kafka.flumeBatchSize = 20 a1.sinks.k1.kafka.producer.acks = 1 a1.sinks.k1.kafka.producer.linger.ms = 1 # channel a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 # bind a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1
2） 启动kafka消费者
3） 进入flume根目录下，启动flume
$ bin/flume-ng agent -c conf/ -n a1 -f jobs/flume-kafka.conf
4） 向 /opt/module/data/flume.log里追加数据，查看kafka消费者消费情况
$ echo hello >> /opt/module/data/flume.log
2 数据分离
0）需求: 将flume采集的数据按照不同的类型输入到不同的topic中
将日志数据中带有atguigu的，输入到Kafka的first主题中，
将日志数据中带有shangguigu的,输入到Kafka的second主题中，
其他的数据输入到Kafka的third主题中
1） 编写Flume的Interceptor
package com.atguigu.kafka.flumeInterceptor;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import javax.swing.text.html.HTMLEditorKit;
import java.util.List;
import java.util.Map;
public class FlumeKafkaInterceptor implements Interceptor {
@Override
public void initialize() {
}
/**
* 如果包含"atguigu"的数据，发送到first主题
* 如果包含"sgg"的数据，发送到second主题
* 其他的数据发送到third主题
* @param event
* @return
*/
@Override
public Event intercept(Event event) {
//1.获取event的header
Map<String, String> headers = event.getHeaders();
//2.获取event的body
String body = new String(event.getBody());
if(body.contains("atguigu")){
headers.put("topic","first");
}else if(body.contains("sgg")){
headers.put("topic","second");
}
return event;
}
@Override
public List<Event> intercept(List<Event> events) {
for (Event event : events) {
intercept(event);
}
return events;
}
@Override
public void close() {
}
public static class MyBuilder implements Builder{
@Override
public Interceptor build() {
return new FlumeKafkaInterceptor();
}
@Override
public void configure(Context context) {
}
}
}
2）将写好的interceptor打包上传到Flume安装目录的lib目录下
3）配置flume
# Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = netcat a1.sources.r1.bind = 0.0.0.0 a1.sources.r1.port = 6666 # Describe the sink a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.topic = third a1.sinks.k1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092 a1.sinks.k1.kafka.flumeBatchSize = 20 a1.sinks.k1.kafka.producer.acks = 1 a1.sinks.k1.kafka.producer.linger.ms = 1 #Interceptor a1.sources.r1.interceptors = i1 a1.sources.r1.interceptors.i1.type = com.atguigu.kafka.flumeInterceptor.FlumeKafkaInterceptor$MyBuilder # # Use a channel which buffers events in memory a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1
4） 启动kafka消费者
5） 进入flume根目录下，启动flume
$ bin/flume-ng agent -c conf/ -n a1 -f jobs/flume-kafka.conf
6） 向6666端口写数据，查看kafka消费者消费情况
标签：Flume,flume,sinks,kafka,a1,k1,c1,KAFKA,Kafka 来源： https://www.cnblogs.com/qiu-hua/p/15224730.html
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。