import Types from pyflink.datastream import StreamExecutionEnvironment from pyflink.table import StreamTableEnvironment # create a StreamExecutionEnvironment which is the entry point of 'DataStream' 'DataStream' program. env = StreamExecutionEnvironment.get_execution_environment() t_env = StreamTableEnvironment.create(env) ds = env.from_collection([(1, 'Hi'), (2, 'Hello'), 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' 'DataStream' program. env = StreamExecutionEnvironment.get_execution_environment() t_env = StreamTableEnvironment.create(env) ds = env.from_collection([(1, 'Hi'), (2, 'Hello'), 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) temp: Double) object MaxSensorReadings { def main(args: Array[String]) { val env = StreamExecutionEnvironment.getExecutionEnvironmentValue( val sensorData = env.addSource(new SensorSource() Double) object MaxSensorReadings { def main(args: Array[String]) { val env = StreamExecutionEnvironment.getExecutionEnvironmentValue(new SensorSource) val sensorData = env.addSource(new

state.checkpoints.dir: path/to/checkpoint/folder/ ## I n your Flink program: val env = StreamExecutionEnvironment.getExecutionEnvironment val checkpointPath: String = ???? // configure path for checkpoints state that belongs to the record it currently processes. ## Java example StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setStreamTimeCharacteristic(TimeCharacteristic

readings object MaxSensorReadings { def main(args: Array[String]) { val env = StreamExecutionEnvironment.getExecutionEnvironment(val sensorData = env.addSource(new SensorSource)) val the StreamExecutionEnvironment: object AverageSensorReadings { def main(args: Array[String]) { // set up the streaming execution environment val env = StreamExecutionEnvironment.ge

g. if the stream contains special records that encode watermark information. val env = StreamExecutionEnvironment.getExecutionEnvironment env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) // assign record timestamp r.timestamp } ## Using a watermark assigner val env = StreamExecutionEnvironment.getExecutionEnvironment // set the event time characteristic env.setStreamTimeCharac

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

env = StreamExecutionEnvironment.getExecutionEnvironment // set checkpointing interval to 10 seconds env.enableCheckpointing(10000L) // get the CheckpointConfig from the StreamExecutionEnvironment val

8efaa36f86af3a412343a176be3c108e/p39_1.jpg) ## Setting the max parallelism val env = StreamExecutionEnvironment.getExecutionEnvironment // set the maximum parallelism for this application env.setMaxParallelism(512)

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