阿里云Flink SQL如何基于SPL实现弱结构化分析 - Simple Log Service

Background

Simple Log Service (SLS) is a cloud-native observability and analytics platform that provides large-scale, low-cost, real-time services for logs, metrics, and traces. You can collect system logs, business logs, and other data into SLS for storage and analysis. Realtime Compute for Apache Flink is a big data analytics platform built by Alibaba Cloud based on Apache Flink, widely used for real-time data analysis and risk monitoring. Realtime Compute for Apache Flink natively supports the SLS connector, allowing you to use SLS as a source table or result table.

The SLS connector for Realtime Compute for Apache Flink handles structured logs directly by mapping log fields one-to-one with Flink SQL table columns. However, many business logs are not fully structured. For example, all log content might be written into a single field, requiring regular expressions or delimiter splitting to extract structured fields. This topic describes how to use SPL in the SLS connector to structure such data, covering log cleansing and format normalization.

Semi-structured log data

The following sample log has a complex format that mixes JSON strings with other data types. The log contains:

The Payload field, which is a JSON string where the nested schedule field is also a JSON structure.
The requestURL field, which is a standard URL path.
The error field, which starts with the string CouldNotExecuteQuery and is followed by a JSON structure.
The __tag__:__path__ field, which contains the log file path, where service_a might be the service name.
The caller field, which contains the file name and line number.

{
  "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.XXX.XXX",
  "__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"
}

Data structuring requirements

To extract valuable information from these logs, you must transform the data by extracting key fields for analysis.

From the error field, extract httpCode, errorCode, errorMessage, and requestID.
From the __tag__:__path_ field, extract _service_a as serviceName.
From the caller field, extract pool.go as fileName and 64 as fileNo.
From the Payload field, extract project. From the nested schedule object within Payload, extract type and name it scheduleType.
Rename the __source__ field to serviceIP.

All other fields are discarded. The final list of required fields is: httpCode, errorCode, errorMessage, requestID, serviceName, fileName, fileNo, project, scheduleType, and serviceIP.

Solutions

Several methods are available to transform the data. This section compares solutions that use SLS and Flink, each suited to different scenarios.

Data transformation solution: In the SLS console, create a data transformation job to cleanse the data and store it in a target logstore.
Flink solution: Define error and payload as fields in the source table. Use Flink SQL's regular expression and JSON functions to parse these fields, write the results to a temporary table, and then perform analysis on that table.
SPL solution: Configure an SPL statement in the Flink SLS connector to transform the data. The Flink source table is then defined with the final, structured schema.

Configuring SPL in the SLS connector is a more lightweight approach. For semi-structured log data, the SPL solution avoids creating an intermediate logstore (required by the data transformation solution) and avoids creating a temporary table in Flink (required by the Flink solution). By performing data transformation closer to the source, you can focus on business logic in the computing platform, creating a clearer separation of responsibilities.

Use SPL in Flink

1. Prepare data in SLS

Make sure you have activated Simple Log Service and created a project and logstore.
Write the preceding log snippet to your target logstore using an SDK to simulate sample data.

After the data is written, go to the Raw Logs tab in the Simple Log Service console to view the simulated data. The log content includes fields such as TaskType (with the value ALERT), Payload (which contains alertName, project, and aliuid), and error information (error code LogStoreNotExist, message logstore k8s-event does not exist, and HTTP status 404).
In the logstore, write an SLS SPL pipeline statement and preview the results.

After you run the SPL statement, the Raw Logs tab displays the parsed log records, which include structured fields such as errorCode, errorMessage, fileName, fileNo, httpCode, project, requestID, scheduleType, and serviceHost.

The SPL query statement is as follows. The SPL pipeline syntax uses the pipe (|) delimiter to separate commands. You can enter one command at a time to see the immediate result, and then add more pipes to iteratively build the final query. For more information, see Scan query syntax.
```
* | 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
```
Syntax explanation:
- Line 1: The project command retains the Payload, error, __tag__:__path__, and caller fields for parsing and discards all other fields.
- Line 2: The parse-json command expands the Payload string into a JSON object. Its top-level fields, such as lastNotified, serviceUri, and jobID, are added to the result.
- Line 3: The project-away command removes the original Payload field.
- Line 4: The parse-regexp command uses a regular expression to extract the JSON part of the error field and assigns it to a new field named errorJson.
- Line 5: The parse-json command expands the errorJson field, extracting httpCode, errorCode, and errorMessage.
- Line 6: The parse-regexp command uses a regular expression to extract the filename from __tag__:__path__ and names it serviceName.
- Line 7: The parse-regexp command uses a regular expression to extract the filename and line number from the caller field and assigns them to the fileName and fileNo fields.
- Line 8: The project-rename command renames the __tag__:__hostname__ field to serviceHost.
- Line 9: The extend command uses the json_extract_scalar function to extract the type field from the schedule object and names it scheduleType.
- Line 10: The project command retains only the final required fields, including the project field that was extracted from Payload.

2. Create a SQL job

Log on to the Realtime Compute for Apache Flink console and click the target workspace.
In the left-side navigation pane, choose Data Development > ETL.
Click Create. In the New Draft dialog box, choose SQL Scripts > Blank Stream Draft, and then click Next.

In the draft editor, enter the following statement to create a temporary table.

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' = '${yourAccessKeyID}',
  'accessKey' = '${yourAccessKeySecret}',
  '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'
  );

The following table describes the parameters in the WITH clause. Replace the example values with your actual values.

Parameter	Description	Example
connector	The connector type. Supported connectors.	sls
endpoint	The SLS internal endpoint. Endpoints.	cn-hangzhou-intranet.log.aliyuncs.com
accessId	Your AccessKey ID. Create an AccessKey.	LTAI****************
accessKey	The AccessKey Secret. Create an AccessKey.	yourAccessKeySecret
starttime	Start time for log consumption.	2025-02-19 00:00:00
project	The SLS project name.	test-project
logstore	The SLS Logstore name.	clb-access-log
query	The SPL statement. Escape string literals with doubled single quotes (`''`) in Flink SQL.	* \| where slbid = ''slb-01''

Select the SQL statement, right-click, and select Running to connect to Simple Log Service.

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-hxxx',
  'accessId' = 'xxx',
  'accessKey' = 'xxx',
  'starttime' = 'xxx',
  'project' = 'xxx',
  'logstore' = 'clb7xxx',
  'query' = '* | prxxx", "__tag__:__hosxxx'
);

3. Run query and view results

In the job editor, enter the following statement to query the data:
```
SELECT * FROM sls_input_complex;
```
Click Debug in the upper-right corner. In the dialog box, select Create new session cluster from the Session Cluster drop-down list and configure as follows.

Set Name to demo-test, Deployment Target to default-queue, Status to RUNNING, and Engine Version to vvr-8.0.11-flink-1.17, then click Create Session Cluster.
In the debug dialog box, select the session cluster you just created, and then click OK.

Note: Debugging with an SLS source table advances the consumer group offset. A deployed job resumes from this new offset.
In the Results tab, you can see that each column in the table corresponds to a field processed by the SPL query.

After you Debug the query, the results table shows the parsed fields. For example, the errorCode column is LogStoreNotExist, the errorMessage column is logstore k8s-event does not exist, and the httpCode column is 404.