Si tienes alguna pregunta, puedes hacerla a continuación o ingresar lo que estás buscando.
Soy nuevo en Apache Flink y estoy tratando de entender cómo se maneja el concepto de EventTime y Windowing en Flink. Aquí está mi escenario: Tengo un programa que se ejecuta como un hilo y crea un archivo con 3 campos cada segundo, siendo el tercer campo la marca de . . . Read more
Tengo un objeto Row de Flink y nombres de columnas. Este objeto Row puede ser accedido mediante nombres de campo o índice. Quiero enviarlo a Kafka en formato JSON utilizando el productor de Kafka vanilla de Flink. ¿Cómo puedo hacerlo? ¿Se requiere un esquema JSON de destino para enviarlo a . . . Read more
Digamos que tengo una función de proceso como esta (con un estado de backend de rocksdb): public class Test extends KeyedProcessFunction<…>{ private transient ValueState<integer> …; … @Override public void open(Configuration parameters) throws Exception { StateTtlConfig ttlConfig = StateTtlConfig .newBuilder(Time.minutes(10)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .cleanupInRocksdbCompactFilter(1000) .build(); ValueStateDescriptor<integer> testDescr = new ValueStateDescriptor<integer>( “test”, TypeInformation.of(Integer.class) . . . Read more
Estoy tratando de construir un sistema con procesamiento de transmisión en tiempo real con Flink, teniendo S3 como origen y Elastic como destino. He probado 4 casos en total para los puntos de control. Exactly_Once con puntos de control alineados. Exactly_Once con puntos de control desalineados. AtLeastOnce con un máximo . . . Read more
En la API de transmisión de datos (datastream api) tenemos un argumento llamado withInactivityInterval, que nos permite establecer el intervalo disponible hasta que se cierre el archivo. Sin embargo, en la API de tabla (table api) no existe algo similar. Si nuestra transmisión de datos hacia la tabla se suspende . . . Read more