日均 TB 级数据，携程支付统一日志框架

一、背景

支付中心作为携程集团公共部门，主要负责的业务包括交易、实名绑卡、账户、收单等，由于涉及到交易相关的资金流转以及用户实名认证，部分用户操作环节的中间数据应内控/审计要求需要长时间保存。当前研发应用多，日志量大、格式各异，对于日志的存储和使用产生较大的挑战，故支付数据与研发团队群策群力，共同开发了一套统一日志框架。

二、总体架构图

核心模块包括：日志生产、日志采集、日志解析，其中调用流程如下：

1）研发应用/服务接入基于 log4j2 扩展的统一日志组件，将日志抛送至 kafka。

2）周期性启动消费 kafka topic 的 camus job 将日志写入 hdfs。

3）T+1 启动 MR job 读取 camus 写入的 hdfs 内容并 load 到 hive 表。

三、日志生产-统一日志组件

支付研发基于 log4j2 自定义了多个 Appender，将应用日志以服务调用形式抛送至 kafka，并被 log_process_service 服务统一处理并提交至携程常用基础日志框架如：CLOG、CAT、ES，各应用无需关心公司日志框架，统一由日志平台处理。

其优点：

• 不同日志框架对应着不同的 Appender，方便进行日志框架接入的扩展。

• 采用 AOP 编程，对于接入的业务侵入性小，接入简单。

• 定义了丰富的 java 注解，便于日志配置化输出，其中可打印日志包括但不限于：类名、方法名、方法入参、返回值、异常等，支持敏感字段脱敏。

存在的问题：

• 日志格式不规范：研发应用数百个，研发人员较多，日志格式差异大，给数据分析和使用带来巨大挑战。

• 存储时长短：当前公司在线 CLOG 存储系统只能查询最近几天数据、ES 保存稍长一段时间数据且不支持批量查询，基础离线 CLOG hive 表由于数据量巨大，仅能做到 T+2，无法满足 T+1 的报表需求。

故支付数据团队在研发团队统一日志组件的基础上，结合数据分析和数据存储生命周期开发了统一日志框架。

3.1 统一日志-埋点设计

支付研发团队负责数百个服务或应用，支持的业务包括：路由、鉴权、免密、卡服务、订单、钱包实名、电子支付等，不同的业务又可拆分 app、h5、online、offline 等项目，整合这些数据是个极大的挑战。如果各系统研发埋点任意指定，会给 BI 数据分析带来极大的困难，数据分析准确性难以得到保障，故支付数据基于业务特点定义了一套统一日志埋点规范。

字段定义主要是基于日常分析需求，致力于简化数据的使用，故总体原则为 json 形式，当前原始日志有两部分组成：tag/message，其中 tag 数据结构为 Map，关键数据一般是通过 tag 内的数据进行检索，明细数据通过 message 进行检索，tag 与 message 的组成格式为：[[$tag]]$message，目前标准字段包括两类：规范性字段和通用性字段。

3.1.1 规范性字段格式

规范性字段需要应用研发一定程度的参与，提供符合字段命名的类和方法。部分字段名称及定义如下：

其中 tag 可以灵活填充，主要扩展字段如下：

3.1.2 通用字段格式

日志框架能够自动获取属性，无需研发编码，即可打印。

3.2 分区/分桶字段的定义

当前离线数据分析基于 hive 引擎，hive 的分区分桶设计极大的影响了查询性能，特别是在日志量巨大的场景下，分区字段的选择尤为关键。如：用户进入支付收银台可能会有上百个场景，而每种场景下会有多次服务调用，其中不同场景下服务调用频率差异很大，占用的空间差异也较大，故针对每种场景分配一个唯一的场景号，通过场景号进行分区，可以高效的进行数据分析，而过多的分区也可能导致较多的小文件，对 hadoop namenode 产生较大的影响，此时结合分桶能够达到优化查询效率且避免分区无限制野蛮增长产生众多过多小文件的问题。

四、日志采集

日志采集框架基于 LinkedIn 的开源项目 Camus，Camus 使用 MapReduce 读取 kafka 数据然后写入 hdfs，由于无 reduce 阶端，所有数据处理及写入都在 Map 侧，很少会发生数据倾斜，Camus 具有接入简单，方便扩展，故我们进行了二次开发，以满足当前业务的需要：

• 自定义 decoder/partitioner，原生的 decoder/partitioner 支持的 hdfs 落地路径主要基于日期，较为粗糙，无法适应业务的需要。故自定义 decoder 抽取原始日志分区字段，然后代入 partitioner 中，生成具有业务含义的 hdfs 输出路径，为特定时间范围数据回刷提供了高效的解决方案。

• 自定义 provider，原生的 StringRecordWriterProver 仅支持 text 文件方式落地，占用空间大、压缩后无法并行切分，容易错列错行，而 orc 格式数据，有效的节约了 hdfs 占用空间，查询效率高且可以切分，有利于日志解析 job 的高效执行。其中在配置 Camus job 过程中需要关注如下问题：

4.1 camus 任务执行

• 执行频率设置

The earliest offset was found to be more than the current offset

由于 kafka 消息保存天数有限和单个分区 size 有限(Server 配置：log.retention.bytes)，携程侧为 3 天和 10G，如果 camus 同步 kafka 频率较低时，可能会出现数据丢失，故需要根据日志量大小，设置 camus 调度任务的执行频率，防止数据丢失。

• 任务重叠执行

Error: java.io.IOException: target exists.the file size(614490 vs 616553) is not the same.

camus 从 kafka 读取数据，任务要以单例形式执行，任务执行完成后才会更新 kafka 的 offset，若一个任务执行了多次，就会导致数据大小无法对齐，此时需要删除配置路径下的所有数据后重新启动任务，即可完成修复。

4.2 如何控制 camus 落地文件的大小

当 kafka 各 partition 数据写入量不平衡时，由于各 partition 会写入一个 hdfs 文件中，如果研发日志集中写入 kafka 某个 partition，会导致这个 partition 对应的 hdfs 文件占用空间特别大，如果恰巧这个文件是不可切分的，极端情况下会导致只有一个线程去解析这个大文件，降低了数据读写的并发度，拉长了数据解析时间，遇到这种问题的解决办法是：

• 临时解决方案：研发日志分散写入 kafka partition，不要导致某类数据集中写入一个 partition；

• 高效解决方案：数据侧采用可切分的输入格式，进行数据切分；

4.3 写入 orc 文件格式注意事项

• orc 写入 timeout

AttemptID:attempt_1587545556983_2611216_m_000001_0 Timed out after 600 secs

orc 文件写入速度较 text 文件会慢很多，如果同时写入的的文件较多或者内存回收占用时间较长，会导致 map 方法在 600 秒内没有读、写或状态更新，job 会被尝试终结，解决方法是调高默认的 task 超时时间，由 10 分钟调高到 20 分钟。

mapreduce.task.timeout=1200000

• OOM 内存溢出

beyond physical memory limits. Current usage: 2.5 GB of 2.5 GB physical memory used; 4.2 GB of 5.3 GB virtual memory used. Killing container.

在 orc 写文件的时候如果出行较多的 OOM，此时需要加大 map 执行的内存。

mapreduce.map.memory.mb=8096mapreduce.map.java.opts=-Xmx6000m

五、统一日志-解析

鉴于日志解析工作主要集中在 MapReduce 的 Map 侧，而 Map 侧通过参数调整能够很容易控制 map 的个数，以提高数据解析的并发度，MapReduce 主要分为：intputformat、map、shuffle、reduce 等几个核心阶段，通过优化各阶段的执行时间，可以显著提高日志解析的速度。

5.1 inputsplit 优化

MR job 中影响 map 的个数主要有：

• 文件个数：如果不采用 CombineFileInputFormat，那么不会进行小文件合并，每个文件至少有一个 map 处理，当小文件太多时，频繁启动和回收线程也会对性能产生影响，同时对集群其它 job 资源分配产生影响。

• 文件属性：当文件较大且可切分时，系统会生成多个 map 处理大文件，inputsplit 块按照 MR 最小单元进行文件切割(split)，并且一个 split 对应一个 MapTask。

前期日志解析程序的性能较高，一天的全量日志解析约 25 分钟，中间有段时间任务执行时间从 25 分钟延迟到 4 个小时，原因是研发将大量订单号为空的日志写入到指定的 partition 中，日志量巨大，导致其中少量 map 在读取大文件时执行时间特别长。

经过分析发现 text+snappy 文件无法切分，只能够被一个 map 处理，将 camus 落地数据格式从 text+snappy 换为 orc+snappy 格式，同时开发了支持 orc 文件格式的 CombineFileInputFormat，既减少了小文件对 hadoop 计算资源果断的占用也提高了 job 的并发程度。

5.2 shuffle 优化

使 map 的输出能够更加均匀的映射到 reduce 侧，由于默认的分区策略是对 map 的输出 key hash 取 reduce 个数的模，容易导致数据倾斜，解决办法是在 key 上面增加时间戳或者重写 partition 函数。

5.3 批量日志解析

当前 MR 的输出会作为 hive 外表的数据源，hive 表会按照业务过程进行分区，所有数据的解析结果路径为：日期+业务过程，而业务过程可能有数百个，采用了 MultipleInputs/MultipleOutputs 能够在一个 mapreduce job 中实现多输入多输出的功能，以适应业务自定义解析，并归一化后统一抛送到 reduce 侧。

5.3.1 空文件生产

在使用的过程中会出现生成众多临时小文件及生成 size 为 0 的小文件，增加了 hdfs namenode 内存压力，同时空文件也会导致 spark 表查询失败，可通过 LazyOutputFormat 进行修复。

5.3.2 文件重复创建

MultipleOutputs 输出文件一般以 name-r-nnnnn 的格式进行命名，其中 name 与程序指定的文件名有关，nnnnn 表示 reduce 任务号。在处理数据较多时，可能会存在 reduce 侧反复创建已存在的文件，导致任务长时间运行而不能成功，中间生成了大量小文件，对 hadoop namenode 产生较大压力，影响整个集群响应时间。

解决方案为：在 reduce 侧进行数据写入时，需要对 exception 进行捕捉，一旦出现数据写入 exception，即将对应的写入 reduce 文件删除并终止程序，由于 MR 支持高可用，当一个 reduce taks 失败后会自动重试，重试一定次数依然不能够成功就会导致整个任务失败，每次重试避免了不停的重复创建已存在的文件，引起 NN 响应时间极速下降。

5.4 reduce 个数调整

目前日志解析的 reduce 侧主要用于 orc 数据写入，当 reduce 个数较少时，会导致 reduce 内存溢出，而 reduce 个数较多时，可能会导致非常多的小文件且占用集群过多资源，可以通过计算 map 侧输入文件的个数及总占用空间，动态计算需要的 reduce 个数，以达到合理利用资源的目的。

六、日志治理

日志落地导致的一个问题是存储空间增长迅速，当前支付中心日均新增 ORC 压缩原始数据量 TB 级别且还在持续增长中。

支付数据侧根据研发、产品的需求对不同类型日志进行分级，对于不同类别的日志设置不同的存储周期，主要划分为：研发排障日志、审计日志、数据分析日志等；同时在 camus 将日志写入 hdfs 时，由于按照业务分区进行落地，导致生成了大量小文件，需要对这些小文件进行合并并且设置 TTL，避免对 hadoop namenode 产生较大的影响。

七、总结与展望

目前日均 TB 级数据解析时间在 30 分钟内完成，后期计划将日志系统导入 clickhouse 等对实时要求高的场景共运营使用，以支持业务精细化运营和分析。

作者介绍：

英明，携程数据研发专家，负责支付离线数据仓库建设及 BI 业务需求，对并行计算、大数据处理及建模等有浓厚兴趣。

本文转载自公众号携程技术（ID：ctriptech）。

原文链接：

日均TB级数据，携程支付统一日志框架