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

SPL permite aplicar el filtrado de filas y la poda de columnas directamente en el servidor de Simple Log Service (SLS) antes de que los datos lleguen a Flink. Esto reduce la sobrecarga de red y los costos de cómputo.

Contexto

Cuando se utiliza SLS como tabla de origen en Realtime Compute for Apache Flink, el conector consume todos los registros del logstore después de una hora de inicio especificada. Esto provoca dos problemas:

El conector extrae una cantidad excesiva de filas o columnas del origen, lo que incrementa la sobrecarga de la red.
Filtrar datos innecesarios dentro de Flink desperdicia recursos de cómputo.

SPL resuelve esta situación mediante el pushdown de filtros y proyecciones para el conector de SLS. Configure el parámetro query para enviar las condiciones de filtro y los campos proyectados al servidor; así se evita la transferencia y el procesamiento del conjunto completo de datos.

Funcionamiento

Sin una instrucción SPL: Flink extrae todas las filas y columnas de SLS para realizar el cómputo.
Con una instrucción SPL: Si la instrucción SPL incluye filtrado de filas o poda de columnas, Flink extrae únicamente el subconjunto coincidente.

Requisitos previos

Cuenta con activated Simple Log Service y created a Project and a Logstore.

Este ejemplo utiliza un logstore poblado con registros simulados de SLB de capa 7. Los datos contienen más de 10 campos y se generan de forma continua. Entrada de registro de ejemplo:

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

El campo slbid tiene tres valores distintos. Una consulta de 15 minutos muestra que slb-01 y slb-02 tienen recuentos comparables.

Ejecute * | select slbid, count(1) cnt group by slbid en el editor de consultas. Resultados: slb-02 tiene 2,001 entradas, slb-01 tiene 1,986 entradas y lb-uf6qfcvsrouodhvw39oog tiene 4,144 entradas.

Procedimiento

Filtrado de filas: Envíe las condiciones de filtro al servidor de SLS mediante el parámetro query para evitar la transferencia del conjunto completo de datos.

Poda de columnas: Envíe los campos proyectados al servidor de SLS mediante el parámetro query para devolver solo las columnas especificadas.

Filtrado de filas

Paso 1: Crear un trabajo SQL

Inicie sesión en la consola de Realtime Compute for Apache Flink y haga clic en el espacio de trabajo deseado.
En el panel de navegación izquierdo, seleccione Data Development > ETL.
Haga clic en Create. En el cuadro de diálogo New Draft, elija SQL Scripts > Blank Stream Draft y luego haga clic en Next.

Copie el siguiente SQL de tabla temporal en el 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'''
);

Parámetros:

Parámetro	Descripción	Ejemplo
connector	Tipo de conector. Supported connectors.	sls
endpoint	Endpoint interno de SLS. Endpoints.	cn-hangzhou-intranet.log.aliyuncs.com
accessId	Su AccessKey ID. Create an AccessKey.	LTAI
accessKey	AccessKey Secret. Create an AccessKey.	yourAccessKeySecret
starttime	Hora de inicio para el consumo de registros.	2025-02-19 00:00:00
project	Nombre del proyecto de SLS.	test-project
logstore	Nombre del logstore de SLS.	clb-access-log
query	Instrucción SPL. En Flink SQL, escape las cadenas literales duplicando las comillas simples (`''`).	*	where slbid = ''slb-01''

Seleccione la instrucción SQL, haga clic derecho sobre ella y luego haga clic en Running para conectarse a 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'
);

Paso 2: Ejecutar una consulta y consultar los resultados

Ejecute la siguiente consulta de agregación sobre el campo slbid.
```
SELECT slbid, count(1) as slb_cnt FROM sls_input GROUP BY slbid;
```
Haga clic en Debug en la esquina superior derecha. En el cuadro de diálogo, seleccione Create new session cluster en la lista desplegable Session Cluster y configure lo siguiente.

Establezca Name en demo-test, Deployment Target en default-queue, Status en RUNNING y Engine Version en vvr-8.0.11-flink-1.17; luego haga clic en Create Session Cluster.
En el cuadro de diálogo de depuración, seleccione el clúster de sesiones que acaba de crear y luego haga clic en OK.

Nota: La depuración con una tabla de origen SLS avanza el desplazamiento del grupo de consumidores. Un trabajo desplegado se reanudará desde este nuevo desplazamiento.
Los resultados muestran que slbid siempre es slb-01, lo cual confirma que sls_input contiene solo las filas que cumplen la condición de filtro.

El valor de slb_cnt es 185.

Poda de columnas

Paso 1: Crear un trabajo SQL

Inicie sesión en la consola de Realtime Compute for Apache Flink y haga clic en el espacio de trabajo deseado.
En el panel de navegación izquierdo, seleccione Data Development > ETL.
Haga clic en Create. En el cuadro de diálogo New Draft, elija SQL Scripts > Blank Stream Draft y luego haga clic en Next.

Copie el siguiente SQL de tabla temporal en el editor. En comparación con el filtrado de filas, el parámetro query agrega una instrucción de proyección. Los comandos se encadenan con | (similar a las tuberías de Unix), y solo los campos proyectados se recuperan del servidor de SLS.

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__"'
);

Parámetros:

Parámetro	Descripción	Ejemplo
connector	Tipo de conector. Supported connectors.	sls
endpoint	Endpoint interno de SLS. Endpoints.	cn-hangzhou-intranet.log.aliyuncs.com
accessId	Su AccessKey ID. Create an AccessKey.	LTAI
accessKey	AccessKey Secret. Create an AccessKey.	yourAccessKeySecret
starttime	Hora de inicio para el consumo de registros.	2025-02-19 00:00:00
project	Nombre del proyecto de SLS.	test-project
logstore	Nombre del logstore de SLS.	clb-access-log
query	Instrucción SPL. En Flink SQL, escape las cadenas literales duplicando las comillas simples (`''`).	*	where slbid = ''slb-01''

Seleccione la instrucción SQL, haga clic derecho sobre ella y luego haga clic en Running para conectarse a SLS.

Paso 2: Ejecutar una consulta y consultar los resultados

Ejecute la siguiente consulta de agregación sobre el campo slbid.

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

Haga clic en Debug en la esquina superior derecha. En el cuadro de diálogo, seleccione Create new session cluster en la lista desplegable Session Cluster y configure lo siguiente.

Establezca Name en demo-test, Deployment Target en default-queue, Status en RUNNING y Engine Version en vvr-8.0.11-flink-1.17; luego haga clic en Create Session Cluster.
En el cuadro de diálogo de depuración, seleccione el clúster de sesiones que acaba de crear y luego haga clic en OK.

Nota: La depuración con una tabla de origen SLS avanza el desplazamiento del grupo de consumidores. Un trabajo desplegado se reanudará desde este nuevo desplazamiento.
Los resultados son similares al caso de uso de filtrado de filas.

Nota
A diferencia del filtrado de filas, la poda de columnas devuelve únicamente los campos especificados, lo que reduce aún más el tráfico de red.

El valor de slb_cnt es 185.