摘要：日志服务的LogHub是流式的数据中心，日志写入后可实时消费。日志服务ETL面向的正是这些流式写入的数据，提供准实时（1分钟级别）的E�TL作业。
概述
ETL（Extract-Transform-Load）用来描述将数据从来源端经过抽取(Extract)、转换(Transform)、加载(Load)至目的端的过程。
传统ETL是构建数据仓库的重要一环，用户从数据源抽取出所需的数据，经过数据清洗，最终按照预先定义好的数据仓库模型，将数据加载到数据仓库中去。
在今天，随着业务需求的日益增加，不同系统的相互大批量数据交互也已成为常态。数据在不同系统中流动起来，有助于充分发掘日志大数据的价值。
在云上，AWS Glue是一个功能完备的ETL产品，通过DevEndpoint、Crawler、MetaStore等模块，分别在ETL函数编写、数据识别加载、meta管理等方向做到了不错的用户体验。如同大部分业界方案，Glue也是batch触发的ETL模型，建议数据源是S3。但对于流式数据（如Kinesis Stream）则不能直接使用，提供的方案是：用户先将数据导入到S3，再使用Glue分析S3上的数据。
日志服务的LogHub是流式的数据中心，日志写入后可实时消费。日志服务ETL面向的正是这些流式写入的数据，提供准实时（1分钟级别）的ETL作业。
日志服务ETL
两个场景
一站式建模分析
通过日志服务，快速完成日志采集、加工、查询、分析。
数据交换
为数据的目的端落地提供支撑，构建云上大数据产品间的数据管道。
ETL模型
实时数据流处理，基于流的模型。ETL Trigger轮询源logstore下各shard的写入位置，并定时生成三元组信息触发函数执行，该三元组用于标识本次ETL任务所应该处理的数据范围。
通过shard的并发做到水平扩展，shard弹性伸缩保证了ETL的动态伸缩，通过定时器触发作业完成持续的数据加载。
在ETL任务执行层面，考虑UDF的灵活性，加工逻辑会跑在函数服务的函数上，而函数服务提供了按需付费、弹性伸缩能力以及自定义代码执行功能，正是很多云上用户所需要的。另一方面，从用户数据端到端延时、大数据吞吐、SQL易用性角度，日志服务未来也考虑把ETL的runtime扩展到流计算引擎（例如阿里云流计算）上，去服务更多的用户场景。
ETL日志
ETL过程日志
这是一类是执行过程日志，这一部分日志是在ETL执行过程中每执行一步的记录关键点和错误，包括某一步骤的开始、结束时间、初始化动作完成情况，模块出错信息等。记录日志的目的是随时可以知道ETL运行情况，如果出错了，可以知道哪里出错。
函数运行产生的日志记录了数据加工过程中关键点、异常：
ETL调度日志
调度日志只记录ETL任务开始的时间、结束时间，任务是否成功以及成功返回的信息。如果ETL任务出错了，不仅要形成ETL出错日志，而且要向系统管理员发送报警邮件或短信。
在调度日志的基础上，可以构建出报表统计ETL的总体运行状况，会在下文实践部分介绍。
“日志服务+函数服务”ETL的优势
一站式采集、存储、加工、分析
全托管加工任务，按时间触发，自动重试
资源按shard水平扩展，满足大数据需求
基于函数服务提供数据加工，弹性资源，按需付费
ETL对用户透明，提供日志、报警功能
持续增加内置函数模板，降低主流需求下的函数开发代价
日志服务ETL实战
对于数据分析工程师而言，ETL过程往往占据整个项目工作60%~70%的工作量。日志服务的目标是使用内置的函数模板的前提下，将构建ETL的时间缩短到15分钟内。
题目：ip归属查找
通过Nginx、apache等HTTP服务器构建的软件，可以记录每一个用户访问日志。本次实践的题目是：看看我们到底服务了哪些地区的用户，这些用户通过什么链路访问我们的服务。
第一步：日志集中化存储
我们使用日志服务的Logtail客户端快速接入机器上的日志文件。本节请参考日志服务实时采集数据，本文不作赘述。
客户端采集nginx访问日志将会集中存储到日志服务的一个logstore中，如下图，forward字段的ip记录了用户请求的来源：
第二步：云端数据加工
1. 登录函数服务控制台创建service
在高级配置中，建议为ETL function配置加工过程中的日志记录的存储logstore，方便通过日志来定位加工过程中的异常行为。为函数授予日志服务AliyunLogFullAccess权限，函数在运行过程中会读源logstore数据，数据处理后再写到目标logstore。
2. 通过内置模板创建函数
默认的函数配置如下：
3. 在函数上新建日志服务触发器
日志服务触发器配置如下：
指定数据源为第一步中采集到中心化nginx日志logstore，例如本例子的project:etl-test/logstore:nginx_access_log。
日志服务将轮询logstore的数据，当数据持续产生时，每60秒创建一次ETL任务，并调用函数执行。触发函数执行以及函数执行结果将会记录到触发器日志logstore：etl-trigger-log中。
函数配置因不同函数的实现和功能而已，ip-lookup的详细配置项说明请参考README。
4. 保存配置，等待1分钟后ETL任务开始执行
可以关注一下ETL过程日志、调度日志，按如上配置，分别在logstore：etl-function-log、etl-trigger-log。
可以通过查询语句构建出如本文日志部分所示的报表：
左上图是每分钟调度函数执行的触发次数，构建自查询语句：
project_name : etl-test and job_name : ceff019ca3d077f85acaad35bb6b9bba65da6717 |selectfrom_unixtime(__time__ - __time__ %60)ast,count(1)asinvoke_countgroupbyfrom_unixtime(__time__ - __time__ %60)orderbytasclimit1000
右上图是ETL任务成功、失败的比例，构建自查询语句：
project_name : etl-testandjob_name : ceff019ca3d077f85acaad35bb6b9bba65da6717 |selecttask_status,count(1)groupbytask_status
左下图是每5分钟的摄入的日志字节数，构建自查询语句：
project_name : etl-testandjob_name : ceff019ca3d077f85acaad35bb6b9bba65da6717andtask_status : Success |selectfrom_unixtime(__time__ - __time__ %300)ast, sum(ingest_bytes)asingest_bytesgroupbyfrom_unixtime(__time__ - __time__ %300)orderbytasclimit1000
右下图则是每5分钟摄入处理的日志行数，构建自查询语句：
project_name : etl-testandjob_name : ceff019ca3d077f85acaad35bb6b9bba65da6717andtask_status : Success |selectfrom_unixtime(__time__ - __time__ %300)ast, sum(ingest_lines)asingest_linesgroupbyfrom_unixtime(__time__ - __time__ %300)orderbytasclimit1000
第三步：加工后数据建模
机器上的nginx日志经由Logtail实时采集到源logstore，再由ETL准实时加工后写出到目标logstore。经函数处理后带ip信息数据如下：
对比加工前后，我们发现，新的数据增加了四个字段（country、省province、city、isp），可以知道：ip源117.136.90.160的请求来自中国山西太原，运营商是中国移动。
接下来，使用日志服务的日志分析功能查询一个时间段内请求ip的城市和isp分布。通过如下两个查询语句构建报表：
* |selectcity, count(1)ascgroupbycityorderbycdesclimit15
* |selectisp, count(1)ascgroupbyisporderbycdesclimit15
至此，本节的实践内容结束。欢迎大家试用自定义ETL。
原文链接
网友评论