Realtime Compute for Apache Flink:Parámetros clave

Notas de uso

Antes de iniciar un trabajo, si no establece explícitamente este parámetro en la sección Parameters, se utiliza la memoria administrada para almacenar datos en búfer en modo mini-batch. Cualquiera de las siguientes condiciones activa el cálculo final y la salida de datos:

• Se recibe un mensaje de marca de agua del operador MiniBatchAssigner.

• La memoria administrada está llena.

• Antes de que comience un checkpoint.

• El trabajo se detiene.

Valores disponibles

• -1 (predeterminado): Indica que se utiliza la memoria administrada para almacenar datos en búfer.

• Otros valores Long negativos: Igual que la configuración predeterminada.

• Otros valores Long positivos: Indica que se utiliza la memoria heap para almacenar datos en búfer. Cuando el número de registros en búfer alcanza este valor (N), el sistema activa automáticamente la operación de salida.

Ejemplo

1. Cree un trabajo de streaming SQL, copie el siguiente código SQL y deploy el trabajo.

   CREATE TEMPORARY TABLE s1 (
     a INT,
     b INT,
     ts TIMESTAMP(3),
     PRIMARY KEY (a) NOT ENFORCED,
     WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
   ) WITH (
     'connector'='datagen',
     'rows-per-second'='1',
     'fields.ts.kind'='random',
     'fields.ts.max-past'='5s',
     'fields.b.kind'='random',
     'fields.b.min'='0',
     'fields.b.max'='10'
   );
   CREATE TEMPORARY TABLE sink (
     a INT,
     b BIGINT,
     PRIMARY KEY (a) NOT ENFORCED
   ) WITH (
     'connector'='print'
   );
   INSERT INTO sink SELECT a, sum(b) FROM s1 GROUP BY a;

2. En la pestaña Configuration de la página Deployments, en el campo Other Configuration de la sección Parameters, establezca los parámetros table.exec.mini-batch.enabled: true y table.exec.mini-batch.allow-latency: 2s, y no configure table.exec.mini-batch.size para usar su valor predeterminado de -1.

3. Start el trabajo. En la pestaña Status, la topología del trabajo incluye los operadores MiniBatchAssigner, LocalGroupAggregate y GlobalGroupAggregate.

Notas de uso

• Este parámetro solo es efectivo en VVR 8.0.8 y versiones posteriores. En versiones anteriores a VVR 8.0.8, el comportamiento equivale a establecer este parámetro en false.

• Al cambiar el valor de false a true, la cantidad de datos enviados aguas abajo desde los operadores MinibatchGlobalAgg y MinibatchAgg puede aumentar, lo que ejerce mayor presión sobre los operadores descendentes.

Valores disponibles

• false (predeterminado): Cuando state TTL está habilitado, los operadores MinibatchGlobalAgg y MinibatchAgg no emiten datos aguas abajo si el resultado de la agregación no cambia.

• true: Cuando state TTL está habilitado, los operadores MinibatchGlobalAgg y MinibatchAgg siguen emitiendo registros actualizados (duplicados) aguas abajo incluso si el resultado de la agregación no cambia.

Ejemplo

1. Cree un trabajo de streaming SQL, copie el siguiente código SQL y deploy el trabajo.

   create temporary table src(
       a int,
       b string
   ) with (
       'connector' = 'datagen',
       'rows-per-second' = '10',
       'fields.a.min' = '1',
       'fields.a.max' = '1',
       'fields.b.length' = '3'
   );
   create temporary table snk(
       a int,
       max_length_b bigint
   ) with (
       'connector' = 'blackhole'
   );
   insert into snk select a, max(CHAR_LENGTH(b)) from src group by a; 

2. En la sección Other Configuration del área Parameters en la pestaña Configuration de la página Deployments, establezca los parámetros table.exec.mini-batch.enabled: true y table.exec.mini-batch.allow-latency: 2s para habilitar la optimización Minibatch Aggregate.

3. Start el trabajo. En la pestaña Status, el trabajo incluye un operador MinibatchGlobalAggregate. Haga clic en el signo "+" del operador para ver que el operador GlobalGroupAggregate no envía datos aguas abajo cuando el resultado de la agregación no cambia.

   El operador muestra RecordsIn como 19 y RecordsOut como 1, lo que significa que 19 registros de entrada produjeron solo 1 salida agregada.

4. Detenga el trabajo y agregue el parámetro table.exec.agg.mini-batch.output-identical-enabled: true a Other Configuration en la sección Parameters de la página Configuration de la página Deployments.

5. Start el trabajo. En la pestaña Status, puede ver que el trabajo incluye un operador MinibatchGlobalAggregate. Haga clic en el signo "+" del operador para observar que el operador GlobalGroupAggregate ahora envía datos aguas abajo incluso cuando el resultado de la agregación no cambia. Después de reiniciar el trabajo, la pestaña Status muestra que tanto RecordsIn como RecordsOut del operador GlobalGroupAggregate son 94. Esto indica que, con table.exec.agg.mini-batch.output-identical-enabled: true habilitado, el operador envía datos aguas abajo aunque el resultado de la agregación no cambie.

Casos de uso

• Utilice esta optimización para conservar el orden de procesamiento por clave en una unión de tabla de dimensiones cuando el flujo tiene pocos mensajes con la misma clave de actualización a lo largo del tiempo.

• En un flujo de Change Data Capture (CDC) con clave primaria, realiza el enriquecimiento de datos con una unión de tabla de dimensiones y escribe en un sink donde la clave primaria del sink coincide con la clave primaria de la fuente. La clave de unión para la tabla de dimensiones es diferente de la clave primaria, y la clave de unión del lado de la dimensión es la clave primaria. Esta optimización realiza un shuffle por la clave primaria de CDC, que se deriva como la clave de actualización. En comparación con la habilitación de la optimización SHUFFLE_HASH para el mismo escenario, evita generar un operador SinkMaterializer antes del sink con un paralelismo mayor. Esto evita posibles problemas de rendimiento de ese operador, especialmente el gran estado que puede acumularse durante ejecuciones largas. Para obtener información sobre SinkUpsertMaterializer, consulte Recommendations.

• La clave de unión para la tabla de dimensiones es diferente de la clave primaria, la clave de unión del lado de la dimensión es la clave primaria y le sigue un operador rank. Esta optimización realiza un shuffle por la clave primaria de CDC, que se deriva como la clave de actualización. En comparación con la habilitación de la optimización SHUFFLE_HASH para el mismo escenario, evita que UpdateFastRank se degrade a RetractRank. Para obtener información sobre cómo optimizar RetractRank a UpdateFastRank, consulte TopN optimization techniques.

Notas de uso

• Si el flujo no tiene clave de actualización, se utiliza toda la fila como clave.

• El throughput disminuye si la misma clave de actualización se modifica frecuentemente en un período corto, ya que los registros para la misma clave de actualización se procesan en orden estricto.

• En comparación con la unión de tabla de dimensiones asíncrona original, el modo Key-Ordered introduce estado con clave. Habilitar o deshabilitar este modo afecta la compatibilidad del estado.

• Esta función solo surte efecto para VVR 8.0.10 y versiones posteriores cuando la entrada de la unión de tabla de dimensiones es un flujo de actualización y configura table.exec.async-lookup.output-mode='ALLOW_UNORDERED' y table.exec.async-lookup.key-ordered-enabled='true'.

Valores disponibles

• false (predeterminado): Deshabilita el modo Key-Ordered.

• true: Habilita el modo Key-Ordered.

Ejemplo

1. El siguiente ejemplo utiliza una unión de tabla de dimensiones Hologres asíncrona. Cree un trabajo de streaming SQL, copie el siguiente código SQL y deploy el trabajo.

   Para obtener más información sobre el conector Hologres, consulte Hologres.

   create TEMPORARY table bid_source(
     auction  BIGINT,
     bidder  BIGINT,
     price  BIGINT,
     channel  VARCHAR,
     url  VARCHAR,
     dateTime  TIMESTAMP(3),
     extra  VARCHAR,
     proc_time as proctime(),
     WATERMARK FOR dateTime AS dateTime - INTERVAL '4' SECOND
   ) with (
     'connector' = 'kafka',  -- A non-insert-only stream connector
     'topic' = 'user_behavior',
     'properties.bootstrap.servers' = 'localhost:9092',
     'properties.group.id' = 'testGroup',
     'scan.startup.mode' = 'earliest-offset',
     'format' = 'csv'
   );
   CREATE TEMPORARY TABLE users (
       user_id STRING PRIMARY KEY NOT ENFORCED, -- Define the primary key
       user_name VARCHAR(255) NOT NULL, 
       age INT NOT NULL
   ) WITH (
       'connector' = 'hologres',    -- A connector that supports asynchronous lookup
       'async' = 'true',
       'dbname' = 'holo db name', --The name of your Hologres database
       'tablename' = 'schema_name.table_name', --The name of the Hologres table to receive data
       'username' = 'access id', --The AccessKey ID of your Alibaba Cloud account
       'password' = 'access key', --The AccessKey Secret of your Alibaba Cloud account
       'endpoint' = 'holo vpc endpoint', --The VPC endpoint of your Hologres instance
   );
   CREATE TEMPORARY TABLE bh ( 
       auction  BIGINT,
       age int
   ) WITH (
       'connector' = 'blackhole'
   );
   insert into bh
   SELECT
       bid_source.auction,
       u.age
   FROM bid_source
       JOIN users FOR SYSTEM_TIME AS OF bid_source.proc_time AS u
       ON bid_source.channel = u.user_id;
   

2. En la página Deployments, en la pestaña Configuration, en la sección Other Configuration del área Parameters, establezca los parámetros table.exec.async-lookup.output-mode='ALLOW_UNORDERED' y table.exec.async-lookup.key-ordered-enabled='true'.

3. Start el trabajo. En la pestaña Status, puede ver que el atributo asíncrono del trabajo es KEY_ORDERED:true.

table.optimizer.window-join-enabled

Este parámetro controla si se habilitan las operaciones de window join. Cuando está habilitado, Flink optimiza el plan de ejecución correspondiente como un window join. Para ventanas pequeñas, esto reduce la sobrecarga de estado y mejora el rendimiento. En comparación con un regular join, un window join también puede evitar emitir mensajes de actualización a los operadores descendentes, lo cual es útil para casos de uso que requieren uniones con condiciones de ventana temporal pequeña.