前言

快速发展的日志数据

伴随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数目。

作业模式

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

对于一个加工任务而言，加工的延迟定义为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字段映射。

对于JSON格式的结构化日志，如下两行代码通过JMES语法对数组做分拆，分拆后每个子对象分别做嵌套字段提取。

更多实践：

数据脱敏
日志时间处理
复杂JSON字段提取
类JSON、非标准JSON、XML格式解析
分隔符日志字段提取
Key-Value格式字段提取
Ngnix日志解析
Syslog数据解析

分发

日志分发、复制是一种典型的数据场景。

例如：Kubernetes上采集的众多pod日志集中化到一个Logstore上，可以通过数据加工快速实现按namespace转发到下游Logstore，在下游Logstore上分别设置存储周期、索引分析字段。

数据除了在Logstore之间做流转以外，还可以流向异构存储系统，例如投递到OSS、MaxCompute、ADB等。

更多实践：

多目标Logstore数据分发
多源Logstore数据汇总
Logstore数据投递OSS

富化

对于一个典型的SLB+ECS+Nginx架构，Nginx access log上包括请求来源（__source__字段，记录vpc子网ip）、请求资源（request_uri字段，参数记录了业务租户的project信息）。

RDS中维护了两张维表：

• 用户元信息表，主键为业务租户的project信息。
• ECS服务器元信息表，主键为内网ip。

数据加工首先对request_uri做参数拆分，获取project信息。接下来分别通过ip与project值与两个维表做join，得到结果是更完整的日志信息（包括后端服务器的tag、租户project的打标内容）。

数据加工目前支持四种数据源做查找富化：本地配置、RDS表、OSS文件、日志服务Logstore。

更多实践：

从RDS MySQL获取数据做富化
从OSS文件获取数据做富化
从日志服务 Logstore获取数据做富化
自定义条件实现数据富化的复杂映射

写在最后，ETL业务场景千变万化，数据加工在数据分析场景支撑的路上将持续迭代优化。

双11福利来了！先来康康#怎么买云服务器最便宜# [并不简单]参团购买指定配置云服务器仅86元/年，开团拉新享三重礼：1111红包+瓜分百万现金+31%返现，爆款必买清单，还有iPhone 11 Pro、卫衣、T恤等你来抽，马上来试试手气
https://www.aliyun.com/1111/2019/home?utm_content=g_1000083110

阅读原文
本文为云栖社区原创内容，未经允许不得转载。