chukwa在数据收集处理方面的应用

迷南。 2022-03-20 15:54 394阅读 0赞

Chukwa 简介

什么是 Chukwa，简单的说它是一个数据收集系统，它可以将各种各样类型的数据收集成适合 Hadoop 处理的文件保存在 HDFS 中供 Hadoop 进行各种 MapReduce 操作。Chukwa 本身也提供了很多内置的功能，帮助我们进行数据的收集和整理。Chukwa 应用场景介绍  
为了更加简单直观的展示 Chukwa，我们先来看一个假设的场景。假设我们有一个规模很大 ( 牵扯到 Hadoop 的总是很大。。。。) 的网站，网站每天产生数量庞大的日志文件，要收集，分析这些日志文件可不是件容易的事情，读者可能会想了，做这种事情 Hadoop 挺合适的，很多大型网站都在用，那么问题来了，分散在各个节点的数据怎么收集，收集到的数据如果有重复数据怎么处理，如何与 Hadoop 集成。如果自己编写代码完成这个过程，一来需要花费不小的精力，二来不可避免的会引入 Bug。这里就是我们 Chukwa 发挥作用的时候了，Chukwa 是一个开源的软件，有很多聪明的开发者在贡献着自己的智慧。它可以帮助我们在各个节点实时监控日志文件的变化，增量的将文件内容写入 HDFS，同时还可以将数据去除重复，排序等，这时 Hadoop 从 HDFS 中拿到的文件已经是 SequenceFile 了。无需任何转换过程，中间繁杂的过程都由 Chukwa 帮我们完成了。是不是很省心呢。这里我们仅仅举了一个应用的例子，它还可以帮我们监控来自 Socket 的数据，甚至定时执行我们指定的命令获取输出数据，等等，具体的可以参看 Chukwa 官方文档。如果这些还不够，我们还可以自己定义自己的适配器来完成更加高级的功能。稍后我们将看到如何定义自己的适配器来做自己想做的事情。怎么样，是不是有些心动了呢。

Chukwa 的架构设计

前面我们简单说了他的一些用途，讲的比较概括，大家不一定能明白，它到底是怎么帮助我们完成功能的，下来我们就先从他的架构设计角度来看看。我们依然先看一张图。

**图 1. 架构示意图**  
![图 1. 架构示意图][1.]

我们还是以刚才提到的日志的例子来看。

这里我们先介绍几个新名词：

**Agent**

什么是 Agent，agent 是驻守在各个节点上的负责收集数据的程序。Agent 又由若干 adapter 组成。adapter 运行在 Agent 进程以内，执行实际收集数据的工作，而 Agent 则负责 adapter 的管理。

**Collector**

什么是 Collector，Collector 收集各个 Agent 传来的数据，并将这些数据写入 HDFS。

了解了这两个关键的名词之后，也许有读者已经在脑海中有了大致的数据流程图了。是的，它就是这么简单：数据被 Agent 收集，并传送到 Collector，由 Collector 写入 HDFS，然后由 Map-Reduce job 进行数据的预处理。

Chukwa 环境搭建与部署

这里我们介绍一下如何安装，部署，应用 Chukwa

1. 前提条件

1.   *  Linux 环境
     *  这里我们使用 Red Hat
     *  JDK 使用 1.6JDK
     *  系统中需要支持 SSH
     *  其他要求

2. 下载 chukwa

1.   *  这里给出的是其中一个镜像的下载地址 wget http://mirror.bjtu.edu.cn/apache/hadoop/chukwa/chukwa-0.4.0/chukwa-0.4.0.tar.gz 这里的链接是 0.4.0 版本的 chukwa. 其他版本可以从官方网站下载。官网地址：  
        [http://incubator.apache.org/chukwa/][http_incubator.apache.org_chukwa]

3. 下载 Hadoop

1.   *  Hadoop 的下载，安装不是本文重点，这里省略。  
        当前的 0.20.2 版是比较稳定的版本。开发中的 0.21.0 版，由于 jar 包结构上的改变以及配置的改变，与当前版本的 chukwa 的不兼容。所以建议使用 stable 版本的 Hadoop。

4. 安装

1.   *  tar -xzf chukwa-0.4.0.tar.gz
     *  tar -xzf hadoop-0.20.2.tar.gz
     *  解压之后，假设目录名称分别 chukwa-0.4.0 和 Hadoop-0.20.2

Hadoop 的配置

可以参考 Hadoop 官方网站的教程，限于篇幅，我们这里省略。

Chukwa 的配置

这里我们将以最简单的步骤完成 agent 和 collector 的配置，使读者可以快速的了解它。

**配置 Agent**

*  1.  编辑 $CHUKWA\_HOME/conf/chukwa-env.sh 文件，这里需要设置 JAVA\_HOME 注释掉 HADOOP\_HOME，HADOOP\_CONF\_DIR，因为 agent 仅仅是用来收集数据，所以不需要 HADOOP 的参与。注释掉 CHUKWA\_PID\_DIR，CHUKWA\_LOG\_DIR，如果不注释的话，那么他指定的位置是在 /tmp 临时目录下，这会导致，PID 和 LOG 文件被无故删除。会在后续的操作中导致异常。注释之后，系统会使用默认路径，默认会在 Chukwa 安装目录下创建 PID 和 LOG 文件。
    2.  编辑 $CHUKWA\_HOME/conf/collectors 文件，这里需要将 collectors 的地址写在这里 , 格式为 http://hostname:port/. 这里可以写多个 collector，每个占一行。Agent 通常会随机选择一个作为 collector 将数据发送给这个 collector, 如果当前的 collector 失败 , 则会继续选择下一个继续尝试。collector 是有 load balance 功能的，不会说所有的 Agent 都将数据写入到一个 Collector，从而导致失败。
    3.  编辑 $CHUKWA\_HOME/conf/initial\_adapters 文件，这里默认带的配置文件 initial\_adapters.template，修改名称为 initial\_adapters，里面默认带了几个 adapter 的例子。很容易理解。这个配置文件的名称，顾名思义是默认初始的 adapter，当 agent 启动的时候，这些 adapter 就会工作。

**配置 Collector**

*  1.  编辑 $CHUKWA\_HOME/conf/chukwa-env.sh 文件，修改 JAVA\_HOME，HADOOP\_HOME，HADOOP\_CONF\_DIR，指定为合适的值。同样的道理，我们需要注释掉 CHUKWA\_PID\_DIR，CHUKWA\_LOG\_DIR

**启动**

1.  启动 hadoop
    
    bin/start-all.sh
2.  启动 collector
    
    bin/chukwa collector
3.  启动 agent
    
    bin/chukwa agent

可以看到的结果

**清单 1. Agent 端日志片段**

<table> 
 <tbody> 
  <tr> 
   <td><pre>				
 2010-12-23 10:20:28,315 INFO Timer-1 ExecAdaptor - calling exec 
 2010-12-23 10:20:28,377 INFO Timer-1 ExecAdaptor - calling exec 
 2010-12-23 10:20:28,438 INFO Timer-1 ExecAdaptor - calling exec 
 2010-12-23 10:20:28,451 INFO HTTP post thread ChukwaHttpSender - 
                                                       collected 14 chunks for post_26923 
 2010-12-23 10:20:28,452 INFO HTTP post thread ChukwaHttpSender - 
                &gt;&gt;&gt;&gt;&gt;&gt; HTTP post_26923 to http://xi-pli:8080/ length = 17788 
 2010-12-23 10:20:28,459 INFO HTTP post thread ChukwaHttpSender - 
        &gt;&gt;&gt;&gt;&gt;&gt; HTTP Got success back from http://xi-pli:8080/chukwa; response length 924 
 2010-12-23 10:20:28,459 INFO HTTP post thread ChukwaHttpSender - 
        post_26923 sent 0 chunks, got back 14 acks 
 2010-12-23 10:20:28,500 INFO Timer-1 ExecAdaptor - calling exec 
</pre></td> 
  </tr> 
 </tbody> 
</table>

从这里我们能看到 `Timer-1 ExecAdaptor - calling exec`已经被定时执行，而且我们还能看到 Agent 将信息发送给了我们指定的 Collector:

**清单 2. Collector 端日志**

<table> 
 <tbody> 
  <tr> 
   <td><pre>				
 2010-12-23 10:30:22,207 INFO Timer-4 SeqFileWriter -
     rotating sink file /chukwa/logs/201023102522181_xipli_15db999712d1106ead87ffe.chukwa 
 2010-12-23 10:30:22,784 INFO Timer-1 root - 
         stats:ServletCollector,numberHTTPConnection:15,numberchunks:1110 
 2010-12-23 10:30:23,220 INFO Timer-3 SeqFileWriter - 
         stat:datacollection.writer.hdfs dataSize=797670 dataRate=26587 
</pre></td> 
  </tr> 
 </tbody> 
</table>

从日志我们可以看到 Collector 已经通过 writer 将收集到的数据写入了文件 /chukwa/logs/201023102522181\_xipli\_15db999712d1106ead87ffe.chukwa

那么怎么知道数据已经被写入 HDFS 了呢。我们可以通过执行 hadoop 的命令来查看 HDFS

**清单 3. 察看日志**

<table> 
 <tbody> 
  <tr> 
   <td><pre>				
 bin/hadoop fs -ls /chukwa/logs 
</pre></td> 
  </tr> 
 </tbody> 
</table>

如果不出意外，我们已经能看到数据已经被写入了 HDFS。

**清单 4. 检查文件**

<table> 
 <tbody> 
  <tr> 
   <td><pre>				
 Found 205 items 
 -rw-r--r--   3 hadoop supergroup     
 676395 2010-12-22 17:12 /chukwa/logs/201022171225851_xipli_18f1ec1212d0d4c13067ffa.done 
 -rw-r--r--   3 hadoop supergroup 6046366 2010-12-22 17:17 
 /chukwa/logs/201022171725877_xipli_18f1ec1212d0d4c13067ff8.chukwa 
 -rw-r--r--   3 hadoop supergroup    
8352420 2010-12-22 17:32 /chukwa/logs/201022173249756_xipli_1f33c45712d0d6c7cdd8000.done 
</pre></td> 
  </tr> 
 </tbody> 
</table>

如果想要将 agent 部署到多个节点去，只需要在其他节点上也做相应的配置即可。但是随着节点数目的增多，我们会发现管理 agent 的难度越来越高。这时我们可以使用单一的节点来批量管理所有的 Agents。

第一个我们需要编辑的文件是 conf 目录下的 agents 文件，默认情况下，它是一个模板文件，名称为 agents.template，我们需要把它改名为 agents。将 agent 的 hostname/IP 逐个添加进这个文件中，每行一个节点。这时通过调用 bin/start-agents.sh 和 bin/stop-agents.sh 可以批量的管理 agents 的启动和关闭。如果有遇到不能正常关闭 agent 的情况，那么一个可用的临时解决方法是将各个节点的 chukwa 脚本中的 kill -1 修改为 kill -9。Collector 也是类似的控制方式。

基本命令介绍

在 bin 目录下有很多的可执行文件。我们这里只关注以下一些文件：

**chukwa**

chukwa 命令的帮助信息如下

**清单 5. 基本命令**

<table> 
 <tbody> 
  <tr> 
   <td><pre>				
 Usage: chukwa [--config confdir] COMMAND 
 where COMMAND is one of: 
  agent         run a Chukwa Agent 
  archive       run the Archive Manager 
  collector     run a Chukwa Collector 
  demux         run the Demux Manager 
  dp            run the Post Demux data processors 
  hicc          run a HICC Webserver 
  droll         run a daily rolling job (deprecated) 
  hroll         run a hourly rolling job (deprecated) 
  version       print the version 
 Utilities: 
  backfill      run a back fill data loader utility 
  dumpArchive   view an archive file 
  dumpRecord    view a record file 
  tail          start tailing a file 
 Most command print help when invoked w/o parameters. 
</pre></td> 
  </tr> 
 </tbody> 
</table>

各参数的功能如下：

bin/chukwa agent 启动本地 agent

bin/chukwa agent stop 关闭本地 agent

bin/chukwa collector 启动本地 collector

bin/chukwa collector stop 关闭本地 collector

bin/chukwa archive 定时运行 archive，将文件整理成 Sequence File. 并且会去除重复内容。

bin/chukwa archive stop 停止运行 archive  
bin/chukwa demux 启动 demux 管理器，相当于启动了一个 M/R Job. 默认情况是 TsProcessor. 我们也可以自己定义自己的数据处理模块，稍后提到。

bin/chukwa demux stop 停止 demux 管理器

bin/chukwa dp 启动 demux post processor 用于定时排序，归并文件，剔除冗余数据。

bin/chukwa dp stop 停止 dp 运行

bin/chukwa hicc 这是一个有意思的东西，它类似一个 portal，将数据以图形的方式展现出来。但是在当前的 0.4.0 版本还有很多问题，读者要是有兴趣，可以试一下开发中的 0.5.0 版本。

后面的命令比较简单，根据提示也能正确运行，不做详细叙述。

**slaves.sh**

slaves.sh 命令 , 很有用，尤其是当你有很多节点的时候，比如有 50 个节点，想要给每个节点下创建一个目录 abc 怎么办呢。如果一个一个去机器上创建，那就太繁琐了。幸好，有它可以帮我们，`bin/slaves.sh mkdir /home/hadoop/abc`. 它就会帮我们在各个节点上创建对应的目录。

**start-agents.sh**

该命令会启动所有注册在 agents 文件中的 Agent

**start-collectors.sh**

该命令会启动所有注册在 collectors 文件中的 Collector

**stop-agents.sh**

该命令会停止所有注册在 agents 文件中的 Agent

**stop-collectors.sh**

该命令会停止所有注册在 collectors 文件中的 Collector

**start-data-processors.sh**

该命令是以下三个命令的组合：

bin/chukwa archive

bin/demux

bin/dp

他会将这三个命令依次启动，不用自己一个一个启动。

**stop-data-processors.sh**

依次停止 archive/demux/dp 三个服务

Agent 端命令：

当 Agent 启动之后，我们还可以动态控制 Agent 中的 adapter。当 Agent 启动之后，每个 Agent 都会启动一个 telnet 服务，用于单独控制 Agent。默认端口是 9093。

当运行了命令 telnet localhost 9093 之后，会进入 telnet 控制台，显示如下：

**清单 6. Agent 控制台信息**

<table> 
 <tbody> 
  <tr> 
   <td><pre>				
 Trying ::1... 
 Connected to localhost. 
 Escape character is '^]'. 
</pre></td> 
  </tr> 
 </tbody> 
</table>

输入回车，并输入 help 命令，会显示详细的命令帮助信息

**清单 7. Agent 控制台信息**

<table> 
 <tbody> 
  <tr> 
   <td><pre>				
 you're talking to the Chukwa agent.  Commands available: 
 add [adaptorname] [args] [offset] -- start an adaptor 
 shutdown [adaptornumber]  -- graceful stop 
 stop [adaptornumber]  -- abrupt stop 
 list -- list running adaptors 
 close -- close this connection 
 stopagent -- stop the whole agent process 
 stopall -- stop all adaptors 
 reloadCollectors -- reload the list of collectors 
 help -- print this message 
         Command names are case-blind. 
</pre></td> 
  </tr> 
 </tbody> 
</table>

此时输入 list，则可以查看到当前正在运行的所有的 adapter。显示如下：

**清单 8. Agent 控制台信息**

<table> 
 <tbody> 
  <tr> 
   <td><pre>				
 adaptor_8567b8b00a5dc746ccd5b8d179873db1)  
     org.apache.hadoop.chukwa.datacollection.adaptor.ExecAdaptor 
      Top 60 /usr/bin/top -b -n 1 -c 505728 
 adaptor_e69999b07d7023e6ba08060c85bd9ad7)  
           org.apache.hadoop.chukwa.datacollection.adaptor.ExecAdaptor 
           Df 60 /bin/df -l 353792 
</pre></td> 
  </tr> 
 </tbody> 
</table>

其他命令都比较浅显易懂，我们下来仅仅看看 add 命令。以下面一个例子来看：

**清单 9. Agent 添加 adapter**

<table> 
 <tbody> 
  <tr> 
   <td><pre>				
 add filetailer.LWFTAdaptor dataType /foo/bar 0 
</pre></td> 
  </tr> 
 </tbody> 
</table>

该命令将 LWFTAdaptor 这个 adaptor 加入到 Agent 中并运行，数据类型是 dataType 类型，这个类型将与后续的 demux 服务一起配合工作，/foo/barbar 是 adaptor 的参数，最后的数字 0 则表示数据的偏移量。目前 chukwa0.4.0 支持的 adaptor 大致分以下几种类型：检测文件 / 目录变化的，监测 UDP 数据的，执行特定 shell 脚本的。具体可以参考 chukwa 官方文档，有较详尽的描述。需要指出的是在 0.5 版本中，adaptor 被更加的强化和规范化。有兴趣的读者可以看看。

`如果要从`telnet`控制台退出，则可以通过输入`close`退出`。

内部数据处理时序介绍

**图 2. 内部数据处理时序**  
![图 2. 内部数据处理时序][2.]

1.  Collector 将 Agent 发送的数据写入到 logs/\*.chukwa 文件，直到文件大小达到 64M 或者达到一定的时间间隔之后，Collector 会将 \*.chukwa 文件重命名为 \*.done 文件表明文件写入结束。
2.  DemuxManager 每隔 20 秒中检查一次 \*.done 文件，如果文件存在，则将文件移动到 demuxProcessing/mrInput 文件夹下。而 Demux MapReduce job 将会以此文件夹为输入进行 map/reduce 操作。如果此操作成功 ( 可重试 3 次 ) 则会将 map/reduce 的输出结果从 demuxProcessing/mrOutput 文件夹下归档，并移动到 dataSinkArchives/\[yyyyMMdd\]/\*/\*.done。同时会将文件输出到 postProcess 目录下。否则如果操作失败，则会将输出结果移动到 dataSinkArchives/InError/\[yyyyMMdd\]/\*/\*.done。
3.  PostProcessManager 每隔几分钟执行一次，负责将 postProcess 目录下的文件进行合并，去重，排序等操作。运行完毕之后，将数据会写入到 repos 目录。目录下会按照 cluster name,data type 等分目录存放。
4.  在以上的操作中，Demux 将是我们下来要关注的内容，很多的数据处理都会在这里进行。我们也可以自己定义自己的数据类型的 Demux processor。

如何支持新的数据类型

在 Chukwa 内置 adaptor 中，可以收集来自于文件和 socket 的数据。我们接下来自己定义一种新的 adaptor 来收集来自 jms 的数据。通常在自定义 adaptor 需要继承 org.apache.hadoop.chukwa.datacollection.adaptor.AbstractAdaptor。我们来看以下的代码：

**清单 10. adapter 示例**

<table> 
 <tbody> 
  <tr> 
   <td><pre>				
import org.apache.hadoop.chukwa.datacollection.adaptor.AbstractAdaptor; 
 import org.apache.hadoop.chukwa.datacollection.adaptor.AdaptorException; 
 import org.apache.hadoop.chukwa.datacollection.adaptor.AdaptorShutdownPolicy;

public class JMSAdaptor  extends AbstractAdaptor{

@Override 
	 public String getCurrentStatus() { 
		 //  TODO Auto-generated method stub 
		 return
				null; 
	 }

@Override 
	 public long shutdown(AdaptorShutdownPolicy shutdownPolicy) 
			  throws AdaptorException { 
		 //  TODO Auto-generated method stub 
		  return 0; 
	 }

@Override 
	 public void start(long offset) throws AdaptorException { 
		 //  TODO Auto-generated method stub 
		
	 }

@Override 
	  public String parseArgs(String s) { 
		 //  TODO Auto-generated method stub 
		  return null; 
	 }

} 
</pre></td> 
  </tr> 
 </tbody> 
</table>

`有 4`个方法需要实现 `：`

1.  parseArgs``将用户的输入解析成程序的入参
2.  start``启动 adaptor
3.  shutdown``关闭 adaptor
4.  getCurrentStatus`获取`adaptor 的当前状态 `，`会被 AdaptorManager 调用 `，`用于定时报告 adaptor 的状态

如何获取数据，完全取决于用户自己的实现了。比如我们以 JMS 为例，我们可以实现 JMS Consumer，然后在 start 方法中启动。Consumer 需要的参数，则通过 parseArgs 获得。当用户在 agent 的 telnet 控制台中将 adaptor 停止时，shutdown 方法就会被调用，从而实现程序的关闭。

当实现了一个自己的 adaptor 之后，我们需要将 adaptor 注册到 agent。注册方法参见前几章节中的基本命令的介绍。

如何自定义数据处理模块

在上一节中，我们定义了自己的 adaptor，用于收集自定义类型的数据。接下来我们会介绍 Collector 是如何收集处理数据的。假设收集的数据类型为 foo 类型。那么我们只需要在 Collector 端的配置文件 chukwa-demux-conf.xml 中增加如下片段即可：

**清单 11. 自定义数据处理模块添加方法**

<table> 
 <tbody> 
  <tr> 
   <td><pre>				
 &lt;property&gt; 
  &lt;name&gt;foo&lt;/name&gt; 
  &lt;value&gt;org.apache.hadoop.chukwa.extraction.demux.processor.mapper.FooProcessor&lt;/value&gt; 
  &lt;description&gt;Parser class for Foo&lt;/description&gt; 
 &lt;/property&gt;

</pre></td> 
  </tr> 
 </tbody> 
</table>

其中定义的 FooProcessor 是我们新加的一个数据处理模块，是 map/reduce 中的 map 部分。该类可以放到任意目录下。注册完 map，我们还需要考虑 reduce，reduce 并不需要在配置文件中做什么配置。看以下代码：

**清单 11. 自定义数据处理模块代码示例**

<table> 
 <tbody> 
  <tr> 
   <td><pre>				
import org.apache.hadoop.chukwa.extraction.demux.processor.mapper.AbstractProcessor;

import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecord; 
  import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecordKey; 
  import org.apache.hadoop.mapred.OutputCollector; 
  import org.apache.hadoop.mapred.Reporter;

public 
				 class FooProcessor  extends AbstractProcessor{

@Override 
	  protected 
				 void parse(String recordEntry, 
			 OutputCollector&lt;ChukwaRecordKey, ChukwaRecord&gt; output, 
			 Reporter reporter)  throws Throwable { 
		 ChukwaRecordKey key1 =  new ChukwaRecordKey(); 
		 String reduceType = "Foo"; 
		 key1.setReduceType(reduceType); 
		 ChukwaRecord record =  new ChukwaRecord(); 
		 .........//map work 
		 output.collect(key1, record); 
	 }

} 
</pre></td> 
  </tr> 
 </tbody> 
</table>

我们可以看到这就是一个很普通的 map/reduce 类，我们可以在其中完成我们自己想做的事情，唯一有些特别的地方是 reduce 需要通过 key 来设置，key1.setReduceType(“reduceType”)。而这里有一点限制，我们自己定义的 reduce 单元必须在 org.apache.hadoop.chukwa.extraction.demux.processor 包下，chukwa 会在根据 reduceType 在这个包下找到相应的 reduce，通过 class.forName 来创建 reduce 实例。reduce 类需要实现 org.apache.hadoop.chukwa.extraction.demux.processor.reducer.ReduceProcessor 接口。接口很简单，有两个方法需要实现，一个是 getDataType 方法，返回该 reduce 处理的数据的类型，返回值为字符串。另一个方法则是很普通的 reduce 函数，这里略去不讲。做完了这些，我们的整个数据流程就完整了，包含了数据的收集，数据处理，以及自定义模块的定义，注册。

结束语

通过以上的介绍，想必大家对 Chukwa 的基本工作原理，如何定制，部署都有了一定的了解。由于它设计理念很简单，结构清晰易懂，而且是开源的产品，我们可以在它的基础之上构建自己更加强大的功能。这里要强调的是 Chukwa 是一个还处于 apache 的孵化器中的开源项目，正处于快速演化的过程中，现在的各个版本之间的差异还是比较明显的，而且由于演化速度很快，文档相对滞后，所以在使用中可能会遇到一些奇怪的问题，这些问题都可以发送到 Chukwa 的邮件组中讨论。最后也希望本文能给大家在工作带来帮助。

[1.]: /images/20220320/b53ade4f830146fb97bf2566e525b113.png
[http_incubator.apache.org_chukwa]: http://incubator.apache.org/chukwa/
[2.]: /images/20220320/13ad0df37922497e87e9a8984d281462.png