local state management • checkpointing state to remote and persistent storage, e.g. a distributed filesystem or a database system • Available state backends in Flink: • In-memory • File system • conf.yaml: # Supported backends are 'jobmanager', 'filesystem', ‘rocksdb' # # state.backend: rocksdb # # Directory for checkpoints filesystem # # state.checkpoints.dir: path/to/checkpoint/folder/ getExecutionEnvironment val checkpointPath: String = ??? // configure path for checkpoints on the remote filesystem val backend = new RocksDBStateBackend(checkpointPath) // configure the state backend env.setS

˓→DynamicTableFactory' in the classpath. Available factory identifiers are: blackhole datagen filesystem print at org.apache.flink.table.factories.FactoryUtil.discoverFactory(FactoryUtil.java:399) at official PyFlink (and also Flink) distribution except the following connectors: blackhole, datagen, filesystem and print. So you need to specify the connector JAR package explicitly when executing PyFlink jobs:

˓→DynamicTableFactory' in the classpath. Available factory identifiers are: blackhole datagen filesystem print at org.apache.flink.table.factories.FactoryUtil.discoverFactory(FactoryUtil.java:399) at official PyFlink (and also Flink) distribution except the following connectors: blackhole, datagen, filesystem and print. So you need to specify the connector JAR package explicitly when executing PyFlink jobs:

is the most volatile and important metric to watch. This is especially true when using Flink’s filesystem statebackend as it keeps all state objects on the JVM Heap. If the size of long-living objects