所有metrics都可以通过Flink提供的REST API来查询。但是，用户可以配置MetricsReporters将metrics发送到外部 系统。Apache Flink为最常用的开箱即用的监控工具(JMX、Prometheus、Datadog、Graphite和InfluxDB)提供报 告程序。有关如何配置报告程序的信息，请参阅Flink的MetricsReporter文档2。 在这篇博客的其余部分中，我们将介绍一些监控Apache

and shrinks • queryable: Flink state can be queried via a REST API Rich Functions • open(Configuration c) • close() • getRuntimeContext() DataStream> input = … DataStream> { private ValueState averageState; @Override public void open (Configuration conf) { ValueStateDescriptor descriptor = new ValueStateDescriptor<>("moving String, String> { private ValueState blocked; @Override public void open(Configuration config) { blocked = getRuntimeContext().getState(new ValueStateDescriptor<>("blocked", Boolean

Boston University 2020 Control: When and how much to adapt? Mechanism: How to apply the re-configuration? 3 • Detect environment changes: external workload and system performance • Identify bottleneck Flink wordcount Every reconfiguration takes ~30s during which the system is unavailable Re-configuration requires state migration with correctness guarantees. ??? Vasiliki Kalavri | Boston University only complete state is migrated Helpers buffer data that cannot yet be safely routed and configuration commands that cannot yet be applied Live state migration ??? Vasiliki Kalavri | Boston University

system execution that yields a system configuration Validity (safety): Termination (liveness): Obtain a valid system configuration A full system configuration is eventually captured A snapshot algorithm system execution that yields a system configuration Validity (safety): Termination (liveness): Obtain a valid system configuration A full system configuration is eventually captured A snapshot algorithm oAk+MBjAE7zAqyOcZ+fNeZ+3lpxiZh9+wfn4BnBTjW4= Epoch-Completeness: Obtain an epoch-complete system configuration 36 ??? Vasiliki Kalavri | Boston University 2020 p1 0 码力 | 81 页 | 13.18 MB | 1 年前

3

factorValues)) } }) 17 Vasiliki Kalavri | Boston University 2020 Configuration options conf/flink-conf.yaml contains the configuration options as a collection of key-value pairs with format key:value

handle object private var lastTempState: ValueState[Double] = _ override def open(parameters: Configuration): Unit = { // create state descriptor val lastTempDescriptor = new ValueStateDescriptor[Double]("lastTemp" rideState;
 private ValueState fareState;
 
 @Override
 public void open(Configuration config) { // initialize the state descriptors here
 rideState = getRuntimeContext()

management: User-defined function registration, dropping, listing, etc. • Executing SQL queries • Job configuration • Python dependency management • Job submission For more details of how to create a TableEnvironment StreamExecutionEnvironment is responsible for: • DataStream Creation • Python dependency management • Job configuration • Job submission [1]: from pyflink.datastream import StreamExecutionEnvironment from pyflink

management: User-defined function registration, dropping, listing, etc. • Executing SQL queries • Job configuration • Python dependency management • Job submission For more details of how to create a TableEnvironment StreamExecutionEnvironment is responsible for: • DataStream Creation • Python dependency management • Job configuration • Job submission [1]: from pyflink.datastream import StreamExecutionEnvironment from pyflink

Boston University 2020 Control: When and how much to adapt? Mechanism: How to apply the re-configuration? 12 • Detect environment changes: external workload and system performance • Identify bottleneck