日志服务数据加工的设计与实践

2022-03-04 23:00:16

前言

快速发展的日志数据

伴随Logging、Metrics、Tracing三者融合趋势的部分显现，日志类数据的范围正在泛化，包括：工业传感器数据、日志文件、Prometheus采集的云原生应用指标、Syslog、网络点击日志、stdout、业务埋点等等。

可以看到：

•数据规模快速增长，日志类数据是big data的主力。

•采集的日志数据类别正在增加，其一是日志类型泛化，其二是过去被丢弃的数据正被重拾起来。

•数据运营的理念在各行各业渗透，更多的日志数据开始得到处理，更多的人开始参与日志分析。

即使在传统日志领域，以Kubernetes为代表的云原生的流行，也带来了新的日志生命周期管理需求。如何从各种日志源采集数据、分析数据是一项复杂的挑战。

不平坦的数据分析之路

除了数据科学家、数据工程师，现在运营、DevOps工程师、用户支持等角色也在分析日志。假设有N种日志用户，对于M种类型日志，可能会产生N*M种日志存储、分析需求。

分析首先需要完整的数据采集，尤其是对大规模数据的集成、预处理和降维能力。
多元用户意味着多样化工具栈，数据应该保存在开放的存储系统，并且可以被更多的工具处理。
在传统的离线分析之外，越来越多的延迟敏感型应用出现，数据得不到及时处理会丧失其大部分价值。

上图（interana.com, State of Data Insights 2017统计）反映了一个事实：73%的人需要花费几天甚至数星期时间从数据中得到分析结果。

另一个被广泛传播的数字：在数据分析过程中，数据集成和预处理所耗费的时间占总体80%以上。

ETL已死？

完成各个数据源采集，接着以尽量统一的方式（UI、模型、计算架构）对数据做加工、查询。其中，ETL是关键的技术。

近一两年来，看到一些关于”ETL已死“的文章。但关键问题还没有解决：

OLAP + OLTP一体化的系统令人向往，但当前在数据规模、计算任务多样化上有其限定条件。
众多ETL pipeline维护的复杂性源自数据业务的复杂，例如：数据采集流程、预处理作业依赖。

因此，银弹还没有出现，对业务需求进行抽象，根据技术指标做合理的存储、计算选型是一个行之有效的办法。ETL没有消失，也在演进：

实时化ETL，无论是数据的采集还是预处理阶段。
随着一些存储、计算系统能力的增强，ETL的过程在向存储系统迁移。例如：过去在数据库难以实现大规模预处理，需要专门的ETL工具；当前则可以在一个分布式数仓内做ELT，完成系统内数据流转。
中心化采集到统一存储后加工（类Kafka模式），一定程度上优化了复杂业务上ETL网状数据拓扑。
数据湖为代表的schema-on-read模式，让ETL的发生后置。

数据预处理的流派

数据预处理是本文主题，在多年的ETL技术发展中诞生了很多相关系统。这里选取一部分做回顾：

采集端系统

日志领域，以Logstash（注：新版本支持hosted模式）、Fluentd、Flume、NXLog、Logtail（阿里巴巴）为代表，可以在采集阶段利用机器资源完成一定程度的预处理，不需要专用计算集群做预处理。不足之处是：

可用性，一旦单机客户端非预期失效，数据链路也会断掉。
很多数据源是易失的，在客户端、插件出问题时或业务逻辑改变时，数据重放是困难的。
ETL运维的复杂度难以得到改善，配置散落在众多机器上，加工逻辑变更或作业运维大多依赖机器上操作。
单机节点可能出现日志生产大于消费情况（取决于软件实现的性能），导致处理瓶颈。

数据库系统

它们首先是存储系统，基于行、列存储模型优化，支持了一定的复杂计算能力。经典的如SQL Server、Oracle数据库，如今有OceanBase、Google Spanner这样的分布式数据库。

绝大部分数据库多用于存储清洗后的数据，用于在线服务场景。

批量计算系统

以Hive、MapReduce计算为代表的Hadoop生态系统，主要是面向 OLAP、批操作设计。以可扩展的计算和海量存储能力，解决了big data分析难题。

在延迟敏感型业务占比越来越大的背景下，离线系统的延迟高、交互性差，已经不能再唱独角戏。

流计算引擎

Flink、Spark是开源社区非常流行的流计算系统，流模式让ETL变得实时化，定位于通用场景。

云上数据平台

以Alooma、AWS Glue、Azure DataFactory、阿里云DataWorks、Google Cloud DataProc为代表，各个云服务厂商基于的存储、计算服务，在一个系统上为用户提供通用、综合的数据集成、开发能力。

流式存储与计算

在很长一段时间内，以Kafka为代表的数据队列系统被用于临时数据存储。经过近些年的发展，流式存储上拓展了数据分层，基于之上的计算也已成为一个事实。例如：AWS Kinesis Streams、Kafka（KSQL/Kafka Streams）、Apache Pulsar（Pulsar Functions）。

日志服务上的数据预处理场景

阿里云日志服务（原SLS）是针对日志类数据的一站式服务，在阿里巴巴集团经历大量大数据场景锤炼而成。为用户提供快捷的日志数据采集、消费、投递以及查询分析等功能，提升运维、运营效率。

数据源

在日志服务，目前每天的数据处理规模在PB级，涵盖主要日志生态的数据源。数据集成手段包括：

客户端采集：处理机器上各种各样的日志文件、程序指标、网络数据包等。
服务端采集：以分布式、全托管服务方式采集云产品、服务上的数据。例如：云产品访问日志（SLB、OSS、CDN、API网关），网络流日志（VPC、CEN），开放服务上存储的数据（OSS文件、MaxCompute表等）。
自建软件：应用程序可以*选择基础的Restful API，多种语言SDK，基于SDK高级封装的producer lib或是logger appender上报数据。
协议网关：日志服务服务端对于Kafka、Syslog等数据协议提供接入网关，最小化日志采集代价。

场景与挑战

在日志服务上，大量的、多样的数据在日志库（Logstore）存储，进行数据分析要解决三个挑战：

规模问题
- 广泛类型的数据采集能力，一套存储完成所有类型数据的集中化。
- 海量、可伸缩的集中式存储，支撑例如审计场景下日志长期存储场景。
- 弹性扩展的数据处理，按照业务峰谷配置计算，降低为burst高峰预留资源带来的高额成本。
多元化分析需求
- 数据链路实时性要求变高，存储和计算要具备微批、流的能力。
- 一份数据可以在多处被使用，让数据开放并*流动。
- 较好的工具集成完整度和丰富的生态对接能力，适应不同用户的分析技术栈。
数据预处理的易用性
- 数据加工代码复杂度尽量低，常见日志处理逻辑做到复用。
- 全托管、服务化处理，屏蔽运维细节（failover，资源扩容）。
- GUI帮助收敛数据流程的调试、维护成本。

数据加工功能

在日志服务，数据加工功能用于完成对Logstore数据的预处理，为后续的分析阶段准备数据。

数据加工基于日志服务的流式存储，调度动态数目的worker做计算。计算上提供丰富的算子和场景化UDF，对于复杂需求则可以通过流程控制、条件判断实现行内逻辑组合，跨行的pipeline组合简化数据的嵌套处理需求。

日志服务数据加工的设计

数据模型与存储

日志服务使用一套通用的数据模型应对各种各样的数据类型。一条Log由保留字段（时间，来源等）和日志内容（多个Key-Value对）组成：

message Log
{
    required uint32 Time = 1;// UNIX Time Format
    message Content
    {
        required string Key = 1;
        required string Value = 2;
    }  
    repeated Content Contents = 2;
}

结构化的数据可以在这个数据模型上定义出表结构：

__time__ : 1572784373
__source__ : 192.168.2.13
key_a : value a
key_b : value b

同样的，对于非结构化或半结构化数据，可以在把全部内容放入一个字段中，并选择性地对字段值做一些处理（例如编码）。

日志服务存储引擎（LogHub）实现了对数据的统一存储，支持以下特性：

流式存储，十毫秒级可见。
分布式服务，多拷贝保证可靠性。
append写入，支持增量（实时）消费以及存量（回搠位置）消费。
支持Ad-hoc构建索引，结合高效编码、列存对原文存储实现快速查询、分析。

存储与计算分离

数据加工实现的是脱离存储系统之外的计算过程。基于Pull模型获取数据，可以根据worker自身的负载情况决定数据加载的速率。worker与存储系统的网络请求走阿里云内部网络，每次读取批量的数据块，结合传输过程的压缩特性，保证了同region下跨系统交换数据不会成为性能瓶颈。

日志服务的一个Logstore的数据分布在多个shard上，每一个shard被append写入数据。调度器负责以下工作：

管理N个worker到M个shard之间的映射关系，保证shard在数量维度上的负载均衡。
支持worker的水平扩展以应对大规模流量，在众多shard情况下，协调多个worker共同、完整地处理整个Logstore的数据。
worker的健康管理，动态地注册新worker或踢出失效worker。
持久化worker对shard消费进度，例如worker#1失效后，其对shard 0/1的处理进度可以被新加入的worker#3继承。

弹性是云服务的标志，在大部分日志的流量特征而言，伸缩能力显得尤为重要。

例如：直播应用的CDN access log，21:00 ~ 23:00是业务访问高峰期并产生大量日志，到了凌晨1:00 ~ 7:00日志流量跌至高峰时的10%。按业务峰值规划资源必将产生大量闲置成本。

处理延迟、数据规模、成本三者看起来是鱼和熊掌的关系，在日志服务上，尝试从两个层面来弹性应对：

存储：基于shard的动态merge/split能力实现对写入存储流量的控制，高峰时使用更多的shard。
计算：数据加工实现了基于流量的并发度控制，shard数目作为一个参考指标，根据当前整体的资源指标（cpu使用率等）动态扩容或缩容worker数目。

作业模式

日志处理场景下绕不过的是时间，时间的定义确又不那么简单。

名称	定义	日志服务上应用
event-time	事件时间，真实的业务时间	一般建议设置值到`__time__`字段，如写入时未做规划则需要从数据中自行提取
server-arrived-time	该事件到达服务端时间	日志服务在接收数据时记录值并填入`__tag__:__receive_time__`字段
processing-time	数据加工处理该事件的时间	不确定，取决于作业模式以及加工速率

对于一个加工任务而言，加工的延迟定义为processing-time - server-arrived-time(latest log)。由于数据可能迟到或生产者发送了乱序数据，event-time与server-arrived-time、processing-time可能会有较大差异。

数据加工根据server-arrived-time定义数据源范围，并提供两种作业模式：

实时模式：持续运行并加载新到来数据，*的流任务，[FROM server-arrived-time, -)。
区间模式：有界的任务，[FROM server-arrived-time, TO server-arrived-time)，常见的有补数据场景，可以重复地对过去一个时间段做加工。

场景化UDF

相较于业内流行的SQL、DSL、Python等ETL语言，日志服务数据加工提供的是类Python DSL，封装了日志领域下通用加工过程。

作为业务逻辑开发的重要一环，数据加工DSL提供以下能力：

函数级能力：支持数据过滤、抽取、分裂、富化、分发操作，可以快速解决如JSON、Nginx access log、Syslog日志解析等场景。
行内组合能力：通过条件判断与流程控制，可以组合多个函数调用完成复杂操作，例如：e_if_else(condition_1, e_compose(operation_1, operation_2, operation_3), operation_4)。
跨行组合能力：常用于数据处理pipeline，作用类似SQL子查询。数据加工跨行组合是类管道式语法，从代码调试效率和可读性上看，比SQL子查询表现更好。

例如，在数据加工DSL中实现对一条日志的分裂、拷贝、条件判断，其内部编排逻辑如下图：

DevOps效率

开发、运维效率是考量数据流程维护成本的重要指标。

日志服务数据加工是全托管的服务，使用它不感知机器资源，通过web控制台实现对作业的管理与监控。

开发与调试：web控制台操作。

部署与迭代：调试完成的代码一键保存作业运行。DSL代码更新后，控制台上进行重启完成重新部署。
指标监控：包括概览、加工吞吐、shard级消费延迟与速率指标。

诊断日志：汇聚了加工过程错误日志，可以根据reason字段进行细节定位。

作业告警：在加工任务运行指标的仪表盘上，可以对某个指标设置监控告警，也可以订阅仪表盘发送到钉钉webhook。做到对数据加工作业的运行状态的充分掌握。

基于数据加工的场景实践

流动的数据

在日志的整个生命周期内，数据采集到日志服务存储，数据加工在这之后起着承转启合作用。通过数据加工完成清洗、预处理、分发，让数据在生态流转起来，并更好地适配目标存储的schema要求。

规整

数据规整包括字段抽取、过滤、清洗等工作，完成后数据被转储到下游。规整的意义在于能为下游带来哪些帮助：

Logstore的数据规整后写入新Logstore，在后者基础上精细化Key-Value索引可以帮助优化成本，提升查询分析效率，让仪表盘与告警表达更加丰富。
Logstore数据规整后写入OSS bucket，如此构建的数据湖可以大大优化存储成本和后续分析效率。参考Analyzing Data in S3 using Amazon Athena数字，对于S3上的ELB访问日志，结构化良好的parquet文件对比普通text文件，可以缩小87%存储空间并在部分场景下提升34倍分析效率。
Logstore数据规整后写入数据库是刚需，整条日志原文存储数据库在后续面临性能开销、不规则数据带来计算不确定性（可能引入复杂的兼容逻辑）。

如下，content字段是完整的Syslog日志原文，这样一条非结构化数据，通过两行加工代码分别完成Syslog字段抽取、priority字段映射。