Camus介绍

Camus

概念

Linkedin/Camus是一个MapReduce作业，从Kafka拉取指定topic到HDFS，供批量/离线数据处理、分析。

其中，Camus sweeper组件提供去重功能,可以对Kafka中由于“at least once”机制而存在重复数据的topic进行去重。

目前Camus已经停止更新了，新项目为gobblin。

特点

Camus有以下几个特点：

  • 自动发现topic
  • Avro schema管理
  • 按时间分区

Camus job

image

每个camus job分为三个阶段：

  • Setup Stage：从Kafka和Zookeeper获取可用的topics，partitions，offset等元信息（Metadata）
  • Hadoop Job Stage：开启多个ETL map task执行topic的数据获取，并写到HDFS。每个task的流程如下：
    • Pulling the data：每个task根据Setup Stage获取到的元数据，消费Kafka的数据，在Schema注册服务查找Avro模式来反序列化。每个task都生成4个文件：Avro data files，Count statistics files，Updated offset files，Error files
    • Committing the data：当一个task完成时，其拉取的数据都被提交到基于时间戳的output目录。
    • Producing Audit Counts：执行成功的tasks会向HDFS写audit counts
    • Storing the offset：每个partition都有offset，这些offset信息存储在HDFS中
  • Cleanup Stage：hadoop job完成后，收集HDFS中的audit counts并进行聚合，然后提交聚合结果到kafka

操作

camus关键在于实现两个类，Decoder和Writer，使其满足我们的系统业务，并且稍加配置就可以嵌入到自己的系统中。

MessageDecoder

MessageDecoder对kafka消息进行解码。

Camus自带了KafkaAvroMessageDecoder，JsonStringMessageDecoder等。也可以通过继承MessageDecoder实现符合自己业务的Decoder。通过camus.message.decoder.class配置。

RecordWriterProvider

RecordWriterProvider在数据写到HDFS之前进行编码。

Camus自带AvroRecordWriterProvider，StringRecordWriterProvider等。也可以通过实现RecordWriterProvider实现自己的写入方法。并通过
etl.record.writer.provider.class来配置

配置

配置项 解释 是否必须 默认值
Schema配置
schema.registry.url schema registry地址 是
max.schemas.per.subject subject下最大schema数量 否 1000
Kafka配置
kafka.brokers kafka broker地址List 是
kafka.max.pull.hrs 从每个分区拉取数据的最长历史小时数（相对于第一条未消费的消息的时间戳），-1表示没限制 否 -1
kafka.max.historical.days Events的时间戳超过将被丢弃，-1表示没限制 否 -1
kafka.max.pull.minutes.per.task MapReduce任务拉取消息的最大时间。超时就会停止，下一个camus job会继续从停止的offset消费 否 -1
kafka.blacklist.topics 黑名单里的topic不会被消费 否 null
kafka.whitelist.topics 白名单里的topic会被消费 否 null
etl.output.record.delimiter 消息分隔符 否 “\n”
Camus Job配置
camus.job.name camus job名 否 Camus Job
camus.message.decoder.class 解码器 否 AvroMessageDecoder
etl.record.writer.provider.class 编码器 否 AvroRecordWriterProvider
etl.partitioner.class Camus输出分区，默认按小时 否 DefaultPartitioner
camus.work.allocator.class Class for creating input splits from ETL requests 否 BaseAllocator
etl.destination.path 文件最终输出的文件夹，Top-level output directory, sub-directories will be dynamically created for each topic pulled. 是
etl.execution.base.path Camus运行信息临时存放的HDFS目录，offsets, error logs and count files等 是
etl.execution.history.path 已完成的Job的运行信息存放的HDFS目录，通常是 etl.execution.base.path的子目录 是
hdfs.default.classpath.dir All files in this directory will be added to the distributed cache and placed on the classpath for Hadoop tasks. 否 null
mapred.map.tasks 最大hadoop任务数, 每个任务可以拉取多个topic partitions 否 30

配置讲解

image
设ts是在runTime-kafka.max.historical之内的还未消费的第一条消息的时间戳。图中当前运行时间runTime为2015年12月5日0点，ts为2015年12月2日0点到5日0点之间还未消费的第一条消息的时间戳，即绿色区域的起点。

如果消息的时间戳在ts,ts + kafka.max.pull.hrs之间）,将被camus任务消费，并且假设在kafka.max.pull.minutes.per.task内任务很快完成。显然图中绿色区域的都会被消费。

配置场景应用

kafka.max.historical.days:可以用于忽略过老的消息
kafka.max.pull.hrs:将此值设置得比camus job的执行频率小，可以防止重复任务

运行

1
2
3
4
5
6
mvn clean package
hadoop jar camus-example-<version>-SNAPSHOT.jar com.linkedin.camus.etl.kafka.CamusJob  
 -D <property=value>   use value for given property  
 -P <arg>              external properties filename  
 -p <arg>              properties filename from the classpath

参考

http://docs.confluent.io/1.0/camus/docs/intro.html
https://github.com/confluentinc/camus
http://getindata.com/blog/post/time-related-configuration-settings-and-assumptions-in-camus/
http://blog.csdn.net/amghost/article/details/44258817
http://blog.csdn.net/amghost/article/details/44258841

本文标题:Camus介绍

文章作者:小懒

发布时间:2016-12-01, 19:22:00

最后更新:2016-12-02, 21:29:08

原始链接:https://fangyeqing.github.io/2016/12/01/Camus介绍/

许可协议: "署名-非商用-相同方式共享 4.0" 转载请保留原文链接及作者。

文章目录
  1. 1. 概念
    1. 1.1. 特点
    2. 1.2. Camus job
  2. 2. 操作
    1. 2.1. MessageDecoder
    2. 2.2. RecordWriterProvider
    3. 2.3. 配置
      1. 2.3.1. 配置讲解
      2. 2.3.2. 配置场景应用
    4. 2.4. 运行
  3. 3. 参考
|