前提：为了测试网络获取是否可行，我们通过Netcat来进行测验
Netcat ：网络测试工具
Netcat 是一款简单的Unix工具，使用UDP和TCP协议。 它是一个可靠的容易被其他程序所启用的后台操作工具，同时它也被用作网络的测试工具或黑客工具。 使用它你可以轻易的建立任何连接。内建有很多实用的工具。
使用
| 参数 | 说明 |
|---|---|
| -4 | 仅使用IPv4 |
| -6 | 仅使用IPv6 |
| -U, –unixsock | 仅使用Unix域套接字 |
| –vsock | 仅使用vsock套接字 |
| -C, –crlf | 将CRLF用于EOL序列 |
| -c, –sh-exec | 通过/bin/sh执行给定命令 |
| -e, –exec | 执行给定命令 |
| –lua-exec | 执行给定的Lua脚本 |
| -g hop1[,hop2,…] | 松散的源路由跳跃点（最大8个） |
| -G | 松散的源路由跳跃指针（4、8、12，…） |
| -m, –max-conns | 最大同时连接 |
| -h, –help | 显示此帮助屏幕 |
|
-d, –delay |
在读/写之间等待时间 |
| -o, –output | 将会话数据转储到文件中 |
| -x, –hex-dump | 将会话数据以十六进制形式转储到文件中 |
|
-i, –idle-timeout |
空闲读/写超时时间 |
| -p, –source-port port | 指定要使用的源端口 |
| -s, –source addr | 指定要使用的源地址（不影响-l） |
| -l, –listen | 绑定并监听传入的连接 |
| -k, –keep-open | 在侦听模式下接受多个连接 |
| -n, –nodns | 不通过DNS解析主机名 |
| -t, –telnet | 回答Telnet协商 |
| -u, –udp | 使用UDP代替默认TCP |
| –sctp | 使用SCTP代替默认TCP |
| -v, –verbose | 使用SCTP代替默认TCP |
|
-w, –wait |
连接超时时间 |
| -z | 零I/O模式，仅报告连接状态 |
安装 Netcat
安装netcat服务
yum install -y nc
开启服务
nc -lk 44444
启动命令
这里要先跳转到下面的netcat 方式，先编写conf 文件
【console:从控制台输入】
./bin/flume-ng agent --name a1 --conf ./conf/ --conf-file ./conf/testjob/netcat-flume-logger.conf -Dflume.root.logger=INFO,console
安装客户端
yum list telnet.*
开启客户端
这里的端口号可以随便起，但是不要和已经使用的冲突，而且要和最上面的一样
使用 netstat -lnpt 命令查看使用的端口，命令不存在的，请安装 net-tools
telnet localhost 44444
Flume
是 Cloudera提供的一个高可用的，高可靠的，分布式的海量
日志采集
、聚合和传输的系统，Flume支持在日志系统中定制各类数据发送方，用于收集数据；同时，Flume提供对数据进行简单处理，并写到各种数据接受方（可定制）的能力。
当前Flume有两个版本Flume 0.9X版本的统称Flume-og，Flume1.X版本的统称Flume-ng。所以一般都是用ng版本
Agent主要由:source,channel,sink三个组件组成.
Source:
从数据发生器接收数据,并将接收的数据以Flume的event格式传递给一个或者多个通道channel,Flume提供多种数据接收的方式,比如Avro,Thrift,twitter1%等
Channel:
channel是一种短暂的存储容器,它将从source处接收到的event格式的数据缓存起来,直到它们被sinks消费掉,它在source和sink间起着桥梁的作用,channel是一个完整的事务,这一点保证了数据在收发的时候的一致性. 并且它可以和任意数量的source和sink链接. 支持的类型有: JDBC channel , File System channel , Memory channel等.
sink:
sink将数据存储到集中存储器比如Hbase和HDFS,它从channels消费数据(events)并将其传递给目标地. 目标地可能是另一个sink,也可能HDFS,HBase.
下载并安装
官方下载链接：
点这里啊
tar …
mv …
配置flume.env.sh 文件
首先进入conf 文件夹
cp flume.env.sh.templete flume.env.sh
vi flume.env.sh
1、配置jdk路径
2、配置需要的JVM内存大小
根据自己的机器分配内存大小
编写conf文件
创建了一个文件夹，用于测试
mkdir testjob
cd testjob
1、通过netcat方式
vi netcat-flume-logger.conf
文件内容为
a1.sources=r1
a1.sinks=k1
a1.channels=c1
a1.sources.r1.type=netcat
a1.sources.r1.bind=localhost
a1.sources.r1.port=44444
a1.channels.c1.type=memory
a1.channels.c1.capacity=1000
a1.channels.c1.transactionCapacity=100
a1.sinks.k1.type=logger
a1.sources.r1.channels=c1
a1.sinks.k1.channel=c1
2、通过文件
a1.sources=r1
a1.sinks=k1
a1.channels=c1
a1.sources.r1.type=exec
a1.sources.r1.command=tail -f /opt/flumelogfile/flumedemo.log
a1.channels.c1.type=memory
a1.channels.c1.capacity=1000
a1.channels.c1.transactionCapacity=100
a1.sinks.k1.type=logger
a1.sources.r1.channels=c1
a1.sinks.k1.channel=c1
3、通过文件夹
events.sources=eventsSource
events.channels=eventsChannel
events.sinks=eventsSink
events.sources.eventsSource.type=spooldir
events.sources.eventsSource.spoolDir=/opt/events
events.sources.eventsSource.deserializer=LINE
events.sources.eventsSource.deserializer=maxLineLength=32000
events.sources.eventsSource.includePattern=events_[0-9]{4}-[0-9]{2}-[0-9]{2}.csv
events.channels.eventsChannel.type=file
events.channels.eventsChannel.checkpointDir=/opt/checkpoint/events
events.channels.eventsChannel.dataDirs=/opt/data/events
events.sinks.eventsSink.type=logger
events.sources.eventsSource.channels=eventsChannel
events.sinks.eventsSink.channel=eventsChannel
【注意：在对应的路径下创建对应的文件夹】
4、写到HDFS
user_friend.sources=userfriendSource
user_friend.channels=userfriendChannel
user_friend.sinks=userfriendSink
user_friend.sources.userfriendSource.type=spooldir
user_friend.sources.userfriendSource.spoolDir=/opt/events
user_friend.sources.userfriendSource.deserializer=LINE
user_friend.sources.userfriendSource.deserializer.maxLineLength=32000
user_friend.sources.includeParttern=user_friends_[0-9]{4}-[0-9]{2}.csv
user_friend.channels.userfriendChannel.type=file
user_friend.channels.userfriendChannel.checkpointDir=/opt/checkpoint/user-friend
user_friend.channels.userfriendChannel.dataDirs=/opt/data/user-friend
user_friend.sinks.userfriendSink.type=hdfs
user_friend.sinks.userfriendSink.hdfs.fileType=DataStream
user_friend.sinks.userfriendSink.hdfs.filePrefix=userfriend
user_friend.sinks.userfriendSink.hdfs.fileSuffix=.csv
user_friend.sinks.userfriendSink.hdfs.path=hdfs://192.168.159.100:9000/flumefile/userfriend/%Y-%m-%d
user_friend.sinks.userfriendSink.hdfs.useLocalTimeStamp=true
user_friend.sinks.userfriendSink.hdfs.batchSize=640
user_friend.sinks.userfriendSink.hdfs.rollCount=0
user_friend.sinks.userfriendSink.hdfs.rollSize=6400000
user_friend.sinks.userfriendSink.hdfs.rollInterval=3
user_friend.sources.userfriendSource.channels=userfriendChannel
user_friend.sinks.userfriendSink.channel=userfriendChannel