1

我正在尝试了解如何使用 flume-ng 跟踪文件，以便可以将数据推送到 HDFS。在第一个实例中，我设置了一个简单的 conf 文件：

tail1.sources = source1
tail1.sinks = sink1
tail1.channels = channel1

tail1.sources.source1.type = exec
tail1.sources.source1.command = tail -F /var/log/apache2/access.log
tail1.sources.source1.channels = channel1

tail1.sinks.sink1.type = logger

tail1.channels.channel1.type = memory
tail1.channels.channel1.capacity = 1000
tail1.channels.channel1.transactionCapacity = 100

tail1.sources.source1.channels = channel1
tail1.sinks.sink1.channel = channel1

这是一个测试，我的期望是我会在控制台上看到输出。我使用以下命令运行它：

flume-ng agent --conf-file tail1.conf -n tail1 -Dflume.root.logger=DEBUG,INFO,console

我得到以下输出：

验证后水槽配置包含代理配置：[tail1] 12/12/05 11:01:07 INFO properties.PropertiesFileConfigurationProvider：创建通道 12/12/05 11:01:08 INFO instrumentation.MonitoredCounterGroup：监控类型的计数器组：CHANNEL，名称：channel1，注册成功。12/12/05 11:01:08 INFO properties.PropertiesFileConfigurationProvider：创建通道 channel1 12/12/05 11:01:08 INFO sink.DefaultSinkFactory：创建接收器实例：sink1，类型：记录器 12/12/05 11： 01:08 INFO nodemanager.DefaultLogicalNodeManager：开始新配置：{ sourceRunners:{source1=EventDrivenSourceRunner: { source:org.apache.flume.source.ExecSource@1839aa9 }} sinkRunners:{sink1=SinkRunner: { policy:org.apache. flume.sink.DefaultSinkProcessor@11f0c98 counterGroup:{ name:null 计数器：

然而，没有进一步的事情发生。

我有另一个会话，其中有以下命令：

tail -F /var/log/apache2/access.log

我可以在哪里看到正在写入的文件：

192.168.1.81 - - [05/Dec/2012:10:58:07 +0000] "GET / HTTP/1.1" 200 483 "-" "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.95 Safari/537.11"
192.168.1.81 - - [05/Dec/2012:10:58:07 +0000] "GET /favicon.ico HTTP/1.1" 404 502 "-" "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.95 Safari/537.11"
192.168.1.81 - - [05/Dec/2012:10:58:21 +0000] "GET / HTTP/1.1" 304 209 "-" "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.95 Safari/537.11"
192.168.1.81 - - [05/Dec/2012:10:58:22 +0000] "GET /favicon.ico HTTP/1.1" 404 502 "-" "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.95 Safari/537.11"

你能帮我吗？我彻底糊涂了。

4

4 回答 4

2

您的配置文件看起来不错。我在 CDH4 中使用它并按照您的预期工作，我所做的只是更改尾部的日志文件位置。我在控制台上看到了输出。在我的情况下，新的日志数据被连续写入我正在跟踪的文件中。您的数据中的时间戳使您的示例看起来可能并非如此。

这是一个更完整的 conf 示例，更符合我认为您正在尝试做的事情。它会拖尾文件并每 10 分钟或 10K 记录写入一个新的 HDFS 文件。将agent1.sources.source1.command更改为您的 tail 命令，并根据您的 HDFS 配置更改agent1.sinks.sink1.hdfs.path和agent1.sinks.sink1.hdfs.filePrefix 。

# A single-node Flume configuration
# uses exec and tail and will write a file every 10K records or every 10 min
# Name the components on this agent
agent1.sources = source1
agent1.sinks = sink1
agent1.channels = channel1

# Describe/configure source1
agent1.sources.source1.type = exec
agent1.sources.source1.command = tail -f /home/cloudera/LogCreator/fortune_log.log

# Describe sink1
agent1.sinks.sink1.type = hdfs
agent1.sinks.sink1.hdfs.path = hdfs://localhost/flume/logtest/
agent1.sinks.sink1.hdfs.filePrefix = LogCreateTest
# Number of seconds to wait before rolling current file (0 = never roll based on time interval)
agent1.sinks.sink1.hdfs.rollInterval = 600
# File size to trigger roll, in bytes (0: never roll based on file size) 
agent1.sinks.sink1.hdfs.rollSize = 0
#Number of events written to file before it rolled (0 = never roll based on number of events) 
agent1.sinks.sink1.hdfs.rollCount = 10000
# number of events written to file before it flushed to HDFS 
agent1.sinks.sink1.hdfs.batchSize = 10000 
agent1.sinks.sink1.hdfs.txnEventMax = 40000
# -- Compression codec. one of following : gzip, bzip2, lzo, snappy
# hdfs.codeC = gzip
#format: currently SequenceFile, DataStream or CompressedStream
#(1)DataStream will not compress output file and please don't set codeC
#(2)CompressedStream requires set hdfs.codeC with an available codeC
agent1.sinks.sink1.hdfs.fileType = DataStream 
agent1.sinks.sink1.hdfs.maxOpenFiles=50
# -- "Text" or "Writable"
#hdfs.writeFormat
agent1.sinks.sink1.hdfs.appendTimeout = 10000
agent1.sinks.sink1.hdfs.callTimeout = 10000
# Number of threads per HDFS sink for HDFS IO ops (open, write, etc.)
agent1.sinks.sink1.hdfs.threadsPoolSize=100 
# Number of threads per HDFS sink for scheduling timed file rolling
agent1.sinks.sink1.hdfs.rollTimerPoolSize = 1 
# hdfs.kerberosPrin--cipal Kerberos user principal for accessing secure HDFS
# hdfs.kerberosKey--tab Kerberos keytab for accessing secure HDFS
# hdfs.round false Should the timestamp be rounded down (if true, affects all time based escape sequences except %t)
# hdfs.roundValue1 Rounded down to the highest multiple of this (in the unit configured using
# hdfs.roundUnit), less than current time.
# hdfs.roundUnit second The unit of the round down value - second, minute or hour.
# serializer TEXT Other possible options include AVRO_EVENT or the fully-qualified class name of an implementation of the EventSerializer.Builder interface.
# serializer.*


# Use a channel which buffers events to a file
# -- The component type name, needs to be FILE.
agent1.channels.channel1.type = FILE 
# checkpointDir ~/.flume/file-channel/checkpoint The directory where checkpoint file will be stored
# dataDirs ~/.flume/file-channel/data The directory where log files will be stored
# The maximum size of transaction supported by the channel
agent1.channels.channel1.transactionCapacity = 1000000 
# Amount of time (in millis) between checkpoints
agent1.channels.channel1.checkpointInterval 30000
# Max size (in bytes) of a single log file 
agent1.channels.channel1.maxFileSize = 2146435071
# Maximum capacity of the channel 
agent1.channels.channel1.capacity 10000000 
#keep-alive 3 Amount of time (in sec) to wait for a put operation
#write-timeout 3 Amount of time (in sec) to wait for a write operation

# Bind the source and sink to the channel
agent1.sources.source1.channels = channel1
agent1.sinks.sink1.channel = channel1
于 2013-01-04T11:16:23.473 回答
0

因为 /var/log/apache2/access.log 不够大，无法让flume打印文件的行所以只需尝试以下操作，您可以在控制台中找到输出

for i in {1..100}; do echo "tail log test$i" >> var/log/apache2/access.log;done
于 2013-05-28T02:26:39.897 回答
0

两个可能的原因：首先，一旦您发出命令并且源启动，接收器必须自己注册并启动。我在您显示的日志中找不到这两行。我希望你没有错过它。通常它应该看起来像这样：

apache@hadoop:/hadoop/projects/apache-flume-1.4.0-SNAPSHOT-bin$ bin/flume-ng agent -n agent1 -c /conf -f conf/agent1.conf
Info: Including Hadoop libraries found via (/hadoop/projects/hadoop-1.0.4/bin/hadoop) for HDFS access
Warning: $HADOOP_HOME is deprecated.

Warning: $HADOOP_HOME is deprecated.

Info: Excluding /hadoop/projects/hadoop-1.0.4/libexec/../lib/slf4j-api-1.4.3.jar from classpath
Info: Excluding /hadoop/projects/hadoop-1.0.4/libexec/../lib/slf4j-log4j12-1.4.3.jar from classpath
+ exec /usr/lib/jvm/java-7-oracle/bin/java -Xmx20m -cp '/conf:/hadoop/projects/apache-flume-1.4.0-SNAPSHOT-bin/lib/*:/hadoop/projects/hadoop-1.0.4/libexec/../conf:/usr/lib/jvm/java-7-oracle/lib/tools.jar:/hadoop/projects/hadoop-1.0.4/libexec/..:/hadoop/projects/hadoop-1.0.4/libexec/../hadoop-core-1.0.4.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/asm-3.2.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/aspectjrt-1.6.5.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/aspectjtools-1.6.5.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-beanutils-1.7.0.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-beanutils-core-1.8.0.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-cli-1.2.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-codec-1.4.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-collections-3.2.1.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-configuration-1.6.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-daemon-1.0.1.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-digester-1.8.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-el-1.0.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-httpclient-3.0.1.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-io-2.1.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-lang-2.4.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-logging-1.1.1.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-logging-api-1.0.4.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-math-2.1.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/commons-net-1.4.1.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/core-3.1.1.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/guava-13.0.1.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/hadoop-capacity-scheduler-1.0.4.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/hadoop-fairscheduler-1.0.4.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/hadoop-thriftfs-1.0.4.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/hsqldb-1.8.0.10.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/jackson-core-asl-1.8.8.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/jackson-mapper-asl-1.8.8.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/jasper-compiler-5.5.12.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/jasper-runtime-5.5.12.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/jdeb-0.8.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/jersey-core-1.8.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/jersey-json-1.8.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/jersey-server-1.8.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/jets3t-0.6.1.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/jetty-6.1.26.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/jetty-util-6.1.26.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/jsch-0.1.42.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/junit-4.5.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/kfs-0.2.2.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/log4j-1.2.15.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/mockito-all-1.8.5.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/oro-2.0.8.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/protobuf-java-2.3.0.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/servlet-api-2.5-20081211.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/xmlenc-0.52.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/zookeeper-3.4.3.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/jsp-2.1/jsp-2.1.jar:/hadoop/projects/hadoop-1.0.4/libexec/../lib/jsp-2.1/jsp-api-2.1.jar' -Djava.library.path=:/hadoop/projects/hadoop-1.0.4/libexec/../lib/native/Linux-amd64-64 org.apache.flume.node.Application -n agent1 -f conf/agent1.conf
12/12/15 02:55:29 INFO node.PollingPropertiesFileConfigurationProvider: Configuration provider starting
12/12/15 02:55:29 INFO node.PollingPropertiesFileConfigurationProvider: Reloading configuration file:conf/agent1.conf
12/12/15 02:55:29 INFO conf.FlumeConfiguration: Processing:HDFS
12/12/15 02:55:29 INFO conf.FlumeConfiguration: Processing:HDFS
12/12/15 02:55:29 INFO conf.FlumeConfiguration: Processing:HDFS
12/12/15 02:55:29 INFO conf.FlumeConfiguration: Processing:HDFS
12/12/15 02:55:29 INFO conf.FlumeConfiguration: Added sinks: HDFS Agent: agent1
12/12/15 02:55:29 INFO conf.FlumeConfiguration: Post-validation flume configuration contains configuration  for agents: [agent1]
12/12/15 02:55:29 INFO node.AbstractConfigurationProvider: Creating channels
12/12/15 02:55:29 INFO channel.DefaultChannelFactory: Creating instance of channel MemoryChannel-2 type memory
12/12/15 02:55:29 INFO node.AbstractConfigurationProvider: Created channel MemoryChannel-2
12/12/15 02:55:29 INFO source.DefaultSourceFactory: Creating instance of source tail, type exec
12/12/15 02:55:29 INFO sink.DefaultSinkFactory: Creating instance of sink: HDFS, type: hdfs
12/12/15 02:55:30 INFO hdfs.HDFSEventSink: Hadoop Security enabled: false
12/12/15 02:55:30 INFO node.Application: Starting new configuration:{ sourceRunners:{tail=EventDrivenSourceRunner: { source:org.apache.flume.source.ExecSource{name:tail,state:IDLE} }} sinkRunners:{HDFS=SinkRunner: { policy:org.apache.flume.sink.DefaultSinkProcessor@137efe53 counterGroup:{ name:null counters:{} } }} channels:{MemoryChannel-2=org.apache.flume.channel.MemoryChannel{name: MemoryChannel-2}} }
12/12/15 02:55:30 INFO node.Application: Starting Channel MemoryChannel-2
12/12/15 02:55:30 INFO instrumentation.MonitoredCounterGroup: Monitoried counter group for type: CHANNEL, name: MemoryChannel-2, registered successfully.
12/12/15 02:55:30 INFO instrumentation.MonitoredCounterGroup: Component type: CHANNEL, name: MemoryChannel-2 started
12/12/15 02:55:30 INFO node.Application: Starting Sink HDFS
12/12/15 02:55:30 INFO node.Application: Starting Source tail
12/12/15 02:55:30 INFO source.ExecSource: Exec source starting with command:tail -F /var/log/apache2/access.log.1
12/12/15 02:55:30 INFO instrumentation.MonitoredCounterGroup: Monitoried counter group for type: SINK, name: HDFS, registered successfully.
12/12/15 02:55:30 INFO instrumentation.MonitoredCounterGroup: Component type: SINK, name: HDFS started

请参阅最后 2 行。

其次，代理不会推送任何数据，直到文件中有新内容，这里是“/var/log/apache2/access.log”。手动将某些内容复制到文件中并重新启动 apache 并执行某些操作，然后检查 /hdfs/flume 目录的内容。

于 2012-12-14T21:46:58.470 回答
0

如果您使用 Flume 1.7.0，我想您可以使用taildir 源

以下是我在项目中使用的：

a1.sources.r1.type = TAILDIR
a1.sources.r1.positionFile = /xxx/env/flume/taildir_position.json
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /xxx/logs/file_name.*.log
a1.sources.r1.headers.f1.headerKey1 = yyy
a1.sources.r1.fileHeader = true
于 2017-04-22T15:37:28.107 回答