数据从kafka到hive

╰半橙微兮° 2022-10-02 12:54 103阅读 0赞

## 背景 ##

公司的系统是一个对外提供服务的接口，每一次调用日志都需要保存到hive中，以便后期做数据分析。每天的调用量在亿级，日志数据量100G以上，在量级还没有这么大的时候，采取的办法比较原始：直接通过log4j打印到日志文件，然后通过抽数工具同步到hive中，每天凌晨同步前一天的数据。随着量级增大，日志文件越来越大，每天抽数就要抽好几个小时，而且偶尔还由于网络问题等原因失败。

## 方案 ##

日志数据不能直接发送给hive，这样耦合度太强了。既然说到去耦合，肯定是采用消息管道了，kafka由于其与大数据结合的紧密程度，成为不二选择。所以初步方案是先将日志发送到kafka，再通过其他工具从kafka读到hive表中，在遇到峰值时，即便kafka挂了，也不会影响接口服务。   
下一步就是如何将数据从kafka读到hive中，kafka的东家LinkedIn给出了解决方案：camus(https://github.com/linkedin/camus)和gobblin(https://github.com/linkedin/gobblin)。camus在2015年已经停止维护了，gobblin是后续产品，camus功能是是gobblin的一个子集，通过执行mapreduce任务实现从kafka读取数据到HDFS，而gobblin是一个通用的数据提取框架，可以将各种来源的数据同步到HDFS上，包括数据库、FTP、KAFKA等。因为只需要同步kafka数据，所以我们采用了实现相对简单的camus。在测试过程中，同步一个小时的数据（5G以上），大概需要2分钟左右，即便日志量翻10倍，也是可以接受的，当然，抽数时间也不会随数据量增大而线性增长。   
只差最后一步了，camus只能把数据读到HDFS，从HDFS到hive是通过shell脚本实现的，shell脚本执行load命令直接将数据搬到hive中。

## 实施 ##

下载camus代码后，直接用maven编译，生成的jar包在camus-example中。源码里面包含一个camus.properties的配置文件，这里说几个重要的配置项：

#数据目标路径，最终取到的数据在HDFS上的位置
    etl.destination.path=/user/username/topics
    #执行信息存放路径，最重要的是上次读取的kafka的offset
    etl.execution.base.path=/user/username/exec
    #消息解码类，camus读到的数据是byte[]格式的，可以在自定义类进行反序列化
    camus.message.decoder.class=com.linkedin.camus.etl.kafka.coders.SimpleMessageDecoder
    #要读取的topic
    kafka.whitelist.topics=
    #时区，这个很重要，因为数据存放是按日期的
    etl.default.timezone=Asia/Chongqing

将jar包和properties文件放在同一目录下，通过hadoop -jar camus-example-0.1.0-SNAPSHOT-shaded.jar -p camus.properties运行任务。   
camus任务执行完毕后再通过脚本将数据load到hive中，脚本内容如下：

date_string=$(date '+%Y/%m/%d/%H') partion=$(date '+%Y-%m-%d_%H')
    table_name= service_log_table
    filePath="/user/username/topics/hourly/"$date_string"/"
    hive<<EOF
    create table if not exists $table_name (
    mapJson STRING, 
    desc STRING,
    str1 STRING,
    str2 STRING,
    str3 STRING) 
    PARTITIONED BY (dt STRING)
    STORED AS TEXTFILE;
    load data inpath '$filePath' overwrite into table $table_name partition (dt='$partion');
    EOF

前面一篇讲到了将数据从kafka读到hdfs使用了开源工具camus，既然用到了开源的代码，免不了研究一下实现过程。所以这里分享一下阅读camus代码了解到的一些细节。

## 前置知识 ##

在讲camus之前，需要提一下hadoop的一些知识。

### 关于inputFormat ###

inputFormat类的原型如下：

public interface InputFormat<K, V> {
            InputSplit[] getSplits(JobConf job, int numSplits) throws IOException;
            RecordReader<K, V> getRecordReader(InputSplit split,JobConf job,Reporter reporter) throws IOException;
    }

hadoop job调用setInputFormat来设置InputFormat类，通过getSplits函数，hadoop可以将输入分割为InputSplit，每个map分配到InputSplit后，再通过getRecordReader获取reader，逐条读取输入文件中Key-Value。hadoop本身提供了一些InputFormat类的实现，如TextInputFormat，KeyValueTextInputFormat，SequenceFileInputFormat等；   
除了使用hadoop本身提供的这些类以外，hadoop允许自定义InputFormat类，只需要实现相应的getSplits函数和RecoderReader类即可。

## 关于OutputFormat 和 OutputCommitter ##

OutputFormat类用于写入记录，原型如下：

public abstract class OutputFormat<K, V> {
        //创建一个写入器
      public abstract RecordWriter<K, V> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException;
       // 检查结果输出的存储空间是否有效
      public abstract void checkOutputSpecs(JobContext context) throws IOException, InterruptedException;
       //创建一个任务提交器
      public abstract OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException;
    }

getRecordWriter和checkOutputSpecs都很好理解，前置就是将Mapper或Reducer传来的key-value写入到文件里面，写入的姿势由它来决定。而checkOutputSpecs一般就是用来检查输出规范，判断输出文件是否存在，如果已经存在，则抛出异常。   
OutputCommitter稍稍难理解一点，里面定义了这几个方法：

//Job开始被执行之前，框架会调用setupJob()为Job创建一个输出路径
    void  setupJob(JobContext jobContext);
    //如果Job成功完成，框架会调用commitJob()提交Job的输出
    void  commitJob(JobContext jobContext);
    //如果Job失败，框架会调用OutputCommitter.abortJob()撤销Job的输出
    void  abortJob(JobContext jobContext, JobStatus.State state);
    //Task执行前的准备工作，类似setupJob
    void  setupTask(TaskAttemptContext taskContext);
    //Task可能没有输出，也就不需要提交，通过needsTaskCommit()来判断
    boolean  needsTaskCommit(TaskAttemptContext taskContext);
    //Task成功执行，提交输出
    void  commitTask(TaskAttemptContext taskContext);
    //Task执行失败，清理
    void  abortTask(TaskAttemptContext taskContext);

一般是这样一个过程：Job执行的时候，Task的输出放到Output路径下的\_temporary目录的以TaskAttemptID命名的子目录中。只有当 Task成功了，相应的输出才会被提交到Output路径下。而只有当整个Job都成功了，才会在Output路径下放置\_SUCCESS文件。 \_SUCCESS文件的存在表明了Output路径下的输出信息是正确且完整的；而如果\_SUCCESS文件不存在，Output下的信息也依然是正确的 （这已经由commitTask保证了），但是不一定是完整的（可能只包含部分Reduce的输出）。

# camus的实现 #

camus重写了inputFormat和outputFormat类。

## 重写getSplits ##

camus是以map-reduce的方式执行的，实际上，它没有设置reduce任务，map完成后，直接将文件写入到了目标目录下。   
camus的主要工作都是在InputFormat和OutputFormat中完成的。前面说到过自定义InputFormat，camus正是通过自定义的EtlInputFormat类完成了读取任务的分割。   
EtlInputFormat的getSplits函数需要将抽数任务分割为几个独立的splits，而kafka的partition之间的消息没有顺序关系，正好符合这种条件。分割流程如下：   
1.获取topic下各个partition的leader信息，得到一个映射关系为LeaderInfo -> List 的map，也就是各个partition的leader连接信息，再加上leader所在的partition列表。关键核心代码如下：

//获取kafka集群TopicMetadata信息，这些信息包括各个partition的leaderId,replicas以及Isr，存在于每一个broker上，所以即便brokerList不完整或者有broker挂了也不会影响。
    List<TopicMetadata> topicMetadataList = getKafkaMetadata(context, new ArrayList<String>());
    //对于每一个topic下的每一个partition，执行操作
    for (TopicMetadata topicMetadata : topicMetadataList) {
        for (PartitionMetadata partitionMetadata : topicMetadata.partitionsMetadata()) {
            //为每个partition的leader创建LeaderInfo对象
            LeaderInfo leader =new LeaderInfo(new URI("tcp://" + partitionMetadata.leader().getConnectionString()),partitionMetadata.leader().id());
            //下面这一段，就是将leader相同的partition放在一起
            //最终得到Map<LeaderInfo,TopicAndPartition>
            if (offsetRequestInfo.containsKey(leader)) {
                    ArrayList<TopicAndPartition> topicAndPartitions = offsetRequestInfo.get(leader);
                    topicAndPartitions.add(new TopicAndPartition(topicMetadata.topic(),partitionMetadata.partitionId()));
                    offsetRequestInfo.put(leader, topicAndPartitions);
            } else {
                //和上面一样，只是在offsetRequestInfo中没有leader时有个新建List的操作
                .....
            }
        }
    }

2.根据LeaderInfo -> List的映射关系，向每个partition的leader请求earliest offset和lastest offset。kafka的每个broker上的log会定期删除，earliest offset就是leader上保存的最早的log偏移量，而lastest offset则是最新log偏移量。在这之后，camus还会读取上次执行信息，其中包含上次任务每个partition读取完成的offset，作为本次任务读取的起始点。上次执行执行信息保存在配置项etl.execution.base.path对应的路径下。根据每个partition的earliest offset、lastest offset、offset组装成List，即请求列表。   
CamusRequest是一个接口，主要信息如下：

//获取当前请求的topic
    public abstract String getTopic();
    //获取当前请求的URI
    public abstract URI getURI();
    //获取当前请求的partition
    public abstract int getPartition();
    //获取请求的起始offset
    public abstract long getOffset();
    public abstract long getEarliestOffset();
    //获取partition的最大偏移量
    public abstract long getLastOffset();
    public abstract long estimateDataSize();

有了这些信息，就可以开始分派request到各个mapper了。   
3.默认使用BaseAllocator分派request，其中的关键函数如下：

@Override
      public List<InputSplit> allocateWork(List<CamusRequest> requests, JobContext context) throws IOException {
        //获取配置项中mapred.map.tasks的map数目，一个mapper可以执行多个partition的抽数任务
        int numTasks = context.getConfiguration().getInt("mapred.map.tasks", 30);
        //按数据量由大到小给每个partition的request排序
        reverseSortRequests(requests);
        //初始化splists，这时候每个splits里面的request数目为0
        List<InputSplit> kafkaETLSplits = new ArrayList<InputSplit>();
        for (int i = 0; i < numTasks; i++) {
          if (requests.size() > 0) {
            kafkaETLSplits.add(new EtlSplit());
          }
        }
        //分配request到各个splits，算法很简单，当前哪个splits的数据量最少，就给那个splits添加request，因为是按由大到小排序的，所以任务比较均匀
        for (CamusRequest r : requests) {
          getSmallestMultiSplit(kafkaETLSplits).addRequest(r);
        }
        return kafkaETLSplits;
      }

allocateWork函数返回的就是List类型，也就是EtlInputFormat.getSplits的返回值。

## 从InputSplit中读取message ##

每个InputSplit中可能包含多个CamusRequest，每个CamusRequest是不同partition的读数请求，其中包含Topic，Partition，leader URI，offset，lastest offset等信息。前面说到过，InputFormat需要重写createRecordReader函数，也就是定义如何从split中读取key-value。   
EtlInputFormat.createRecordReader返回一个EtlRecordReader实例，由该实例读取kafka message发送给mapper。   
RecordReader需要重写函数boolean nextKeyValue()，在数据读完之前，该函数一直返回true。   
EtlRecordReader.nextKeyValue 核心代码(有删减)如下：

while (true) {
        //reader为KafkaReader类型，该类专门负责逐条读取kafka消息
        if (reader == null || !reader.hasNext()) {
            //从InputSplit中弹出request
            EtlRequest request = (EtlRequest) split.popRequest();
            //传给mapper的key，这里是类成员变量，后面直接由getCurrentKey()返回
            key.set(request.getTopic(), request.getLeaderId(), request.getPartition(), request.getOffset(),request.getOffset(), 0);
            //在reader为null，或者一个partition中的数据已经读完的情况下，新建reader，reader负责从kafka中读取数据
            reader =new KafkaReader(inputFormat, context, request, CamusJob.getKafkaTimeoutValue(mapperContext),CamusJob.getKafkaBufferSize(mapperContext));
            //创建消息decoder，由配置camus.message.decoder.class指定，从kafka中读取出的为byte[]类型，decoder类负责将字节转换为最终在hdfs上存放的格式
            decoder = createDecoder(request.getTopic());
        }
        while ((message = reader.getNext(key)) != null) {
            //用decoder解码消息，封装在CamusWrapper中
            CamusWrapper wrapper = getWrappedRecord(message);
             value = wrapper;
             return true;
        }

EtlRecordReader.getCurrentKey和EtlRecordReader.getCurrentValue直接返回nextKeyValue中的key，value。Mapper正是直接将这两个函数的返回值作为入参。至此，InputFormat的工作就做完了。Mapper收到key-value后什么也没做，直接调用context.write(key, val)写入，因为没有reducer，所以结果会直接由OutputFormat输出。

## OutputFormat输出message ##

EtlMultiOutputFormat重写了OutputFormat的getRecordWriter和getOutputCommitter函数。getRecordWriter返回一个EtlMultiOutputRecordWriter类的实例，它负责将Task的输出放到Output路径下，而由getOutputCommitter获取的EtlMultiOutputCommitter类实例在Task完成后将输出拷贝到最终目的地。   
EtlMultiOutputRecordWriter.write(EtlKey key, Object val)负责执行写入key-value工作，核心代码如下：

//key中包含的是关于kafka的信息，如当前消息的offset，topic，parition等
    //val中是具体的message，也就是前面从kafka 消息的bytes[]中获取的
    public void write(EtlKey key, Object val) throws IOException, InterruptedException {
        //当前message的key是否和前面message的一致，因为一个split可能包含不同paritition的数据，也就是一个mapper就可能处理不同partition
        if (!key.getTopic().equals(currentTopic)) {
            //如果topic不一致，则需要清除writer，重新创建新的topic的，之所有有很多writer，是因为topic下会有多个partition，不同partition写入到不同的文件中
            for (RecordWriter<IEtlKey, CamusWrapper> writer : dataWriters.values()) {
                writer.close(context);
            }
            dataWriters.clear();
            //更新当前topic
            currentTopic = key.getTopic();
        }
        //将当前的key的offset存入到 (topic+partition)-> offset 的map中，因为同一partition消息是顺序的，所以最后存入的肯定是offset最大的消息，committer会将这个map写入到执行记录目录中，下一次运行时以这个offset为起点
        committer.addCounts(key);
        CamusWrapper value = (CamusWrapper) val;
        //获取写入文件名
        String workingFileName =EtlMultiOutputFormat.getWorkingFileName(context, key);
        //如果现有的writer中不包含对应文件的，则新建一个
        if (!dataWriters.containsKey(workingFileName)) {
            dataWriters.put(workingFileName, getDataRecordWriter(context,workingFileName, value));
            log.info("Writing to data file: " + workingFileName);
        }
        //将key-value写入到Taks的输出目录下临时文件中
        dataWriters.get(workingFileName).write(key, value);

大部分的信息都在注释中写了，这个理解起来也简单，就是将key-value写到临时文件里，同时把每个key的offset按partition存放起来，在任务成功时写到文件里。   
EtlMultiOutputCommitter只实现了commitTask函数，该函数在任务（不是整个Job）完成后被调用。前面已经将输入写到输出目录下的临时文件里了，commitTask要做的就是将文件拷贝到配置etl.destination.path对应的路径下，拷贝完后再将 write函数中写入的(topic+partition)-> offset 键值对写入到etl.execution.base.path目录下，下次任务执行时，会从这个目录下读取offset，作为起始offset。如果某个Task失败，offset也不会写入，下次读取时还是会重新读一遍，而其他Task不受影响。   
还是贴一部分核心代码：

//获取配置etl.destination.path对应的目录，作为最终输出目录
    Path baseOutDir = EtlMultiOutputFormat.getDestinationPath(context);
    //这里fs是前面write函数写入的目录，也就是Task的output路径
    //列举其中的文件，数据文件的文件名都是以data开头的
    for (FileStatus f : fs.listStatus(workPath)) {
        String file = f.getPath().getName();
        if (file.startsWith("data")) {
            //数据文件名中包含topic，partition，offset，time等信息
            //根据文件名，消息计数等信息获得最终的输出路径和输出文件名
            //输出路径是以小时分割的，basePath/year/month/day/hour/filename
            String partitionedFile =
            getPartitionedPath(context, file, count.getEventCount(), count.getLastKey().getOffset());
            //如果路径不存在，创建路径
            Path dest = new Path(baseOutDir, partitionedFile);
            if (!fs.exists(dest.getParent())) {
                mkdirs(fs, dest.getParent());
            }
            //拷贝文件，从Task输出目录下拷贝到目标文件
            commitFile(context, f.getPath(), dest)；
        }
    }
    //后面还有一段写入offset的，只有写入数据成功后才会写入offset

转：[http://www.aboutyun.com/thread-20701-1-1.html][http_www.aboutyun.com_thread-20701-1-1.html]

[http_www.aboutyun.com_thread-20701-1-1.html]: http://www.aboutyun.com/thread-20701-1-1.html