import Types from pyflink.datastream import StreamExecutionEnvironment from pyflink.table import StreamTableEnvironment # create a StreamExecutionEnvironment which is the entry point of `DataStream` program program. env = StreamExecutionEnvironment.get_execution_environment() t_env = StreamTableEnvironment.create(env) ds = env.from_collection([(1, 'Hi'), (2, 'Hello')], type_info=Types.ROW_NAMED( ["id", "data"] you can refer to the latest version of PyFlink DataStream API doc StreamExecutionEnvironment Creation StreamExecutionEnvironment is the entry point and central concept for creating DataStream API programs

import Types from pyflink.datastream import StreamExecutionEnvironment from pyflink.table import StreamTableEnvironment # create a StreamExecutionEnvironment which is the entry point of `DataStream` program program. env = StreamExecutionEnvironment.get_execution_environment() t_env = StreamTableEnvironment.create(env) ds = env.from_collection([(1, 'Hi'), (2, 'Hello')], type_info=Types.ROW_NAMED( ["id", "data"] you can refer to the latest version of PyFlink DataStream API doc StreamExecutionEnvironment Creation StreamExecutionEnvironment is the entry point and central concept for creating DataStream API programs

Double)
 
 object MaxSensorReadings { def main(args: Array[String]) {
 val env = StreamExecutionEnvironment.getExecutionEnvironment
 val sensorData = env.addSource(new SensorSource)
 val Double)
 
 object MaxSensorReadings { def main(args: Array[String]) {
 val env = StreamExecutionEnvironment.getExecutionEnvironment
 val sensorData = env.addSource(new SensorSource)
 val Double)
 
 object MaxSensorReadings { def main(args: Array[String]) {
 val env = StreamExecutionEnvironment.getExecutionEnvironment
 val sensorData = env.addSource(new SensorSource)
 val

state.checkpoints.dir: path/to/checkpoint/folder/ In your Flink program: val env = StreamExecutionEnvironment.getExecutionEnvironment val checkpointPath: String = ??? // configure path for checkpoints processes. 19 Keyed state scope Vasiliki Kalavri | Boston University 2020 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 env.setStreamTimeCharacteristic(TimeCharacteristic

University 2020 object MaxSensorReadings { def main(args: Array[String]) {
 val env = StreamExecutionEnvironment.getExecutionEnvironment
 val sensorData = env.addSource(new SensorSource) 
 define time when you are creating windows. The time characteristic is a property of the StreamExecutionEnvironment: Configuring a time characteristic 4 object AverageSensorReadings { def main(args: main(args: Array[String]) { // set up the streaming execution environment val env = StreamExecutionEnvironment.getExecutionEnvironment // use event time for the application env.setStreamTime

if the stream contains special records that encode watermark information. val env = StreamExecutionEnvironment.getExecutionEnvironment env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) // record timestamp r.timestamp } } Vasiliki Kalavri | Boston University 2020 val env = StreamExecutionEnvironment.getExecutionEnvironment // set the event time characteristic env.setStreamTimeCharact

• Event time • Ingestion time • Processing time (default) final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setStreamTimeCharacteristic(TimeCharacteristic

Kalavri | Boston University 2020 32 ??? Vasiliki Kalavri | Boston University 2020 val env = StreamExecutionEnvironment.getExecutionEnvironment // set the maximum parallelism for this application env.setMaxParallelism(512)

temp: Double)
 
 object MaxSensorReadings { def main(args: Array[String]) {
 val env = StreamExecutionEnvironment.getExecutionEnvironment
 val sensorData = env.addSource(new SensorSource)
 val

optimizations ??? Vasiliki Kalavri | Boston University 2020 Task chaining: Fusion in Flink 31 StreamExecutionEnvironment .disableOperatorChaining() val input: DataStream[X] = ... val result: DataStream[Y] =