配置 Flink - Amazon EMR
并行选项可配置的文件在包括多个主节点的 EMR 群集中配置 Flink。配置内存进程大小配置日志输出文件大小
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅中国的 Amazon Web Services 服务入门。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

配置 Flink

建议您使用配置文件配置 Flink。例如，Flink 的主配置文件的名称为 flink-conf.yaml。可以使用 Amazon EMR 配置 API 对其进行配置。

使用 Amazon CLI 配置用于 Flink 的任务槽的数目

  1. 创建文件 configurations.json 并输入以下内容： 

    [
    {
      "Classification": "flink-conf",
      "Properties": {
        "taskmanager.numberOfTaskSlots":"2"
      }
    }
]

  2. 接下来，使用以下配置创建集群：

    aws emr create-cluster --release-label emr-5.33.0 \
--applications Name=Flink \
--configurations file://./configurations.json \
--region us-east-1 \
--log-uri s3://myLogUri \
--instance-type m5.xlarge \
--instance-count 2 \
--service-role EMR_DefaultRole \ 
--ec2-attributes KeyName=YourKeyName,InstanceProfile=EMR_EC2_DefaultRole
注意

也可以使用 Flink API 更改某些配置。有关更多信息，请参阅 。概念在 Fink 文档中。

使用 Amazon EMR 版本 5.21.0 及更高版本，您可以覆盖集群配置，并为运行的集群中的每个实例组指定额外的配置分类。您可以使用 Amazon EMR 控制台、Amazon Command Line Interface(Amazon CLI），或Amazon开发工具包。有关更多信息，请参阅 。为运行的集群中的实例组提供配置。

作为应用程序所有者，您最了解应将哪些资源分配给 Flink 中的任务。在本文档的示例中，使用与用于应用程序的从属实例的数目相同的任务。通常，我们建议对初始并行级别执行此操作，但您也可以使用任务槽来增加并行粒度，它一般不应超过每实例虚拟核心数。有关 Flink 的体系结构的更多信息，请参阅概念在 Fink 文档中。

目前，可在 Amazon EMR 配置 API 中配置的文件包括：

  • flink-conf.yaml

  • log4j.properties

  • flink-log4j-session

  • log4j-cli.properties

在包括多个主节点的 EMR 群集中进行主节点故障转移的过程中，Flink 的 JobManager 仍然可用。从 Amazon EMR 版本 5.28.0 开始，JobManager 的高可用性也会自动启用。无需手动配置。

对于 Amazon EMR 版本 5.27.0 或更早版本，JobManager 是单点故障。当 JobManager 失败时，它会失去所有作业状态，并且不会恢复正在运行的作业。通过配置应用程序尝试计数、开展检查点检验并启用 ZooKeeper 作为 Flink 的状态存储，您可以启用 JobManager 高可用性，如以下示例所示： 

[
  {
    "Classification": "yarn-site",
    "Properties": {
      "yarn.resourcemanager.am.max-attempts": "10"
    }
  },
  {
    "Classification": "flink-conf",
    "Properties": {
        "yarn.application-attempts": "10",
        "high-availability": "zookeeper",
        "high-availability.zookeeper.quorum": "%{hiera('hadoop::zk')}",
        "high-availability.storageDir": "hdfs:///user/flink/recovery",
        "high-availability.zookeeper.path.root": "/flink"
    }
  }
]

您必须同时为 YARN 和 Flink 配置最大的应用程序主尝试次数。有关更多信息，请参阅 。YARN 集群的高可用性配置。您可能还需要配置 Flink 检查点，以使重新启动的 JobManager 从先前完成的检查点恢复正在运行的作业。有关更多信息，请参阅 。Flink 检查点。

对于使用 Flink 1.11.x 的亚马逊 EMR 版本（5.31、5.32、6.1 和 6.2），您需要为两个 JobManager（jobmanager.memory.process.size) 和任务管理器 (taskmanager.memory.process.size)flink-conf.yaml。您可以通过使用配置 API 配置集群或通过 SSH 手动取消注释这些字段来设置这些值。开源代码给出的默认值是：

  • jobmanager.memory.process.size: 1600 米

  • taskmanager.memory.process.size: 1728 米

如果要排除 JVM 元空间和开销，请使用 Flink 总内存大小（taskmanager.memory.flink.size）而不是taskmanager.memory.process.size。的默认值为taskmanager.memory.process.size是一亿八千米。不建议同时设置taskmanager.memory.process.size和taskmanager.memory.process.size。

所有使用 Fink 1.12.0 及更高版本的亚马逊 EMR 版本都将 Fink 的开源设置中列出的默认值作为亚马逊 EMR 上的默认值，因此您无需自行配置。

Fink 应用程序容器创建并写入三种类型的日志文件：.out文件，.log文件和.err文件。仅限.err文件被压缩并从文件系统中删除，而.log和.out日志文件仍保留在文件系统中。为了确保这些输出文件保持可管理且集群保持稳定，您可以在log4j.properties设置文件的最大数量并限制其大小： 

[
  {
    "Classification": "flink-log4j",
    "Properties": {
      "log4j.appender.file": "org.apache.log4j.RollingFileAppender",
      "log4j.appender.file.append":"true",
      # keep up to 4 files and each file size is limited to 100MB
      "log4j.appender.file.MaxFileSize":"100MB",
      "log4j.appender.file.MaxBackupIndex":4,
      "log4j.appender.file.layout":"org.apache.log4j.PatternLayout",
      "log4j.appender.file.layout.ConversionPattern":"%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n"
    },
  }
]