flume通过tcp/udp采集数据并存到kafka配置及操作方式

/*官方提供的kafka sink*/
a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = mytopic
a1.sinks.k1.kafka.bootstrap.servers = localhost:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
a1.sinks.ki.kafka.producer.compression.type = snappy

/**在kafka中创建一个名为 flumetest的topic*/
bin/kafka-console-consumer.sh --zookeeper mpc5:2181,mpc6:2181,mpc7 --from-beginning --topic flumetest
/**kafka中在控制台上消费消息队列的启动方式*/
bin/kafka-console-consumer.sh --zookeeper mpc5:2181,mpc6:2181,mpc7 --from-beginning --topic flumetest

/*比较好用的kafka sink 但是不可靠*/
producer.sinks.r.type = org.apache.flume.plugins.KafkaSink
producer.sinks.r.metadata.broker.list=dn1:9092,dn2:9092,dn3:9092
producer.sinks.r.partition.key=0
producer.sinks.r.partitioner.class=org.apache.flume.plugins.SinglePartition
producer.sinks.r.serializer.class=kafka.serializer.StringEncoder
producer.sinks.r.request.required.acks=0
producer.sinks.r.max.message.size=1000000
producer.sinks.r.producer.type=sync
producer.sinks.r.custom.encoding=UTF-8
producer.sinks.r.custom.topic.name=test

/*一个配置文件开始*/

a1.sources = r1
a1.channels = c1
a1.sinks =k1

#source
a1.sources.r1.type = netcat
a1.sources.r1.bind = mpc1
a1.sources.r1.port = 11203
a1.sources.r1.channels = c1

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

#sink
a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = flumetest
a1.sinks.k1.kafka.bootstrap.servers = mpc5:9092,mpc6:9092,mpc7:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 0
a1.sinks.k1.kafka.producer.compression.type = snappy
a1.sinks.k1.kafka.producer.zk.connect=mpc5:2181,mpc6:2181,mpc7:2181
a1.sinks.k1.kafka.producer.serializer.class=kafka.serializer.StringEncoder

/*一个配置文件结束*/

\x0D

/*tcp采集源*/

a1.sources = r1
a1.channels = c1
a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 5140
a1.sources.r1.host = localhost
a1.sources.r1.channels = c1

/*多个采集源的flume，分别是tcp和udp；sink到kafka中开始*/
a1.sources = r1 r2
a1.channels = c1
a1.sinks =k1

#source
a1.sources.r1.type = multiport_syslogtcp
a1.sources.r1.channels = c1
a1.sources.r1.host = mpc1
a1.sources.r1.ports = 10001 10002 10003
a1.sources.r1.portHeader = port

a1.sources.r2.type = syslogudp
a1.sources.r2.port = 5140
a1.sources.r2.host = mpc1
a1.sources.r2.channels = c1

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

/*多个采集源的flume，分别是tcp和udp；sink到kafka中结束*/

/*启动flume的方法*/
bin/flume-ng agent -n a1 -c conf -f conf/flume-conf.properties

/*删除一个topic在zookeeper上的元数据*/
bin/kafka-run-class.sh kafka.admin.DeleteTopicCommand --topic mytri1 --zookeeper mpc5:2181

/**flume到kafka利用id来进行分区的配置*/

a1.sources = r1 r2
a1.channels = c1
a1.sinks =k1

#source
a1.sources.r1.type = multiport_syslogtcp
a1.sources.r1.channels = c1
a1.sources.r1.host = mpc1
a1.sources.r1.ports = 10001 10002 10003
a1.sources.r1.portHeader = port
a1.sources.r1.interceptors = i2
a1.sources.r1.interceptors.i2.type=org.apache.flume.sink.solr.morphline.UUIDInterceptor$Builder
a1.sources.r1.interceptors.i2.headerName=key
a1.sources.r1.interceptors.i2.preserveExisting=false

a1.sources.r2.type = syslogudp
a1.sources.r2.port = 5140
a1.sources.r2.host = mpc1
a1.sources.r2.channels = c1
a1.sources.r2.interceptors = i2
a1.sources.r2.interceptors.i2.type=org.apache.flume.sink.solr.morphline.UUIDInterceptor$Builder
a1.sources.r2.interceptors.i2.headerName=key
a1.sources.r2.interceptors.i2.preserveExisting=false

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

#sink
a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = mytri3
a1.sinks.k1.kafka.bootstrap.servers = mpc5:9092,mpc6:9092,mpc7:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 0
a1.sinks.k1.kafka.producer.zk.connect=mpc5:2181,mpc6:2181,mpc7:2181
a1.sinks.k1.kafka.producer.serializer.class=kafka.serializer.StringEncoder

flume通过tcp/udp采集数据并存到kafka配置及操作方式相关推荐

moxa Nport配置tcp/udp实现数据发送和回传(补充数据接收错乱处理方法)
moxa Nport配置tcp/udp实现数据发送和回传–记一次吐血经历目的现有一内网环境需要连外网进行钉钉推送,开始使用串口连接实体电脑的方式实现,现需要撤除实体电脑,故选用串口服务器的方式实现 ...
TCP,UDP发送数据包大小浅析
MTU最大传输单元,这个最大传输单元实际上和链路层协议有着密切的关系,EthernetII帧的结构DMAC+SMAC+Type+Data+CRC由于以太网传输电气方面的限制,每个以太网帧都有最小的大小 ...
TCP,UDP伪首部描述
伪首部(pseudo header),通常有TCP伪首部和UDP伪首部.在UDP/TCP伪首部中,包含32位源IP地址,32位目的IP地址,8位填充0,8位协议,16位TCP/UDP长度.通过伪首部的 ...
以太网，IP,TCP,UDP数据包分析【转】
原文地址:http://www.cnblogs.com/feitian629/archive/2012/11/16/2774065.html 1.ISO开放系统有以下几层: 7 应用层 6 表示层 5 ...
以太网数据包、IP包、TCP/UDP 包的结构（转）
源:以太网数据包.IP包.TCP/UDP 包的结构版本号(Version):长度4比特.标识目前采用的IP协议的版本号.一般的值为0100(IPv4),0110(IPv6). IP包头长度(Head ...
Tcp方式采集CNC兄弟设备数据
转载地址:http://www.cnblogs.com/jonney-wang/p/6238768.html Tcp方式采集CNC兄弟设备数据先说下为了采集CNC兄弟设备的数据可谓是一波三折. 因为 ...
flume采集数据易出现的bug
1.内存不足修改文件内容 <文件根目录>/bin/flume-ng JAVA_OPTS="-Xms100m -Xmx4096m -Dcom.sun.management. ...
flume+kafka整合采集数据案例
一.flume简介 1.1.1 概述 Flume是一个分布式.可靠.和高可用的海量日志采集.聚合和传输的系统. Flume可以采集文件,socket数据包.文件.文件夹.kafka等各种形式源数据,又 ...
Flume跨服务器采集数据
在大数据的采集中,我们常用Flume来进行数据的采集,一般的我们会从Web Server服务器中收集数据,将数据存储在另一台服务器的hdfs文件系统做离线分析或者sink到另一台服务器的kafka消息 ...

flume通过tcp/udp采集数据并存到kafka配置及操作方式

flume通过tcp/udp采集数据并存到kafka配置及操作方式相关推荐

最新文章

热门文章