Lectura de datos

Este tema explica cómo leer datos de Fluss en modo streaming y en modo batch.

Modo streaming

Fluss admite varios consumption modes. Para una tabla con clave primaria, el modo de consumo predeterminado es full, que primero consume todos los datos completos y después los incrementales.

Inicie sesión en la consola de Realtime Compute for Apache Flink.
En la columna Actions del espacio de trabajo deseado, haga clic en Console.
En el panel de navegación izquierdo, haga clic en Development > ETL.
Haga clic en + y luego en New Blank Stream Draft. Introduzca un Name, seleccione una Engine Version y haga clic en Create.

Escriba el código del trabajo.

SELECT * FROM `fluss-catalog`.`my_db`.`my_table`;

# Use the earliest consumption mode
SELECT * FROM my_table /*+ OPTIONS('scan.startup.mode' = 'earliest') */;

Modo batch

Tanto las tablas con clave primaria como las tablas de registro admiten lecturas con LIMIT para devolver rápidamente un número específico de resultados y simplificar las vistas previas de datos.
Las tablas con clave primaria y las tablas de registro (versión 0.9-ali-1.0 y posteriores) admiten consultas de agregación con COUNT(*).
Las tablas con clave primaria también permiten búsquedas puntuales mediante coincidencias exactas de claves primarias y de índice.

Inicie sesión en la consola de Realtime Compute for Apache Flink.
En la columna Actions del espacio de trabajo deseado, haga clic en Console.
En el panel de navegación izquierdo, haga clic en Development > ETL.
Haga clic en + y luego en New Blank Batch Draft. Introduzca un Name, seleccione una Engine Version y un SQL Dialect, y haga clic en Create.

Escriba el código del trabajo.

# LIMIT read
SELECT * FROM `fluss-catalog`.`my_db`.`my_table` LIMIT 10;

# Point lookup (primary key tables only)
SELECT * FROM `fluss-catalog`.`my_db`.`my_pk_tbl` WHERE shop_id = 10000 and user_id = 123456;

# Aggregate query
SELECT count(*) FROM `fluss-catalog`.`my_db`.`my_log_table`;

Optimización de consultas

Pushdown de predicados

Para las condiciones de filtro escritas en una cláusula SELECT ... WHERE, Flink transfiere automáticamente los predicados correspondientes al servidor de Fluss. El servidor evalúa estos predicados con base en las estadísticas de columnas de cada RecordBatch, lo que le permite omitir bloques de datos que no coinciden y reducir la transferencia de red y la sobrecarga de deserialización.

Operadores de filtro

Operadores de comparación: =, <>, >, >=, < y <=
IN (...)
IS NULL, IS NOT NULL
BETWEEN ... AND ...
LIKE: Admite coincidencia de prefijo ('abc%'), de sufijo ('%abc') y de contenido ('%abc%').
Operadores lógicos: AND, OR

Limitaciones

Aplica únicamente a tablas de registro.
Todas las columnas referenciadas en la condición de filtro deben tener habilitadas las estadísticas de columna (table.statistics.columns).
Si las estadísticas no están habilitadas para alguna columna referenciada, la condición de filtro no se transfiere.
Este filtrado de grano grueso se basa en estadísticas a nivel de lote y puede generar falsos positivos (devolver datos adicionales). Esto no afecta la semántica de los resultados de la consulta.

Ejemplos

Paso 1: Crear una tabla y habilitar estadísticas de columna

CREATE TABLE sensor_log (
  sensor_id BIGINT,
  temperature DOUBLE,
  humidity DOUBLE,
  event_time TIMESTAMP(3)
) WITH (
  'table.statistics.columns' = 'sensor_id,temperature'
);

Paso 2: Consultar datos mediante una cláusula WHERE

SELECT sensor_id, temperature, event_time
FROM sensor_log
WHERE temperature > 30.0 AND sensor_id = 1001;

Paso 3: Verificar el pushdown de predicados con EXPLAIN

EXPLAIN SELECT sensor_id, temperature, event_time
FROM sensor_log
WHERE temperature > 30.0 AND sensor_id = 1001;

En la salida de EXPLAIN, la cláusula filter=[...] del nodo TableSourceScan indica que la condición de filtro se ha transferido al servidor de Fluss.

Nota

Es un comportamiento esperado que el nodo Calc en la salida de EXPLAIN conserve la condición de filtro. El cliente de Flink filtra nuevamente los datos recibidos como medida de seguridad.

Para obtener más detalles sobre los principios y la configuración del pushdown de predicados, consulte Table Properties.

Consulta de datos

Nota

Tanto las tablas con clave primaria como las tablas de registro admiten lecturas con LIMIT para devolver rápidamente un número específico de resultados y simplificar las vistas previas de datos.
Las tablas con clave primaria y las tablas de registro (versión 0.9-ali-1.0 y posteriores) admiten consultas de agregación con COUNT(*).
Las tablas con clave primaria también permiten búsquedas puntuales mediante coincidencias exactas de claves primarias y de índice.

Inicie sesión en la consola de Realtime Compute for Apache Flink.
En la columna Actions del espacio de trabajo deseado, haga clic en Console.

En el panel de navegación izquierdo, haga clic en Scripts > New Script.

SELECT * FROM `fluss-catalog`.`my_db`.`customer` 
WHERE c_custkey = 1;

Seleccione el bloque de código y haga clic en Run para ver los resultados.

Modos de consumo

Modo de consumo	Tabla con clave primaria	Tabla de registro	Casos de uso
full (predeterminado)	Primero consume todos los datos completos y después los incrementales.	Consume datos desde el desplazamiento más antiguo.	Ideal para escenarios que requieren procesar todos los datos, como análisis de datos completos.
earliest	Consume datos a partir del desplazamiento de binlog más antiguo.	Consume datos desde el desplazamiento más antiguo.	Adecuado para escenarios que requieren procesar datos históricos de cambios, como la carga inicial de datos.
latest	Consume datos a partir del desplazamiento de binlog más reciente.	Consume datos desde el desplazamiento más reciente.	Recomendado cuando solo se necesitan los datos más recientes, como en monitoreo en tiempo real.
timestamp	Consume datos de binlog a partir de una hora específica indicada por la opción `scan.startup.timestamp`.	Consume datos a partir de una hora específica indicada por la opción `scan.startup.timestamp`.	Diseñado para recuperación de datos o retroceso basado en un punto específico en el tiempo.

Ejemplos de SQL

// full mode (default)
SELECT * FROM my_table ;

// earliest consumption mode
SELECT * FROM my_table /*+ OPTIONS('scan.startup.mode' = 'earliest') */;

// latest consumption mode
SELECT * FROM my_table /*+ OPTIONS('scan.startup.mode' = 'latest') */;

// Use a timestamp
SELECT * FROM my_table
/*+ OPTIONS('scan.startup.mode' = 'timestamp',
'scan.startup.timestamp' = '1678883047356') */;

// Use a time string
SELECT * FROM my_table
/*+ OPTIONS('scan.startup.mode' = 'timestamp',
'scan.startup.timestamp' = '2023-12-09 23:09:12') */;