【Hadoop大数据技术】 |
您所在的位置:网站首页 › 收集数据和分析数据的方案 › 【Hadoop大数据技术】 |
📖 前言:在大数据系统的开发中,数据收集工作无疑是开发者首要解决的一个难题,但由于生产数据的源头丰富多样,其中包含网站日志数据、后台监控数据、用户浏览网页数据等,数据工程师要想将它们分门别类的采集到HDFS系统中,就可以使用Apache Flume(数据采集)系统。 目录 🕒 1. Flume概述🕒 2. Flume运行机制🕒 3. Flume日志采集系统结构🕒 4. Flume的部署🕘 4.1 安装与配置🕘 4.2 Flume信息采集实例 🕒 5. Flume的采集方案🕘 5.1 Flume Sources🕘 5.2 Flume Channels🕘 5.3 Flume Sinks 🕒 6. Flume的可靠性保证🕘 6.1 负载均衡🕘 6.2 故障转移 🕒 7. Flume拦截器 🕒 1. Flume概述Flume原是Cloudera公司提供的一个高可用的、高可靠的、分布式海量日志采集、聚合和传输系统,而后纳入到了Apache旗下,作为一个顶级开源项目。Apache Flume不仅只限于日志数据的采集,由于Flume采集的数据源是可定制的,因此Flume还可用于传输大量事件数据,包括但不限于网络流量数据、社交媒体生成的数据、电子邮件消息以及几乎任何可能的数据源。 Flume自发展以来,分为两个版本,分别是Flume 0.9x版本,称为Flume-og(original generation,原始一代)和Flume 1.x版本,称为Flume-ng(next generation,下一代)。Flume-og版本由于早期设计时,存在不易扩展、代码臃肿等一系列问题,在被纳入到Apache软件基金会之后,开发人员便对其代码进行了重构,使Flume更易于使用和扩展,于是便出现了Flume-ng版本,采用三层架构,分别为agent,collector``和storage`,每一层均可以水平扩展。Flume-ng版本在实际开发中应用最为广泛,接下来,主要基于Flume-ng版本对Flume进行讲解。 🕒 2. Flume运行机制Flume的核心是把数据从数据源(例如Web服务器)通过数据采集器(Source)收集过来,再将收集的数据通过缓冲通道(Channel)汇集到指定的接收器(Sink)。 Flume基本架构中有一个Agent(代理),它是Flume的核心角色,Flume Agent是一个JVM进程,它承载着数据从外部源流向下一个目标的三个核心组件:Source、Channel和Sink。 Source:用于采集数据源的数据,并将数据写入到Channel。一个Source可以连接一个或多个Channel。Channel:用于缓存Source写入的数据,并将数据写入到Sink,待Sink将数据写入到存储设备或者下一个Source之后,Flume会删除Channel中缓存的数据。Sink:用于接收Channel写入的数据,并将数据写入到存储设备。在整个数据传输过程,即Source→Channel→Sink,Flume将流动的数据封装到一个事件(Event)中,它是Flume内部数据传输的基本单元。 🕒 3. Flume日志采集系统结构在实际开发中, Flume需要采集数据的类型多种多样,同时还会进行不同的中间操作,所以根据具体需求,可以将Flume日志采集系统分为简单结构和复杂结构。 简单结构通常应用于采集数据的数据源单一,数据内容简单,并且采集数据的存储设备单一,这时在Flume日志采集系统中,可以直接使用一个Agent来实现。 Flume复杂结构–多Agent:在某些实际应用场景中,Flume日志采集系统需要采集数据的数据源分布在不同的服务器上,可以再分配一个Agent来采集其他Agent从Web Server采集的日志,对这些日志进行汇总后写入到存储系统。 🕒 4. Flume的部署 🕘 4.1 安装与配置在Ubuntu中直接打开网站下载: 🔎 Flume官网 建议下载稳定版本apache-flume-1.9.0-bin.tar.gz 稳定版下载地址:http://archive.apache.org/dist/flume/stable/ 1、打开终端,解压安装包至路径 /opt,命令如下: sudo tar -zxvf apache-flume-1.9.0-bin.tar.gz -C /opt 2、将解压的文件夹重命名为flume并添加flume的权限 sudo mv apache-flume-1.9.0-bin/ ./flume sudo chown -R hadoop flume3、配置环境变量 将flume目录添加到path中,这样,启动flume就无需到/opt/flume目录下,大大的方便了flume的使用。编辑bashrc文件。 vim ~/.bashrc请在bashrc文件中添加如下内容,如果以前添加过JDK请勿重复添加: export JAVA_HOME=/usr/lib/jvm/jdk1.8.0_371 export FLUME_HOME=/opt/flume export FLUME_CONF_DIR=$FLUME_HOME/conf export PATH=$FLUME_HOME/bin:$PATH4、保存.bashrc文件并退出vim编辑器。然后,继续执行如下命令让.bashrc文件的配置立即生效: source ~/.bashrc5、启动flume,查看flume版本,确定flume是否安装成功,命令如下: flume-ng version 🕘 4.2 Flume信息采集实例1、使用Flume接收来自AvroSource的信息 AvroSource可以发送一个给定的文件到Flume,Flume接收以后可以进行处理后显示到屏幕上。 1)在/opt/flume/conf目录下创建Agent配置文件avro.conf,命令如下。 sudo vim /opt/flume/conf/avro.conf2)在avro.conf文件中写入如下内容 a1.sources=r1 a1.sinks=k1 a1.channels=c1 a1.sources.r1.type=avro a1.sources.r1.channels=c1 a1.sources.r1.bind=0.0.0.0 a1.sources.r1.port=4141 a1.sinks.k1.type=logger a1.channels.c1.type=memory a1.channels.c1.capacity=1000 a1.channels.c1.transactionCapacity=100 a1.sources.r1.channels=c1 a1.sinks.k1.channel=c13)启动Flume Agent a1,执行如下命令启动日志控制台 flume-ng agent -c . -f /opt/flume/conf/avro.conf -n a1 -Dflume.root.logger=INFO,console4)打开另外一个Linux终端,在/opt/flume目录下创建一个文件log.01,并在文件中加入一行内容“Hello Flume”,命令如下: sudo sh -c 'echo "Hello Flume"'> /opt/flume/log.015)再打开另外一个Linux终端,执行如下命令: flume-ng avro-client --conf conf -H localhost -p 4141 -F /opt/flume/log.01在该命令中,4141是前面文件avro.conf里自定义的端口号。 执行该命令后,AvroSource就向Flume发送了一个文件log.01,切换到第3)步的日志控制台端口,就可以看到Flume已经接收到了信息,如下图所示,通过最后一行可以看出Flume已经成功接收到了“Hello Flume”。 2、使用Flume接收来自NetSource的信息 NetcatSource可以把用户实时输入的信息持续不断地发给Flume,Flume处理后可以显示到输出屏幕上。 1)执行如下命令,在/opt/flume/conf目录下新建test.conf代理配置文件,并写入如下内容: sudo vim /opt/flume/conf/test.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.sinks.k1.type=logger a1.channels.c1.type=memory a1.channels.c1.capacity=1000 a1.channels.c1.transactionCapacity=100 a1.sources.r1.channels=c1 a1.sinks.k1.channel=c12)执行如下命令启动Flume Agent a1日志控制台 flume-ng agent --conf /opt/flume/conf --conf-file /opt/flume/conf/test.conf --name a1 -Dflume.root.logger=INFO,console3)再打开一个终端,输入如下命令: telnet localhost 44444该命令中的44444是前面自定义的test.conf文件中的端口号。执行后,出现如下图所示信息: 这个终端窗口称为“NetcatSource”终端窗口,在这个终端窗口中可以输入任意字符,该字符会被实时发送到Flume Agent a1,另外一个终端窗口“日志控制台”就会同步显示输入的内容,例如,在“NetcatSource”终端窗口输入“Hello Flume by HinsCoder”,如下图所示。 4)日志控制台终端窗口就会同步显示输入的内容,如下图所示。 🕒 5. Flume的采集方案 🕘 5.1 Flume Sources在编写Flume采集方案时,首先必须明确采集的数据源类型、出处;接着,根据这些信息与Flume已提供支持的Flume Sources进行匹配,选择对应的数据采集器类型(即sources.type);再根据选择的数据采集器类型,配置必要和非必要的数据采集器属性,Flume提供并支持的Flume Sources种类如下所示。 Avro SourceThrift SourceExec SourceJMS SourceSpooling Directory SourceTwitter 1% firehose SourceKafka SourceNetCat TCP SourceNetCat UDP SourceSequence Generator SourceSyslog TCP SourceMultiport Syslog TCP SourceSyslog UDP SourceHTTP SourceStress SourceAvro Legacy SourceThrift Legacy SourceCustom SourceScribe SourceTaildir Source监听Avro端口并从外部Avro客户端流中接收event数据,当与另一个Flume Agent上的Avro Sink配对时,可创建分层集合拓扑,利用Avro Source可以实现多级流动、扇出流、扇入流等效果,Avro Source常用配置属性如下。 属性名称默认值相关说明channels–type–组件类型名需必须是avrobind–要监听的主机名或IP地址port–要监听的服务端口threads–要生成的工作线程的最大数目sslFALSE将此设置为true以启用SSL加密,则还必须指定“keystore”和“keystore-password”keystore–SSL所必需的通往Java秘钥存储路径keystore-password–SSL所必需的Java密钥存储的密码Spooling Directory Source允许对指定磁盘上的文件目录进行监控来提取数据,它将查看文件的指定目录的新增文件,并将文件中的数据读取出来。 Spooling Directory Source常用配置属性如下表所示。 属性名称默认值相关说明channels–type–组件类型名需必须是spooldirspoolDir–从中读取文件的目录fileSuffix.COMPLETED附加到完全摄取的文件后缀deletePolicynever何时删除已完成的文件:never或immediatefileHeaderFALSE是否添加存储绝对路径文件名的标头includePattern^.*$正则表达式,指定要包含的文件ignorePattern^$正则表达式指定要忽略的文件Taildir Source用于观察指定的文件,几乎可以实时监测到添加到每个文件的新行。如果文件正在写入新行,则此采集器将重试采集它们以等待写入完成,Source常用配置属性如下所示。 属性名称默认值相关说明channels–type–组件类型名需必须是TAILDIRfilegroups–以空格分隔的文件组列表filegroups.–文件组的绝对路径idleTimeout120000关闭非活动文件的时间(毫秒)writePosInterval3000写入位置文件上每个文件的最后位置的间隔时间batchSize100一次读取和发送到通道的最大行数backoffSleepIncrement1000当最后一次尝试未找到任何新数据时,每次重新尝试轮询新数据之间的最大时间延迟fileHeaderFALSE是否添加存储绝对路径文件名的标头fileHeaderKeyfile将绝对路径文件名附加到event header时使用的header关键字HTTP Source可以通过HTTP POST和GET请求方式接收event数据,GET通常只能用于测试使用,POST请求发送的所有的events都被认为是一个批次,会在一个事务中插入channel,Taildir Source常用配置属性如下所示。 属性名称默认值相关说明channels–type组件类型名需必须是httpport–采集源要绑定的端口bind0.0.0.0要监听绑定的主机名或IP地址handlerorg.apache.flume.source.http.JSONHandlerhandler类的全路径名 🕘 5.2 Flume ChannelsChannels通道是event在Agent上暂存的存储库,Source向Channel中添加event,Sink在读取完数据后再删除它。在配置Channels时,需要明确的就是将要传输的sources数据源类型;根据这些信息结合开发中的实际需求,选择Flume已提供的支持的Flume Channels;再根据选择的Channel类型,配置必要和非必要的Channel属性,Flume提供并支持的Flume Channels种类如下所示。 Memory ChannelJDBC ChannelKafka ChannelFile ChannelSpillable Memory ChannelPseudo Transaction ChannelCustom ChannelMemory Channel会将event存储在具有可配置最大尺寸的内存队列中,适用于需要更高吞吐量的流量,但在Agent发生故障时会丢失部分阶段数据,下表为Memory Channel常用配置属性。 属性名称默认值相关说明type–组件类型名需必须是memorycapacity100存储在channel中的最大event数transactionCapacity100channel从source接收或向sink传递的每个事务中最大event数keep-alive3添加或删除event的超时时间(秒)byteCapacityBufferPercentage20定义byteCapacity与channel中所有event所占百分比byteCapacity等于JVM可用的最大内存的80%允许此channel中所有event的的最大内存字节数总和File Channel是Flume的持久通道,它将所有event写入磁盘,因此不会丢失进程或机器关机、崩溃时的数据。File Channel通过在一次事务中提交多个event来提高吞吐量,做到了只要事务被提交,那么数据就不会有丢失,File Channel常用配置属性如下所示。 属性名称默认值相关说明type–组件类型名需必须是filecheckpointDir1~/.flume/file-channel/checkpoint检测点文件所存储的目录useDualCheckpointsFALSE备份检测点如果设置为true,backupChec kpointDir必须设置backupCheckpointDir–备份检查点目录。此目录不能与数据目录或检查点目录相同dataDirs~/.flume/file-channel/data数据存储所在的目录设置transactionCapacity10000事务容量的最大值设置checkpointInterval30000检测点之间的时间值设置(单位微秒)maxFileSize2146435071一个单一日志的最大值设置(以字节为单位)capacity1000000channel的最大容量transactionCapacity10000事务容量的最大值设置 🕘 5.3 Flume SinksFlume Soures采集的数据通过Channels通道流向Sink中,此时Sink类似一个集结的递进中心,它需要根据需求进行配置,从而最终选择发送目的地。 配置Sinks时,明确将要传输的数据目的地、结果类型;然后根据实际需求信息,选择Flume已提供支持的Flume Sinks;再根据选择的Sinks类型,配置必要和非必要的Sinks属性。Flume提供并支持的Flume Sinks种类如下所示。 HDFS SinkHive SinkLogger SinkAvro SinkThrift SinkIRC SinkFile Roll SinkNull SinkHBaseSinkAsyncHBase SinkMorphlineSolr SinkElasticSearch SinkKite Dataset SinkKafka SinkHTTP SinkCustom SinkHDFS Sink将event写入Hadoop分布式文件系统(HDFS),它目前支持创建文本和序列文件,以及两种类型的压缩文件,下表为HDFS Sink常用配置属性。 属性名称默认值相关说明channel–type–组件类型名需必须是hdfshdfs.path–HDFS目录路径hdfs.filePrefixFlumeData为在hdfs目录中由Flume创建的文件指定前缀hdfs.roundFALSE是否应将时间戳向下舍入hdfs.roundValue1舍入到此最高倍数,小于当前时间hdfs.roundUnitsecond舍入值的单位 - 秒、分钟或小时hdfs.rollInterval30滚动当前文件之前等待的秒数hdfs.rollSize1024触发滚动的文件大小,以字节为单位hdfs.rollCount10在滚动之前写入文件的事件数Logger Sink用于记录INFO级别event,它通常用于调试。Logger Sink接收器的不同之处是它不需要在“记录原始数据”部分中说明额外的配置,Logger Sink常用配置属性如下所示。 属性名称默认值相关说明channel–type–组件类型名需必须是loggermaxBytesToLog16要记录的event body的最大字节数Avro Sink形成Flume分层收集支持的一半,发送到此接收器的Flume event转换为Avro event,并发送到对应配置的主机名/端口,event将从配置的channel中批量获取配置的批处理大小,Avro Sink常用配置属性如下所示。 属性名称默认值相关说明channel–type–组件类型名需必须是avrohostname–要监听的主机名或IP地址port–要监听的服务端口batch-size100要一起批量发送的event数connect-timeout20000允许第一次(握手)请求的时间量(ms)request-timeout20000在第一个之后允许请求的时间量(ms) 🕒 6. Flume的可靠性保证 🕘 6.1 负载均衡配置的采集方案是通过唯一一个Sink作为接收器接收后续需要的数据,但会出现当前Sink故障或数据收集请求量较大的情况,这时单一Sink配置可能就无法保证Flume开发的可靠性。因此,Flume 提供Flume Sink Processors解决上述问题。 Sink处理器允许定义Sink groups,将多个sink分组到一个实体中,Sink处理器就可通过组内多个sink为服务提供负载均衡功能。 负载均衡接收器处理器(Load balancing sink processor)提供了在多个sink上进行负载均衡流量的功能,它维护一个活跃的sink索引列表,需在其上分配负载,还支持round_robin(轮询)和random(随机)选择机制进行流量分配,默认选择机制为round_robin。Load balancing sink processor提供的配置属性如下所示。 属性名称默认值相关说明sinks–以空格分隔的参与sink组的sink列表processor.typedefault组件类型名需必须是load_balanceprocessor.backoffFALSE设置失败的sink进入黑名单processor.selectorround_robin选择机制processor.selector.maxTimeOut30000失败sink放置在黑名单的超时时间 🕘 6.2 故障转移故障转移接收器处理器(Failover Sink Processor)维护一个具有优先级的sink列表,保证在处理event时,只需有一个可用的sink即可。 故障转移机制工作原理是将故障的sink降级到故障池中,在池中为它们分配一个冷却期,在重试之前冷却时间会增加,当sink成功发送event后,它将恢复到活跃池中。Failover Sink Processor提供的配置属性如下所示。 属性名称默认值相关说明sinks–以空格分隔的参与sink组的sink列表processor.typedefault组件类型名需必须是failoverprocessor.priority.–设置sink的优先级取值processor.maxpenalty30000失败sink的最大退避时间 🕒 7. Flume拦截器Flume Interceptors(拦截器)用于对Flume系统数据流中event的修改操作。使用Flume拦截器时,只需参考官方配置属性在采集方案中选择性的配置即可,当涉及到配置多个拦截器时,拦截器名称间需用空格分隔,且拦截器配置顺序就是拦截顺序。Flume 1.9.0版本中,Flume提供并支持的拦截器有很多,具体如下所示。 Timestamp InterceptorHost InterceptorStatic InterceptorRemove Header InterceptorUUID InterceptorMorphline InterceptorSearch and Replace InterceptorRegex Filtering InterceptorRegex Extractor InterceptorTimestamp Interceptor(时间戳拦截器)将流程执行时间插入到event的header头部,此拦截器插入带有timestamp键的标头,其值为对应时间戳。若配置中已存在时间戳时,此拦截器可保留现有时间戳,Timestamp Interceptor提供的常用配置属性如下所示。 属性名称默认值相关说明type–组件类型名需必须是timestampheadertimestamp用于放置生成的时间戳的标头的名称preserveExistingFALSE如果时间戳已存在,是否应保留, true或falseStatic Interceptor(静态拦截器)允许用户将具有静态值的静态头附加到所有event。当前不支持一次指定多个header头,但是用户可定义多个Static Interceptor来为每一个拦截器都追加一个header,Static Interceptor提供的常用配置属性如下所示。 属性名称默认值相关说明type–组件类型名需必须是staticpreserveExistingTRUE如果配置的header已存在,是否应保留keykey应创建的header的名称valuevalue应创建的header对应的静态值OK,以上就是本期知识点“Flume日志采集系统”的知识啦~~ ,感谢友友们的阅读。后续还会继续更新,欢迎持续关注哟📌~ 💫如果有错误❌,欢迎批评指正呀👀~让我们一起相互进步🚀 🎉如果觉得收获满满,可以点点赞👍支持一下哟~ ❗ 转载请注明出处 作者:HinsCoder 博客链接:🔎 作者博客主页 |
今日新闻 |
点击排行 |
|
推荐新闻 |
图片新闻 |
|
专题文章 |
CopyRight 2018-2019 实验室设备网 版权所有 win10的实时保护怎么永久关闭 |