使用Flume将日志导入OSS-阿里云开发者社区

开发者社区> 开发与运维> 正文
登录阅读全文

使用Flume将日志导入OSS

简介:

前言

Apache Flume是一个高可用、高可靠的分布式日志采集、聚合与传输的系统。它基于流式的数据传输,架构简单、灵活。它简单可扩展的模型,也适合在线的数据分析。

image
上图是它的简单数据流模型。Flume的数据流由Event贯穿始终,这个Event由外部的Source生成(如Web Server),携带日志数据并带有一些额外的信息。Source捕获到Event后会进行格式化,然后会推送到一个或者多个Channel中。可以认为Channel是一个Event的缓冲区,它将负责保存Event直到Sink处理完该事件。

Flume的核心运行组件是Agent,一个Agent是一个JVM,是一个完整的数据收集工具。Flume的Agent由三部分组成:Source、Channel和Sink。其中,Flume目前支持了很多种类型的Sink: HDFS Sink、HBase Sink、Thrift Sink、Avro Sink、Elastic Search Sink等。本文主要介绍如何使用HDFS Sink使Flume能够将数据导入到OSS。

配置

配置主要分三块:Java、Hadoop、Flume。配置Java比较简单,设置好JAVA_HOME即可。

Hadoop

首先需要下载Hadoop(以3.0.1版本为例)。Hadoop从2.9.1与3.0.0开始支持OSS,作为默认支持的文件系统。Hadoop的配置可以参考这篇文章。主要是修改$HADOOP_HOME/ etc/hadoop/core-site.xml配置文件,并且配好fs.oss.endpoint、fs.oss.accessKeyId、fs.oss.accessKeySecret和fs.oss.impl。

Flume

下载Apache Flume 1.8.0,并且修改conf目录下的文件

mv conf/flume-conf.properties.template cpnf/flume-conf.properties
mv conf/flume-env.sh.template conf/flume-env.sh

修改conf/flume-env.sh文件,添加JAVA_HOME与FLUME_CLASSPATH,其中,需要将HADOOP的lib与conf目录放入到FLUME_CLASSPATH里面。示例如下

export JAVA_HOME=/usr/lib/jdk1.8.0_152
export FLUME_CLASSPATH=$HADOOP_HOME/etc/hadoop:$HADOOP_HOME/share/hadoop/common:$HADOOP_HOME/share/hadoop/hdfs:$HADOOP_HOME/share/hadoop/tools/lib/*

新增conf/hdfs.properties,在这个里面配置Source、Channel与Sink。假设我们想监控某个目录下新产生的文件,并把新产生的文件写到OSS里面去,我们可以这样配置:

LogAgent.sources = apache
LogAgent.channels = fileChannel
LogAgent.sinks = HDFS

#sources config
LogAgent.sources.apache.type = spooldir
LogAgent.sources.apache.spoolDir = /tmp/logs
LogAgent.sources.apache.channels = fileChannel
LogAgent.sources.apache.basenameHeader = true
LogAgent.sources.apache.basenameHeaderKey = fileName

#sinks config
LogAgent.sinks.HDFS.channel = fileChannel
LogAgent.sinks.HDFS.type = hdfs
LogAgent.sinks.HDFS.hdfs.path = oss://{Your Bucket Name}/logs/%Y%m%d
LogAgent.sinks.HDFS.hdfs.fileType = DataStream
LogAgent.sinks.HDFS.hdfs.writeFormat = TEXT
LogAgent.sinks.HDFS.hdfs.filePrefix = %{fileName}.%H:%M:%S
LogAgent.sinks.HDFS.hdfs.fileSuffix = .log
LogAgent.sinks.HDFS.hdfs.batchSize = 1000
LogAgent.sinks.HDFS.hdfs.rollSize = 0
LogAgent.sinks.HDFS.hdfs.rollCount = 0
LogAgent.sinks.HDFS.hdfs.rollInterval = 30
LogAgent.sinks.HDFS.hdfs.useLocalTimeStamp = true

#channels config
LogAgent.channels.fileChannel.type = memory
LogAgent.channels.fileChannel.capacity = 1000000
LogAgent.channels.fileChannel.transactionCapacity = 10000

注意修改 LogAgent.sinks.HDFS.hdfs.path配置项为用户自己在OSS上路径。

启动Agent

bin/flume-ng agent --conf-file  conf/hdfs.properties -c conf/ --name LogAgent -Dflume.root.logger=console

这个时候会发现抛出异常:

Exception in thread "SinkRunner-PollingRunner-DefaultSinkProcessor" java.lang.NoSuchFieldError: INSTANCE
    at org.apache.http.conn.ssl.SSLConnectionSocketFactory.<clinit>(SSLConnectionSocketFactory.java:144)
    at com.aliyun.oss.common.comm.DefaultServiceClient.createHttpClientConnectionManager(DefaultServiceClient.java:244)
    at com.aliyun.oss.common.comm.DefaultServiceClient.<init>(DefaultServiceClient.java:85)
    at com.aliyun.oss.OSSClient.<init>(OSSClient.java:209)
    at org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystemStore.initialize(AliyunOSSFileSystemStore.java:150)
    at org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem.initialize(AliyunOSSFileSystem.java:318)
    at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3288)
    at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:123)
    at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:3337)
    at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:3305)
    at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:476)
    at org.apache.hadoop.fs.Path.getFileSystem(Path.java:361)
    at org.apache.flume.sink.hdfs.BucketWriter$1.call(BucketWriter.java:260)
    at org.apache.flume.sink.hdfs.BucketWriter$1.call(BucketWriter.java:252)
    at org.apache.flume.sink.hdfs.BucketWriter$9$1.run(BucketWriter.java:701)
    at org.apache.flume.auth.SimpleAuthenticator.execute(SimpleAuthenticator.java:50)
    at org.apache.flume.sink.hdfs.BucketWriter$9.call(BucketWriter.java:698)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

这个异常是由于Apache Flume使用的http client版本比较低的缘故,把flume lib目录下的httpclient-4.2.1.jar和httpcore-4.4.1.jar删除,重新启动即可。

然后我们往刚才配置的LogAgent.sources.apache.spoolDir里面放文件,使用如下命令

for((i=0;i<=5;i++)); do echo "Hello, world" >> /tmp/logs/sample.data.$i; sleep 30s; done

Flume会每隔30秒钟检查这个目录,最终的效果如下:
image

参考资料

  1. http://flume.apache.org/FlumeUserGuide.html
  2. https://yq.aliyun.com/articles/292792?spm=a2c4e.11155435.0.0.7ccba82fbDwfhK

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
开发与运维
使用钉钉扫一扫加入圈子
+ 订阅

集结各类场景实战经验,助你开发运维畅行无忧

其他文章