How to implement row filtering and column pruning in Alibaba Cloud Flink SQL using SPL - Simple Log Service

Background

When you use SLS as a source table in Realtime Compute for Apache Flink, the connector consumes all Logstore records after a specified start time. This causes two issues:

The connector pulls excessive rows or columns from the source, increasing network overhead.
Filtering unnecessary data in Flink wastes compute resources.

SPL addresses this with filter pushdown and projection pushdown for the SLS connector. Configure the query parameter to push filter conditions and projected fields to the server, avoiding full-dataset transfer and computation.

How it works

Without an SPL statement: Flink pulls all rows and columns from SLS for computation.
With an SPL statement: If the SPL statement includes row filtering or column pruning, Flink pulls only the matching subset.

Prerequisites

You have activated Simple Log Service and created a Project and a Logstore.

This example uses a Logstore populated with simulated Layer-7 SLB logs. The data contains more than 10 fields and is generated continuously. Sample log entry:

{
  "__source__": "127.0.0.1",
  "__tag__:__receive_time__": "1706531737",
  "__time__": "1706531727",
  "__topic__": "slb_layer7",
  "body_bytes_sent": "3577",
  "client_ip": "114.137.XXX.XXX",
  "host": "www.pi.mock.com",
  "http_host": "www.cwj.mock.com",
  "http_user_agent": "Mozilla/5.0 (Windows NT 6.2; rv:22.0) Gecko/20130405 Firefox/23.0",
  "request_length": "1662",
  "request_method": "GET",
  "request_time": "31",
  "request_uri": "/request/path-0/file-3",
  "scheme": "https",
  "slbid": "slb-02",
  "status": "200",
  "upstream_addr": "42.63.XXX.XXX",
  "upstream_response_time": "32",
  "upstream_status": "200",
  "vip_addr": "223.18.XX.XXX"
}

The slbid field has three distinct values. A 15-minute query shows that slb-01 and slb-02 have comparable counts.

Run * | select slbid, count(1) cnt group by slbid in the query editor. Results: slb-02 has 2,001 entries, slb-01 has 1,986 entries, and lb-uf6qfcvsrouodhvw39oog has 4,144 entries.

Procedure

Row filtering: Push filter conditions to the SLS server via the query parameter, preventing full-dataset transfer.

Column pruning: Push projected fields to the SLS server via the query parameter, returning only specified columns.

Row filtering

Step 1: Create an 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.

Copy the following temporary table SQL into the editor.

CREATE TEMPORARY TABLE sls_input(
  request_uri STRING,
  scheme STRING,
  slbid STRING,
  status STRING,
  `__topic__` STRING METADATA VIRTUAL,
  `__source__` STRING METADATA VIRTUAL,
  `__timestamp__` STRING METADATA VIRTUAL,
   __tag__ MAP<VARCHAR, VARCHAR> METADATA VIRTUAL,
  proctime as PROCTIME()
) WITH (
  'connector' = 'sls',
  'endpoint' ='cn-hangzhou-intranet.log.aliyuncs.com',
  'accessId' = 'yourAccessKeyID',
  'accessKey' = 'yourAccessKeySecret',
  'starttime' = '2025-02-19 00:00:00',
  'project' ='test-project',
  'logstore' ='clb-access-log',
  'query' = '* | where slbid = ''slb-01'''
);

Parameters:

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 it, and click Running to connect to SLS.

CREATE TEMPORARY TABLE sls_input(
  request_uri STRING,
  scheme STRING,
  slbid STRING,
  status STRING,
  `__topic__` STRING METADATA VIRTUAL,
  `__source__` STRING METADATA VIRTUAL,
  `__timestamp__` STRING METADATA VIRTUAL,
  __tag__ MAP<VARCHAR, VARCHAR> METADATA VIRTUAL,
  proctime as PROCTIME()
) WITH (
  'connector' = 'sls',
  'endpoint' = 'cn-haxxx',
  'accessId' = 'xxx',
  'accessKey' = 'xxx',
  'starttime' = '202xxx',
  'project' = 'test-pxxx',
  'logstore' = 'clb-axxx',
  'query' = '*|whexxx'
);

Step 2: Run a query and view results

Run the following aggregate query on the slbid field.

SELECT slbid, count(1) as slb_cnt FROM sls_input GROUP BY slbid;

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.
The results show that slbid is always slb-01, confirming that sls_input contains only rows matching the filter condition.

The slb_cnt value is 185.

Column pruning

Step 1: Create an 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.

Copy the following temporary table SQL into the editor. Compared to row filtering, the query parameter adds a projection statement. Commands are piped with | (like Unix pipes), and only the projected fields are retrieved from the SLS server.

CREATE TEMPORARY TABLE sls_input_project(
  request_uri STRING,
  scheme STRING,
  slbid STRING,
  status STRING,
  `__topic__` STRING METADATA VIRTUAL,
  `__source__` STRING METADATA VIRTUAL,
  `__timestamp__` STRING METADATA VIRTUAL,
   __tag__ MAP<VARCHAR, VARCHAR> METADATA VIRTUAL,
  proctime as PROCTIME()
) WITH (
  'connector' = 'sls',
  'endpoint' ='cn-hangzhou-intranet.log.aliyuncs.com',
  'accessId' = 'yourAccessKeyID',
  'accessKey' = 'yourAccessKeySecret',
  'starttime' = '2025-02-19 00:00:00',
  'project' ='test-project',
  'logstore' ='clb-access-log',
  'query' = '* | where slbid = ''slb-01'' | project request_uri, scheme, slbid, status, __topic__, __source__, "__tag__:__receive_time__"'
);

Parameters:

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 it, and click Running to connect to SLS.

Step 2: Run a query and view results

Run the following aggregate query on the slbid field.

SELECT slbid, count(1) as slb_cnt FROM sls_input_project GROUP BY slbid;

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.
The results are similar to the row filtering use case.

Note
Unlike row filtering, column pruning returns only the specified fields, further reducing network traffic.

The slb_cnt value is 185.