本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # Apache Flink 设置 Managed Service for Apache Flink 是 Apache Flink 框架的实现。Managed Service for Apache Flink 使用本节中描述的默认值。其中一些值可以由 Managed Service for Apache Flink 应用程序在代码中设置，而其他值则无法更改。 使用本节中的链接详细了解 Apache Flink 设置以及哪些设置是可以修改的。 **Topics** + [Apache Flink 配置](#apache-flink-configuration) + [状态后端](#reference-defaults-state-backend) + [检查点](#reference-defaults-checkpoint) + [保存点](#reference-defaults-savepoint) + [堆大小](#reference-defaults-heap) + [缓冲区消胀](#reference-defaults-buffer-debloating) + [可修改的 Flink 配置属性](reference-modifiable-settings.md) + [编程 Flink 配置属性](programmatic-configuration.md) + [查看已配置的 Flink 属性](viewing-modifiable-settings.md) ## Apache Flink 配置 Managed Service for Apache Flink 提供了默认 Flink 配置，包括 Apache Flink 为大多数属性推荐的值和一些基于常见应用程序配置文件的推荐值。有关 Flink 配置的更多信息，请参阅 Flink [配置](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/)。服务提供的默认配置适用于大多数应用程序。但是，为调整 Flink 配置属性以提高某些具有高并行度、高内存和状态使用率的应用程序的性能，或者在 Apache Flink 中启用新的调试功能，则可以通过请求支持案例来更改某些属性。有关更多信息，请参阅 [AWS 支持中心](https://console.aws.amazon.com/support/home#/)。您可以使用 [Apache Flink 控制面板检查应用程序](https://docs.aws.amazon.com/managed-flink/latest/java/how-dashboard.html)的当前配置。 ## 状态后端 Managed Service for Apache Flink 将瞬态数据存储在状态后端。适用于 Apache Flink 的托管服务使用 **Roc DBState** ks 后端。调用 `setStateBackend` 以设置不同的后端无效。 我们在状态后端上启用以下功能： + 增量状态后端快照 + 异步状态后端快照 + 本地检查点恢复 有关状态后端的更多信息，请参阅 Apache Flink 文档中的[状态后端](https://nightlies.apache.org/flink/flink-docs-release-1.19/ops/state/state_backends.html)。 ## 检查点 Managed Service for Apache Flink使用具有以下值的默认检查点配置。其中一些值可以使用进行更改[CheckpointConfiguration](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CheckpointConfiguration.html)。您必须将 `CheckpointConfiguration.ConfigurationType` 设置为 适用于 Managed Service for Apache Flink 的 `CUSTOM`，才能使用修改的检查点值。 **** | 设置 | 是否可以修改？ | 操作方法 | 默认值 | | --- | --- | --- | --- | | CheckpointingEnabled | 可修改 | [创建应用程序](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CreateApplication.html) [更新应用程序](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html) [CloudFormation](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/AWS_KinesisAnalyticsV2.html) | True | | CheckpointInterval | 可修改 | [创建应用程序](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CreateApplication.html) [更新应用程序](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html) [CloudFormation](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/AWS_KinesisAnalyticsV2.html) | 60000 | | MinPauseBetweenCheckpoints | 可修改 | [创建应用程序](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CreateApplication.html) [更新应用程序](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html) [CloudFormation](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/AWS_KinesisAnalyticsV2.html) | 5000 | | 未对齐的检查点 | 可修改 | [支持案例](https://console.aws.amazon.com/support/home#/) | False | | 并发检查点数 | 不能修改 | 不适用 | 1 | | 检查点模式 | 不能修改 | 不适用 | 恰好一次 | | 检查点保留策略 | 不能修改 | 不适用 | 失败时 | | 检查点超时 | 不能修改 | 不适用 | 60 分钟 | | 保留的最大检查点数 | 不能修改 | 不适用 | 1 | | 检查点和保存点位置 | 不能修改 | 不适用 | 我们将持久的检查点和保存点数据存储到服务拥有的 S3 存储桶中。 | ## 保存点 默认情况下，从保存点中还原时，恢复操作尝试将保存点的所有状态映回到用于还原的程序。如果删除了一个操作符，默认情况下，从包含与缺少的操作符对应的数据的保存点中还原将失败。通过将应用程序的*AllowNonRestoredState*参数设置为，可以允许操作成功`true`。[FlinkRunConfiguration](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_FlinkRunConfiguration.html)这样，恢复操作就可以跳过无法映射到新程序的状态。 有关更多信息，请参阅 [Apache Flink 文档](https://nightlies.apache.org/flink/flink-docs-release-1.15/)中的 [Allowing Non-Restored State](https://nightlies.apache.org/flink/flink-docs-release-1.15/ops/state/savepoints.html#allowing-non-restored-state)（允许未还原状态）。 ## 堆大小 Managed Service for Apache Flink为每个 KPU 分配 3 GiB JVM 堆，并为本机代码分配保留 1 GiB。有关增加应用程序容量的信息，请参阅[实施 Managed Service for Apache Flink 中的应用程序扩展](how-scaling.md)。 有关 JVM 堆大小的更多信息，请参阅[ Apache Flink 文档](https://nightlies.apache.org/flink/flink-docs-release-1.15/)中的[配置](https://nightlies.apache.org/flink/flink-docs-release-1.15/ops/config.html)。 ## 缓冲区消胀 缓冲区消胀可以帮助具有高背压的应用。如果您的应用程序遇到检查点/保存点失败，则启用此功能可能会很有用。为此，请申请[支持案例](https://console.aws.amazon.com/support/home#/)。 有关更多信息，请参阅 [Apache Flink 文档](https://nightlies.apache.org/flink/flink-docs-release-1.15/)中的[缓冲区消胀机制](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/memory/network_mem_tuning/#the-buffer-debloating-mechanism)。 # 可修改的 Flink 配置属性 以下是您可以使用[支持案例](https://console.aws.amazon.com/support/home#/)修改的 Flink 配置设置。通过指定应用程序前缀，您可以一次修改多个属性，也可以同时修改多个应用程序的属性。如果您要修改此列表之外的其他 Flink 配置属性，根据您的具体情况指定确切的属性。 ## 重新启动策略 从 Flink 1.19 及更高版本开始，我们默认使用 `exponential-delay` 重新启动策略。默认情况下，所有以前的版本都使用 `fixed-delay` 重新启动策略。 `restart-strategy:` `restart-strategy.fixed-delay.delay:` `restart-strategy.exponential-delay.backoff-muliplier:` `restart-strategy.exponential-delay.initial-backoff:` `restart-strategy.exponential-delay.jitter-factor:` `restart-strategy.exponential-delay.reset-backoff-threshold:` ## 检查点和状态后端 `state.backend.type:` `state.backend.fs.memory-threshold:` `state.backend.incremental:` ## 检查点 `execution.checkpointing.unaligned:` `execution.checkpointing.interval-during-backlog:` ## RocksDB 原生指标 RocksDB 原生指标不提供给。 CloudWatch启用后，可以通过 Flink 控制面板或使用自定义工具通过 Flink REST API 访问这些指标。 适用于 Apache Flink 的托管服务允许客户使用 [API 以只读模式访问最新的 Flink REST API](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/ops/rest_api/)（或您正在使用的支持的版本）。[CreateApplicationPresignedUrl](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CreateApplicationPresignedUrl.html)此 API 由 Flink 自己的控制面板使用，但也可以由自定义监控工具使用。 `state.backend.rocksdb.metrics.actual-delayed-write-rate:` `state.backend.rocksdb.metrics.background-errors:` `state.backend.rocksdb.metrics.block-cache-capacity:` `state.backend.rocksdb.metrics.block-cache-pinned-usage:` `state.backend.rocksdb.metrics.block-cache-usage:` `state.backend.rocksdb.metrics.column-family-as-variable:` `state.backend.rocksdb.metrics.compaction-pending:` `state.backend.rocksdb.metrics.cur-size-active-mem-table:` `state.backend.rocksdb.metrics.cur-size-all-mem-tables:` `state.backend.rocksdb.metrics.estimate-live-data-size:` `state.backend.rocksdb.metrics.estimate-num-keys:` `state.backend.rocksdb.metrics.estimate-pending-compaction-bytes:` `state.backend.rocksdb.metrics.estimate-table-readers-mem:` `state.backend.rocksdb.metrics.is-write-stopped:` `state.backend.rocksdb.metrics.mem-table-flush-pending:` `state.backend.rocksdb.metrics.num-deletes-active-mem-table:` `state.backend.rocksdb.metrics.num-deletes-imm-mem-tables:` `state.backend.rocksdb.metrics.num-entries-active-mem-table:` `state.backend.rocksdb.metrics.num-entries-imm-mem-tables:` `state.backend.rocksdb.metrics.num-immutable-mem-table:` `state.backend.rocksdb.metrics.num-live-versions:` `state.backend.rocksdb.metrics.num-running-compactions:` `state.backend.rocksdb.metrics.num-running-flushes:` `state.backend.rocksdb.metrics.num-snapshots:` `state.backend.rocksdb.metrics.size-all-mem-tables:` ## RocksDB 选项 `state.backend.rocksdb.compaction.style:` `state.backend.rocksdb.memory.partitioned-index-filters:` `state.backend.rocksdb.thread.num:` ## 高级状态后端选项 `state.storage.fs.memory-threshold:` ## 完整 TaskManager 选项 `task.cancellation.timeout:` `taskmanager.jvm-exit-on-oom:` `taskmanager.numberOfTaskSlots:` `taskmanager.slot.timeout:` `taskmanager.memory.network.fraction:` `taskmanager.memory.network.max:` `taskmanager.network.request-backoff.initial:` `taskmanager.network.request-backoff.max:` `taskmanager.network.memory.buffer-debloat.enabled:` `taskmanager.network.memory.buffer-debloat.period:` `taskmanager.network.memory.buffer-debloat.samples:` `taskmanager.network.memory.buffer-debloat.threshold-percentages:` ## 内存配置 `taskmanager.memory.jvm-metaspace.size:` `taskmanager.memory.jvm-overhead.fraction:` `taskmanager.memory.jvm-overhead.max:` `taskmanager.memory.managed.consumer-weights:` `taskmanager.memory.managed.fraction:` `taskmanager.memory.network.fraction:` `taskmanager.memory.network.max:` `taskmanager.memory.segment-size:` `taskmanager.memory.task.off-heap.size:` ## RPC /Akka `akka.ask.timeout:` `akka.client.timeout:` `akka.framesize:` `akka.lookup.timeout:` `akka.tcp.timeout:` ## 客户端 `client.timeout:` ## 高级集群选项 `cluster.intercept-user-system-exit:` `cluster.processes.halt-on-fatal-error:` ## 文件系统配置 `fs.s3.connection.maximum:` `fs.s3a.connection.maximum:` `fs.s3a.threads.max:` `s3.upload.max.concurrent.uploads:` ## 高级容错选项 `heartbeat.timeout:` `jobmanager.execution.failover-strategy:` ## 内存配置 `jobmanager.memory.heap.size:` ## 指标 `metrics.latency.interval:` ## REST 端点和客户端的高级选项 `rest.flamegraph.enabled:` `rest.server.numThreads:` ## 高级SSL 安全选项 `security.ssl.internal.handshake-timeout:` ## 高级计划选项 `slot.request.timeout:` ## Flink Web 用户界面的高级选项 `web.timeout:` # 编程 Flink 配置属性 以下是您可以直接在应用程序代码中修改的 Flink 配置属性。从 MSF 2.2 开始，如果您尝试在应用程序代码中修改此页面上未列出的属性，则会引发异常。 **注意** 这些属性不同于通过 Su AWS pport case 修改的设置。有关基础架构级别的设置，例如 TaskManager 内存、状态后端、RocksDB 调整和重启策略，请参阅。[可修改的 Flink 配置属性](reference-modifiable-settings.md) ## 管道配置 `execution.runtime-mode` `pipeline.auto-generate-uids` `pipeline.auto-watermark-interval` `pipeline.cached-files` `pipeline.classpaths` `pipeline.closure-cleaner-level` `pipeline.force-avro` `pipeline.force-kryo` `pipeline.generic-types` `pipeline.global-job-parameters` `pipeline.jars` `pipeline.jobvertex-parallelism-overrides` `pipeline.max-parallelism` `pipeline.name` `pipeline.object-reuse` `pipeline.operator-chaining.chain-operators-with-different-max-parallelism` `pipeline.operator-chaining.enabled` `pipeline.serialization-config` `pipeline.vertex-description-mode` `pipeline.vertex-name-include-index-prefix` `pipeline.watermark-alignment.allow-unaligned-source-splits` ## Python API `python.execution-mode` `python.operator-chaining.enabled` `python.job-options` `python.internal.archives-key-map` `python.internal.files-key-map` `python.internal.requirements-file-key` ## 表 API/SQL `table.exec.async-lookup.buffer-capacity` `table.exec.async-lookup.key-ordered-enabled` `table.exec.async-lookup.output-mode` `table.exec.async-lookup.timeout` `table.exec.async-ml-predict.max-concurrent-operations` `table.exec.async-ml-predict.output-mode` `table.exec.async-ml-predict.timeout` `table.exec.async-scalar.max-concurrent-operations` `table.exec.async-scalar.max-attempts` `table.exec.async-scalar.retry-delay` `table.exec.async-scalar.retry-strategy` `table.exec.async-scalar.timeout` `table.exec.async-state.enabled` `table.exec.async-table.max-concurrent-operations` `table.exec.async-table.max-retries` `table.exec.async-table.retry-delay` `table.exec.async-table.retry-strategy` `table.exec.async-table.timeout` `table.exec.async-vector-search.max-concurrent-operations` `table.exec.async-vector-search.output-mode` `table.exec.async-vector-search.timeout` `table.exec.deduplicate.insert-update-after-sensitive-enabled` `table.exec.deduplicate.mini-batch.compact-changes-enabled` `table.exec.delta-join.cache-enabled` `table.exec.disabled-operators` `table.exec.iceberg.use-v2-sink` `table.exec.interval-join.min-cleanup-interval` `table.exec.legacy-cast-behaviour` `table.exec.local-hash-agg.adaptive.distinct-value-rate-threshold` `table.exec.local-hash-agg.adaptive.enabled` `table.exec.local-hash-agg.adaptive.sampling-threshold` `table.exec.mini-batch.allow-latency` `table.exec.mini-batch.enabled` `table.exec.mini-batch.size` `table.exec.operator-fusion-codegen.enabled` `table.exec.simplify-operator-name-enabled` `table.exec.sink.keyed-shuffle` `table.exec.sink.nested-constraint-enforcer` `table.exec.sink.not-null-enforcer` `table.exec.sink.rowtime-inserter` `table.exec.sink.type-length-enforcer` `table.exec.sink.upsert-materialize` `table.exec.sink.upsert-materialize-strategy.adaptive.threshold.high` `table.exec.sink.upsert-materialize-strategy.adaptive.threshold.low` `table.exec.sink.upsert-materialize-strategy.type` `table.exec.sort.async-merge-enabled` `table.exec.sort.default-limit` `table.exec.sort.max-num-file-handles` `table.exec.source.cdc-events-duplicate` `table.exec.source.idle-timeout` `table.exec.spill-compression.block-size` `table.exec.spill-compression.enabled` `table.exec.state.ttl` `table.exec.uid.format` `table.exec.uid.generation` `table.exec.unbounded-over.version` `table.exec.window-agg.buffer-size-limit` `table.optimizer.agg-phase-strategy` `table.optimizer.adaptive-broadcast-join.strategy` `table.optimizer.bushy-join-reorder-threshold` `table.optimizer.delta-join.strategy` `table.optimizer.distinct-agg.split.bucket-num` `table.optimizer.distinct-agg.split.enabled` `table.optimizer.dynamic-filtering.enabled` `table.optimizer.incremental-agg-enabled` `table.optimizer.join-reorder-enabled` `table.optimizer.multi-join.enabled` `table.optimizer.multiple-input-enabled` `table.optimizer.non-deterministic-update.strategy` `table.optimizer.reuse-optimize-block-with-digest-enabled` `table.optimizer.reuse-sink-enabled` `table.optimizer.reuse-source-enabled` `table.optimizer.reuse-sub-plan-enabled` `table.optimizer.runtime-filter.enabled` `table.optimizer.skewed-join-optimization.skewed-factor` `table.optimizer.skewed-join-optimization.skewed-threshold` `table.optimizer.skewed-join-optimization.strategy` `table.optimizer.source.report-statistics-enabled` `table.optimizer.union-all-as-breakpoint-enabled` `table.builtin-catalog-name` `table.builtin-database-name` `table.catalog-modification.listeners` `table.column-expansion-strategy` `table.display.max-column-width` `table.dml-sync` `table.dynamic-table-options.enabled` `table.generated-code.max-length` `table.legacy-nested-row-nullability` `table.local-time-zone` `table.plan.compile.catalog-objects` `table.plan.force-recompile` `table.plan.restore.catalog-objects` `table.rtas-ctas.atomicity-enabled` `table.sql-dialect` # 查看已配置的 Flink 属性 您可以通过 Apache Flink 控制面板查看自己配置或通过[支持案例](https://support.console.aws.amazon.com/support/home#/)请求修改的 Apache Flink 属性，并按照以下步骤操作： 1. 前往 Flink 控制面板 1. 在左侧导航窗格中选择 **Job Manager**。 1. 选择 “**配置**” 查看 Flink 属性列表。