数据从kafka到hive ╰半橙微兮° 2022-10-02 12:54 104阅读 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
还没有评论,来说两句吧...