弱结构化日志 Flink SQL 怎么写？SLS SPL 来帮忙

背景

日志服务SLS是云原生观测与分析平台，为Log、Metric、Trace等数据提供大规模、低成本、实时的平台化服务，基于日志服务的便捷的数据接入能力，可以将系统日志、业务日志等接入SLS进行存储、分析；阿里云Flink是阿里云基于Apache Flink构建的大数据分析平台，在实时数据分析、风控检测等场景应用广泛。阿里云Flink原生支持阿里云日志服务SLS的Connector，用户可以在阿里云Flink平台将SLS作为源表或者结果表使用。

阿里云Flink SLS Connector对于结构化的日志非常直接，通过配置，SLS的日志字段可以与Flink SQL的Table字段列一一映射；然后仍有大量的业务日志并非完全的结构化，例如会将所有日志内容写入一个字段中，需要正则提前、分隔符拆分等手段才可以提取出结构化的字段，基于这个场景，本文介绍一种使用SLS SPL配置SLS Connector完成数据结构化的方案，覆盖日志清洗与格式规整场景。

弱结构化日志处理的痛点

弱结构化日志现状与结构化处理需求的矛盾

日志数据往往是多种来源，多种格式，往往没有固定的Schema，所以在数据处理前，需要先对数据进行清洗、格式规整，然后在进行数据分析；这类数据内容格式是不固定的，可能是JSON字符串、CSV格式，甚至是不规则的Java堆栈日志。

Flink SQL是一种兼容SQL语法的实时计算模型，可以基于SQL对结构化数据进行分析，但同时也要求源数据模式固定：字段名称、类型、数量是固定；这也是SQL计算模型的基础。

日志数据的弱结构化特点与Flink SQL结构化分析之间有着一道鸿沟，跨越这道鸿沟需要一个中间层来进行数据清洗、规整；这个中间层的方案有多种选择可以使用，下面会对不同的方案做简单对比，并提出一种新的基于SLS SPL的方案来轻量化完成解决数据清洗规整的工作。

弱结构化日志数据

下面是一条日志示例，日志格式较为复杂，既有JSON字符串，又有字符串与JSON混合的场景。其中：

Payload为JSON字符串，其中schedule字段的内容也是一段JSON结构。
requestURL为一段标准的URL Path路径。
error字段是前半部分包含CouldNotExecuteQuery：字符串，后半部分是一段JSON结构。
__tag__:__path__包含日志文件的路径，其中service_a可能是业务名称。
caller中包含文件名与文件行数。

{
"Payload": "{\"lastNotified\": 1705030483, \"serverUri\": \"http://test.alert.com/alert-api/tasks\", \"jobID\": \"44d6ce47bb4995ef0c8052a9a30ed6d8\", \"alertName\": \"alert-12345678-123456\", \"project\": \"test-sls-project\", \"projectId\": 123, \"aliuid\": \"1234567890\", \"alertDisplayName\": \"\\u6d4b\\u8bd5\\u963f\\u91cc\\u4e91\\u544a\\u8b66\", \"checkJobUri\": \"http://test.alert.com/alert-api/task_check\", \"schedule\": {\"timeZone\": \"\", \"delay\": 0, \"runImmediately\": false, \"type\": \"FixedRate\", \"interval\": \"1m\"}, \"jobRunID\": \"bf86aa5e67a6891d-61016da98c79b-5071a6b\", \"firedNotNotified\": 25161}",
"TaskID": "bf86aa5e67a6891d-61016da98c79b-5071a6b-334f81a-5c38aaa1-9354-43ec-8369-4f41a7c23887",
"TaskType": "ALERT",
"__source__": "11.199.97.112",
"__tag__:__hostname__": "iabcde12345.cloud.abc121",
"__tag__:__path__": "/var/log/service_a.LOG",
"caller": "executor/pool.go:64",
"error": "CouldNotExecuteQuery : {\n \"httpCode\": 404,\n \"errorCode\": \"LogStoreNotExist\",\n \"errorMessage\": \"logstore k8s-event does not exist\",\n \"requestID\": \"65B7C10AB43D9895A8C3DB6A\"\n}",
"requestURL": "/apis/autoscaling/v2beta1/namespaces/python-etl/horizontalpodautoscalers/cn-shenzhen-56492-1234567890123?timeout=30s",
"ts": "2024-01-29 22:57:13"
}

结构化数据处理需求

对于这样的日志提取出更有价值的信息需要进行数据清洗，首先需要提取重要的字段，然后对这些字段进行数据分析；本篇关注重要字段的提取，分析仍然可以在Flink中进行。

假设提取字段具体需求如下：

提取error中的httpCode、errorCode、errorMessage、requestID
提取__tag__:__path__中的service_a作为serviceName
提取caller中的pool.go作为fileName，64作为fileNo
提取Payload中的project；提取Payload下面的schedule中的type为scheuleType
重命名__source__为serviceIP
其余字段舍弃

最终需要的字段列表如下，基于这样一个表格模型，我们可以便捷的使用Flink SQL进行数据分析。

解决方案

实现这样的数据清洗，有很多种方法，这里列举几种基于SLS与Flink的方案，不同方案之间没有绝对的优劣，需要根据不同的场景选择不同的方案。

数据加工方案：在SLS控制台创建目标Logstore，通过创建数据加工任务，完成对数据的清洗。

Flink方案：将error和payload指定为源表字段，通过SQL 正则函数、JSON函数对字段进行解析，解析后的字段写入临时表，然后对临时表进行分析。

SPL方案：在Flink SLS Connector中配置SPL语句，对数据进行清洗，Flink中源表字段定义为清洗后的数据结构。

从上述三种方案的原理不难看出，在需要数据清洗的场景中，在SLS Connector 中配置SPL是一种更轻量化的方案，具有轻量化、易维护、易扩展的特点。

在日志数据弱结构化的场景中，SPL方案既避免了方案一中创建临时中间Logstore，也避免了方案二中在Flink中创建临时表，在离数据源更近的位置进行数据清洗，在计算平台关注业务逻辑，职责分离更加清晰。

如何在Flink中使用SPL

接下来以一个一段弱结构化日志为例，来介绍基于SLS SPL的能力来使用Flink。为了便于演示，这里在Flink控制台配置SLS的源表，然后开启一个连续查询以观察效果。在实际使用过程中，仅需修改SLS源表配置，即可完成数据清洗与字段规整。

SLS准备数据

开通SLS，在SLS创建Project，Logstore，并创建具有消费Logstore的权限的账号AK/SK。
当前Logstore数据使用SLS SDK写入模拟数据，格式使用上述日志片段，其中包含JSON、复杂字符串等弱结构化字段。

预览SPL效果

在Logstore可以可以开启扫描模式，类似Unix管道，SLS SPL管道式语法使用|分隔符分割不同的指令，每次输入一个指令可以即时查看结果，然后在增加管道数，渐进式、探索式获取最终结果。

对上图中的SPL进行简单描述

* | project Payload, error, "__tag__:__path__", "__tag__:__hostname__", caller 
| parse-json Payload 
| project-away Payload 
| parse-regexp error, 'CouldNotExecuteQuery : ({[\w":\s,\-}]+)' as errorJson 
| parse-json errorJson 
| parse-regexp "__tag__:__path__", '\/var\/log\/([\w\_]+).LOG' as serviceName 
| parse-regexp caller, '\w+/([\w\.]+):(\d+)' as fileName, fileNo 
| project-rename serviceHost="__tag__:__hostname__" 
| extend scheduleType = json_extract_scalar(schedule, '$.type') 
| project httpCode, errorCode,errorMessage,requestID,fileName, fileNo, serviceHost,scheduleType, project

1行：project指令：从原始结果中保留Payload、error、__tag__:__path__、caller字段，舍弃其他字段，这些字段用于后续解析
2行：parse-json指令：将Payload字符串展开为JSON，第一层字段出现在结果中，包括lastNotified、serviceUri、jobID等
3行：project-away指令：去除原始Payload字段
4行：parse-regexp指令：按照error字段中的内容，解析其中的部分JSON内容，置于errorJson字段
5行：parse-json指令：展开errorJson字段，得到httpCode、errorCode、errorMessage等字段
6行：parse-regexp指令：通过正则表达式解析出__tag__:__path__种的文件名，并命名为serviceName
7行：parse-regexp指令：通过正则表达式捕获组解析出caller种的文件名与行数，并置于fileName、fileNo字段
8行：project-rename指令：将__tag__:__hostname__字段重命名为serviceHost
9行：extend指令：使用json_extract_scalar函数，提取schedule中的type字段，并命名为scheduleType
10行：project指令：保留需要的字段列表，其中project字段来自于Payload

创建SQL作业

在阿里云Flink控制台创建一个空白的SQL的流作业草稿，点击下一步，进入作业编写

在作业草稿中输入如下创建临时表的语句：

CREATE TEMPORARY TABLE sls_input_complex (
errorCode STRING,
errorMessage STRING,
fileName STRING,
fileNo STRING,
httpCode STRING,
requestID STRING,
scheduleType STRING,
serviceHost STRING,
project STRING,
proctime as PROCTIME()
) WITH (
'connector' = 'sls',
'endpoint' ='cn-beijing-intranet.log.aliyuncs.com',
'accessId' = '${ak}',
'accessKey' = '${sk}',
'starttime' = '2024-02-01 10:30:00',
'project' ='${project}',
'logstore' ='${logtore}',
'query' = '* | project Payload, error, "__tag__:__path__", "__tag__:__hostname__", caller | parse-json Payload | project-away Payload | parse-regexp error, ''CouldNotExecuteQuery : ({[\w":\s,\-}]+)'' as errorJson | parse-json errorJson | parse-regexp "__tag__:__path__", ''\/var\/log\/([\w\_]+).LOG'' as serviceName | parse-regexp caller, ''\w+/([\w\.]+):(\d+)'' as fileName, fileNo | project-rename serviceHost="__tag__:__hostname__" | extend scheduleType = json_extract_scalar(schedule, ''$.type'') | project httpCode, errorCode,errorMessage,requestID,fileName, fileNo, serviceHost,scheduleType,project'
);

其中${ak},${sk},${project},${logstore}需要替换为有消费权限的AK账号。
query字段，替换为上述SPL，注意在阿里云Flink控制台需要对单引号使用单引号转义，并且消除换行符。
SPL最终得到的字段列表与TABLE中字段对应。

连续查询及效果

在作业中输入分析语句，查看结果数据

SELECT * FROM sls_input_complex

点击右上角调试按钮，进行调试，可以看到TABLE中每一列的值，对应SPL处理后的结果。

总结

为了适应弱结构化日志数据的需求，Flink SLS Connector进行了升级，支持直接通过Connector配置SPL的方式实现SLS数据源的清洗下推，特别是需要正则字段提取、JSON字段提取、CSV字段提取场景下，相较原数据加工方案和原Flink SLS Connector方案更轻量级，让数据清洗的职责更加清晰，在数据源端完成数据清洗工作，也可以减少数据的网络传输流量，使得到达Flink的数据已经是规整好的数据，可以更加专注在Flink中进行业务数据分析。

同时为了便于SPL验证测试，SLS 扫描查询也已支持使用SPL进行查询，可以实时看到SPL管道式语法执行结果。

参考

日志服务概述：https://help.aliyun.com/zh/sls/product-overview/what-is-log-service
SPL概述：https://help.aliyun.com/zh/sls/user-guide/spl-overview
阿里云Flink Connector SLS：https://help.aliyun.com/zh/flink/developer-reference/log-service-connector
SLS 扫描查询：https://help.aliyun.com/zh/sls/user-guide/scan-based-query-overview

背景

弱结构化日志处理的痛点

弱结构化日志现状与结构化处理需求的矛盾

弱结构化日志数据

下面是一条日志示例，日志格式较为复杂，既有JSON字符串，又有字符串与JSON混合的场景。其中：

Payload为JSON字符串，其中schedule字段的内容也是一段JSON结构。
requestURL为一段标准的URL Path路径。
error字段是前半部分包含CouldNotExecuteQuery：字符串，后半部分是一段JSON结构。
__tag__:__path__包含日志文件的路径，其中service_a可能是业务名称。
caller中包含文件名与文件行数。

{
"Payload": "{\"lastNotified\": 1705030483, \"serverUri\": \"http://test.alert.com/alert-api/tasks\", \"jobID\": \"44d6ce47bb4995ef0c8052a9a30ed6d8\", \"alertName\": \"alert-12345678-123456\", \"project\": \"test-sls-project\", \"projectId\": 123, \"aliuid\": \"1234567890\", \"alertDisplayName\": \"\\u6d4b\\u8bd5\\u963f\\u91cc\\u4e91\\u544a\\u8b66\", \"checkJobUri\": \"http://test.alert.com/alert-api/task_check\", \"schedule\": {\"timeZone\": \"\", \"delay\": 0, \"runImmediately\": false, \"type\": \"FixedRate\", \"interval\": \"1m\"}, \"jobRunID\": \"bf86aa5e67a6891d-61016da98c79b-5071a6b\", \"firedNotNotified\": 25161}",
"TaskID": "bf86aa5e67a6891d-61016da98c79b-5071a6b-334f81a-5c38aaa1-9354-43ec-8369-4f41a7c23887",
"TaskType": "ALERT",
"__source__": "11.199.97.112",
"__tag__:__hostname__": "iabcde12345.cloud.abc121",
"__tag__:__path__": "/var/log/service_a.LOG",
"caller": "executor/pool.go:64",
"error": "CouldNotExecuteQuery : {\n \"httpCode\": 404,\n \"errorCode\": \"LogStoreNotExist\",\n \"errorMessage\": \"logstore k8s-event does not exist\",\n \"requestID\": \"65B7C10AB43D9895A8C3DB6A\"\n}",
"requestURL": "/apis/autoscaling/v2beta1/namespaces/python-etl/horizontalpodautoscalers/cn-shenzhen-56492-1234567890123?timeout=30s",
"ts": "2024-01-29 22:57:13"
}

结构化数据处理需求

假设提取字段具体需求如下：

提取error中的httpCode、errorCode、errorMessage、requestID
提取__tag__:__path__中的service_a作为serviceName
提取caller中的pool.go作为fileName，64作为fileNo
提取Payload中的project；提取Payload下面的schedule中的type为scheuleType
重命名__source__为serviceIP
其余字段舍弃

最终需要的字段列表如下，基于这样一个表格模型，我们可以便捷的使用Flink SQL进行数据分析。

解决方案

实现这样的数据清洗，有很多种方法，这里列举几种基于SLS与Flink的方案，不同方案之间没有绝对的优劣，需要根据不同的场景选择不同的方案。

数据加工方案：在SLS控制台创建目标Logstore，通过创建数据加工任务，完成对数据的清洗。

Flink方案：将error和payload指定为源表字段，通过SQL 正则函数、JSON函数对字段进行解析，解析后的字段写入临时表，然后对临时表进行分析。

SPL方案：在Flink SLS Connector中配置SPL语句，对数据进行清洗，Flink中源表字段定义为清洗后的数据结构。

从上述三种方案的原理不难看出，在需要数据清洗的场景中，在SLS Connector 中配置SPL是一种更轻量化的方案，具有轻量化、易维护、易扩展的特点。

如何在Flink中使用SPL

SLS准备数据

开通SLS，在SLS创建Project，Logstore，并创建具有消费Logstore的权限的账号AK/SK。
当前Logstore数据使用SLS SDK写入模拟数据，格式使用上述日志片段，其中包含JSON、复杂字符串等弱结构化字段。

预览SPL效果

对上图中的SPL进行简单描述

* | project Payload, error, "__tag__:__path__", "__tag__:__hostname__", caller 
| parse-json Payload 
| project-away Payload 
| parse-regexp error, 'CouldNotExecuteQuery : ({[\w":\s,\-}]+)' as errorJson 
| parse-json errorJson 
| parse-regexp "__tag__:__path__", '\/var\/log\/([\w\_]+).LOG' as serviceName 
| parse-regexp caller, '\w+/([\w\.]+):(\d+)' as fileName, fileNo 
| project-rename serviceHost="__tag__:__hostname__" 
| extend scheduleType = json_extract_scalar(schedule, '$.type') 
| project httpCode, errorCode,errorMessage,requestID,fileName, fileNo, serviceHost,scheduleType, project

1行：project指令：从原始结果中保留Payload、error、__tag__:__path__、caller字段，舍弃其他字段，这些字段用于后续解析
2行：parse-json指令：将Payload字符串展开为JSON，第一层字段出现在结果中，包括lastNotified、serviceUri、jobID等
3行：project-away指令：去除原始Payload字段
4行：parse-regexp指令：按照error字段中的内容，解析其中的部分JSON内容，置于errorJson字段
5行：parse-json指令：展开errorJson字段，得到httpCode、errorCode、errorMessage等字段
6行：parse-regexp指令：通过正则表达式解析出__tag__:__path__种的文件名，并命名为serviceName
7行：parse-regexp指令：通过正则表达式捕获组解析出caller种的文件名与行数，并置于fileName、fileNo字段
8行：project-rename指令：将__tag__:__hostname__字段重命名为serviceHost
9行：extend指令：使用json_extract_scalar函数，提取schedule中的type字段，并命名为scheduleType
10行：project指令：保留需要的字段列表，其中project字段来自于Payload

创建SQL作业

在阿里云Flink控制台创建一个空白的SQL的流作业草稿，点击下一步，进入作业编写

在作业草稿中输入如下创建临时表的语句：

CREATE TEMPORARY TABLE sls_input_complex (
errorCode STRING,
errorMessage STRING,
fileName STRING,
fileNo STRING,
httpCode STRING,
requestID STRING,
scheduleType STRING,
serviceHost STRING,
project STRING,
proctime as PROCTIME()
) WITH (
'connector' = 'sls',
'endpoint' ='cn-beijing-intranet.log.aliyuncs.com',
'accessId' = '${ak}',
'accessKey' = '${sk}',
'starttime' = '2024-02-01 10:30:00',
'project' ='${project}',
'logstore' ='${logtore}',
'query' = '* | project Payload, error, "__tag__:__path__", "__tag__:__hostname__", caller | parse-json Payload | project-away Payload | parse-regexp error, ''CouldNotExecuteQuery : ({[\w":\s,\-}]+)'' as errorJson | parse-json errorJson | parse-regexp "__tag__:__path__", ''\/var\/log\/([\w\_]+).LOG'' as serviceName | parse-regexp caller, ''\w+/([\w\.]+):(\d+)'' as fileName, fileNo | project-rename serviceHost="__tag__:__hostname__" | extend scheduleType = json_extract_scalar(schedule, ''$.type'') | project httpCode, errorCode,errorMessage,requestID,fileName, fileNo, serviceHost,scheduleType,project'
);

其中${ak},${sk},${project},${logstore}需要替换为有消费权限的AK账号。
query字段，替换为上述SPL，注意在阿里云Flink控制台需要对单引号使用单引号转义，并且消除换行符。
SPL最终得到的字段列表与TABLE中字段对应。

连续查询及效果

在作业中输入分析语句，查看结果数据

SELECT * FROM sls_input_complex

点击右上角调试按钮，进行调试，可以看到TABLE中每一列的值，对应SPL处理后的结果。

总结

同时为了便于SPL验证测试，SLS 扫描查询也已支持使用SPL进行查询，可以实时看到SPL管道式语法执行结果。

参考

日志服务概述：https://help.aliyun.com/zh/sls/product-overview/what-is-log-service
SPL概述：https://help.aliyun.com/zh/sls/user-guide/spl-overview
阿里云Flink Connector SLS：https://help.aliyun.com/zh/flink/developer-reference/log-service-connector
SLS 扫描查询：https://help.aliyun.com/zh/sls/user-guide/scan-based-query-overview

弱结构化日志 Flink SQL 怎么写？SLS SPL 来帮忙

背景

弱结构化日志处理的痛点

弱结构化日志现状与结构化处理需求的矛盾

弱结构化日志数据

结构化数据处理需求

解决方案

如何在Flink中使用SPL

SLS准备数据

预览SPL效果

创建SQL作业

连续查询及效果

总结

参考

弱结构化日志 Flink SQL 怎么写？SLS SPL 来帮忙

背景

弱结构化日志处理的痛点

弱结构化日志现状与结构化处理需求的矛盾

弱结构化日志数据

结构化数据处理需求

解决方案

如何在Flink中使用SPL

SLS准备数据

预览SPL效果

创建SQL作业

连续查询及效果

总结

参考

UModel 最佳实践 Vol.1 UModel 数据建模全景解读

云监控2.0全景：可观测范式升级与智能运维蓝图

推荐文章

给 OpenClaw 加上企业级 Memory

阿里云 STAROps 全域智能运维平台发布！

阿里云正式发布 RCA Benchmark

推荐视频

UModel 最佳实践 Vol.1 UModel 数据建模全景解读

云监控2.0全景：可观测范式升级与智能运维蓝图

推荐工具

云监控 2.0 沙箱体验

免费网络拨测工具