亚马逊网络服务（AWS）

EMR：弹性MapReduce
- 标准EMR安装
- 自定义EMR安装
S3：简单存储服务
- Shaded Hadoop / Presto S3文件系统（推荐）
  - 配置访问凭据
    - 身份和访问管理（IAM）（推荐）
    - 访问Keys（气馁）
- Hadoop提供的S3文件系统 - 手动设置
常见问题

译者：flink.sojb.cn

Amazon Web Services提供可以运行Flink的云计算服务。

EMR：弹性MapReduce

Amazon Elastic MapReduce（Amazon EMR）是一种Web服务，可以轻松快速地设置Hadoop集群。这是在AWS上运行Flink 的推荐方法，因为它负责设置所有内容。

标准EMR安装

Flink是Amazon EMR上受支持的应用程序。亚马逊的文档描述了配置Flink，创建和监控集群以及处理作业。

自定义EMR安装

Amazon EMR服务会定期更新到新版本，但可以在库存EMR群集中手动安装不可用的Flink版本。

创建EMR集群

EMR文档包含显示如何启动EMR群集的示例。您可以按照该指南安装任何EMR版本。您不需要安装EMR版本的All Applications部分，但可以坚持使用Core Hadoop。

注意访问S3存储区需要在创建EMR集群时配置IAM角色。

在EMR群集上安装Flink

创建群集后，您可以连接到主节点并安装Flink：

转到下载页面并下载与您的EMR集群的Hadoop版本匹配的Flink二进制版本，例如Hadoop 2.7 for EMR版本4.3.0,4.4.0或4.5.0。
解压缩Flink发行版，您可以在设置Hadoop配置目录后通过YARN部署Flink作业：

HADOOP_CONF_DIR=/etc/hadoop/conf ./bin/flink run -m yarn-cluster -yn 1 examples/streaming/WordCount.jar

S3：简单存储服务

Amazon Simple Storage Service（Amazon S3）为各种用例提供云对象存储。您可以将S3与Flink一起用于读取和写入数据以及流状态后台甚至作为YARN对象存储。

您可以通过以下格式指定路径来使用常规文件等S3对象：

s3://<your-bucket>/<endpoint>

端点可以是单个文件或目录，例如：

// Read from S3 bucket
env.readTextFile("s3://<bucket>/<endpoint>");
// Write to S3 bucket
stream.writeAsText("s3://<bucket>/<endpoint>");
// Use S3 as FsStatebackend
env.setStateBackend(new FsStateBackend("s3://<your-bucket>/<endpoint>"));

请注意，这些示例并非详尽无遗，您也可以在其他地方使用S3，包括高可用性设置或RocksDBStateBackend ; Flink期望文件系统URI到处都是。

对于大多数用例，您可以使用我们的shaded flink-s3-fs-hadoop和flink-s3-fs-prestoS3文件系统打包器之一，这些打包器非常容易设置。但是，对于某些情况，例如，使用S3作为YARN的资源存储目录，可能需要设置特定的Hadoop S3 FileSystem实现。两种方式如下所述。

Shaded Hadoop / Presto S3文件系统（推荐）

注意：如果在EMR上运行Flink，则无需手动配置。

要使用flink-s3-fs-hadoop或flink-s3-fs-presto，请在启动Flink之前将相应的JAR文件从opt目录复制到libFlink分发的目录，例如

cp ./opt/flink-s3-fs-presto-1.7-SNAPSHOT.jar ./lib/

双方flink-s3-fs-hadoop并flink-s3-fs-presto注册与URI的默认文件系统的打包s3://方案，flink-s3-fs-hadoop也注册了s3a://。

配置访问凭据

设置S3 FileSystem打包器后，您需要确保允许Flink访问您的S3存储桶。

身份和访问管理（IAM）（推荐）

在AWS上设置凭据的推荐方法是通过身份和访问管理（IAM）。您可以使用IAM函数为Flink实例安全地提供他们访问S3存储桶所需的凭据。有关如何执行此算子操作的详细信息超出了本文档的范围。请参阅AWS用户指南。您正在寻找的是IAM角色。

如果您正确设置此选项，则可以在AWS中管理对S3的访问，并且不需要将任何访问Keys分发给Flink。

访问Keys（气馁）

可以通过您的访问和Keys对授予对S3的访问权限。请注意，自从引入IAM角色以来，不鼓励这样做。

您需要同时配置s3.access-key和s3.secret-key 在Flink的 flink-conf.yaml：

s3.access-key: your-access-key
s3.secret-key: your-secret-key

Hadoop提供的S3文件系统 - 手动设置

注意：如果在EMR上运行Flink，则无需手动配置。

这个设置有点复杂，我们建议使用我们的阴影Hadoop / Presto文件系统（见上文），除非另有要求，例如通过fs.defaultFSHadoop中的配置属性将S3用作YARN的资源存储目录core-site.xml。

设置S3 FileSystem

与S3的交互通过Hadoop的S3 FileSystem客户端之一进行：

S3AFileSystem（推荐用于Hadoop 2.7及更高版本）：用于在内部使用Amazon SDK读取和写入常规文件的文件系统。没有最大文件大小并且与IAM角色一起使用。
NativeS3FileSystem（对于Hadoop 2.6及更早版本）：用于读写常规文件的文件系统。最大对象大小为5GB，不适用于IAM角色。

`S3AFileSystem` （推荐的）

这是推荐使用的S3 FileSystem实现。它在内部使用Amazon的SDK并与IAM角色一起使用（请参阅配置访问凭据）。

您需要将Flink指向有效的Hadoop配置，该配置包含以下属性core-site.xml：

<configuration>
<property>
  <name>fs.s3.impl</name>
  <value>org.apache.hadoop.fs.s3a.S3AFileSystem</value>
</property>
<!-- Comma separated list of local directories used to buffer
     large results prior to transmitting them to S3\. -->
<property>
  <name>fs.s3a.buffer.dir</name>
  <value>/tmp</value>
</property>
</configuration>

这将注册S3AFileSystem为具有该s3a://方案的URI的默认文件系统。

`NativeS3FileSystem`

此文件系统仅限于最大5GB的文件，并且不适用于IAM角色（请参阅配置访问凭据），这意味着您必须在Hadoop配置文件中手动配置AWS凭据。

您需要将Flink指向有效的Hadoop配置，该配置包含以下属性core-site.xml：

<property>
  <name>fs.s3.impl</name>
  <value>org.apache.hadoop.fs.s3native.NativeS3FileSystem</value>
</property>

这将注册NativeS3FileSystem为具有该s3://方案的URI的默认文件系统。

Hadoop配置

例如，您可以通过各种方式指定Hadoop配置，将Flink指向Hadoop配置目录的路径

通过设置环境变量HADOOP_CONF_DIR，或
通过fs.hdfs.hadoopconf在flink-conf.yaml以下位置设置配置选项：

fs.hdfs.hadoopconf: /path/to/etc/hadoop

这将/path/to/etc/hadoop在Flink中注册为Hadoop的配置目录。Flink将查找指定目录中的core-site.xml和hdfs-site.xml文件。

配置访问凭据

注意：如果在EMR上运行Flink，则无需手动配置。

设置S3 FileSystem后，您需要确保允许Flink访问您的S3存储桶。

身份和访问管理（IAM）（推荐）

使用时S3AFileSystem，在AWS上设置凭据的推荐方法是通过身份和访问管理（IAM）。您可以使用IAM函数为Flink实例安全地提供他们访问S3存储桶所需的凭据。有关如何执行此算子操作的详细信息超出了本文档的范围。请参阅AWS用户指南。您正在寻找的是IAM角色。

如果您正确设置此选项，则可以在AWS中管理对S3的访问，并且不需要将任何访问Keys分发给Flink。

请注意，这只适用于S3AFileSystem而不是NativeS3FileSystem。

访问Keys`S3AFileSystem`（不鼓励）

可以通过您的访问和Keys对授予对S3的访问权限。请注意，自从引入IAM角色以来，不鼓励这样做。

对于S3AFileSystem您需要配置fs.s3a.access.key并fs.s3a.secret.key 在Hadoop的 core-site.xml：

<property>
  <name>fs.s3a.access.key</name>
  <value></value>
</property>
<property>
  <name>fs.s3a.secret.key</name>
  <value></value>
</property>

访问Keys`NativeS3FileSystem`（不鼓励）

可以通过您的访问和Keys对授予对S3的访问权限。但是这是不鼓励的，你应该使用S3AFileSystem 所需的IAM角色。

对于NativeS3FileSystem您需要配置fs.s3.awsAccessKeyId并fs.s3.awsSecretAccessKey 在Hadoop的 core-site.xml：

<property>
  <name>fs.s3.awsAccessKeyId</name>
  <value></value>
</property>
<property>
  <name>fs.s3.awsSecretAccessKey</name>
  <value></value>
</property>

提供S3 FileSystem依赖关系

注意：如果在EMR上运行Flink，则无需手动配置。

Hadoop的S3 FileSystem客户端打包在hadoop-aws工件中（Hadoop 2.6及更高版本）。需要将此JAR及其所有依赖项添加到Flink的类路径中，即Job和TaskManagers的类路径。根据您使用的FileSystem实现以及您使用的Flink和Hadoop版本，您需要提供不同的依赖项（请参阅下文）。

有多种方法可以将JAR添加到Flink的类路径中，最简单的方法就是将JAR放在Flink的lib文件夹中。您需要复制hadoop-aws具有所有依赖项的JAR。您还可以HADOOP_CLASSPATH在所有计算机上将包含这些JAR的目录导出为环境变量的一部分。

Flink for Hadoop 2.7

根据您使用的文件系统，请添加以下依赖项。您可以在以下位置找到这些作为Hadoop二进制文件的一部分hadoop-2.7/share/hadoop/tools/lib：

S3AFileSystem：
- hadoop-aws-2.7.3.jar
- aws-java-sdk-s3-1.11.183.jar 及其依赖：
  - aws-java-sdk-core-1.11.183.jar
  - aws-java-sdk-kms-1.11.183.jar
  - jackson-annotations-2.6.7.jar
  - jackson-core-2.6.7.jar
  - jackson-databind-2.6.7.jar
  - joda-time-2.8.1.jar
  - httpcore-4.4.4.jar
  - httpclient-4.5.3.jar
NativeS3FileSystem：
- hadoop-aws-2.7.3.jar
- guava-11.0.2.jar

请注意，hadoop-common它可作为Flink的一部分提供，但番石榴被Flink遮蔽。

Flink for Hadoop 2.6

根据您使用的文件系统，请添加以下依赖项。您可以在以下位置找到这些作为Hadoop二进制文件的一部分hadoop-2.6/share/hadoop/tools/lib：

S3AFileSystem：
- hadoop-aws-2.6.4.jar
- aws-java-sdk-1.7.4.jar 及其依赖：
  - jackson-annotations-2.1.1.jar
  - jackson-core-2.1.1.jar
  - jackson-databind-2.1.1.jar
  - joda-time-2.2.jar
  - httpcore-4.2.5.jar
  - httpclient-4.2.5.jar
NativeS3FileSystem：
- hadoop-aws-2.6.4.jar
- guava-11.0.2.jar

请注意，hadoop-common它可作为Flink的一部分提供，但番石榴被Flink遮蔽。

Flink for Hadoop 2.4及更早版本

这些Hadoop版本只支持NativeS3FileSystem。这是预装了Flink for Hadoop 2的一部分hadoop-common。您不需要向类路径添加任何内容。

常见问题

以下部分列出了在AWS上使用Flink时的常见问题。

缺少S3文件系统配置

如果您的作业提交失败，并显示异常消息，指出No file system found with scheme s3这意味着没有为S3配置FileSystem。有关如何正确配置的详细信息，请查看我们的着色Hadoop / Presto或通用Hadoop文件系统的配置部分。

org.apache.flink.client.program.ProgramInvocationException: The program execution failed:
  Failed to submit job cd927567a81b62d7da4c18eaa91c3c39 (WordCount Example) [...]
Caused by: org.apache.flink.runtime.JobException: Creating the input splits caused an error:
  No file system found with scheme s3, referenced in file URI 's3://<bucket>/<endpoint>'. [...]
Caused by: java.io.IOException: No file system found with scheme s3,
  referenced in file URI 's3://<bucket>/<endpoint>'.
    at o.a.f.core.fs.FileSystem.get(FileSystem.java:296)
    at o.a.f.core.fs.Path.getFileSystem(Path.java:311)
    at o.a.f.api.common.io.FileInputFormat.createInputSplits(FileInputFormat.java:450)
    at o.a.f.api.common.io.FileInputFormat.createInputSplits(FileInputFormat.java:57)
    at o.a.f.runtime.executiongraph.ExecutionJobVertex.<init>(ExecutionJobVertex.java:156)

未指定AWS访问KeysID和密钥访问Keys

如果您发现您的作业失败并显示异常AWS Access Key ID and Secret Access Key must be specified as the username or password，并且未正确设置您的访问凭据。有关如何配置此信息的详细信息，请参阅我们的着色Hadoop / Presto或通用Hadoop文件系统的访问凭据部分。

org.apache.flink.client.program.ProgramInvocationException: The program execution failed:
  Failed to submit job cd927567a81b62d7da4c18eaa91c3c39 (WordCount Example) [...]
Caused by: java.io.IOException: The given file URI (s3://<bucket>/<endpoint>) points to the
  HDFS NameNode at <bucket>, but the File System could not be initialized with that address:
  AWS Access Key ID and Secret Access Key must be specified as the username or password
  (respectively) of a s3n URL, or by setting the fs.s3n.awsAccessKeyId
  or fs.s3n.awsSecretAccessKey properties (respectively) [...]
Caused by: java.lang.IllegalArgumentException: AWS Access Key ID and Secret Access Key must
  be specified as the username or password (respectively) of a s3 URL, or by setting
  the fs.s3n.awsAccessKeyId or fs.s3n.awsSecretAccessKey properties (respectively) [...]
    at o.a.h.fs.s3.S3Credentials.initialize(S3Credentials.java:70)
    at o.a.h.fs.s3native.Jets3tNativeFileSystemStore.initialize(Jets3tNativeFileSystemStore.java:80)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at o.a.h.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:187)
    at o.a.h.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:102)
    at o.a.h.fs.s3native.$Proxy6.initialize(Unknown Source)
    at o.a.h.fs.s3native.NativeS3FileSystem.initialize(NativeS3FileSystem.java:330)
    at o.a.f.runtime.fs.hdfs.HadoopFileSystem.initialize(HadoopFileSystem.java:321)

ClassNotFoundException：找不到NativeS3FileSystem / S3AFileSystem

如果您看到此异常，则S3 FileSystem不是Flink的类路径的一部分。有关如何正确配置的详细信息，请参阅S3 FileSystem相关性部分。

Caused by: java.lang.RuntimeException: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3native.NativeS3FileSystem not found
  at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2186)
  at org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.getHadoopWrapperClassNameForFileSystem(HadoopFileSystem.java:460)
  at org.apache.flink.core.fs.FileSystem.getHadoopWrapperClassNameForFileSystem(FileSystem.java:352)
  at org.apache.flink.core.fs.FileSystem.get(FileSystem.java:280)
  at org.apache.flink.core.fs.Path.getFileSystem(Path.java:311)
  at org.apache.flink.api.common.io.FileInputFormat.createInputSplits(FileInputFormat.java:450)
  at org.apache.flink.api.common.io.FileInputFormat.createInputSplits(FileInputFormat.java:57)
  at org.apache.flink.runtime.executiongraph.ExecutionJobVertex.<init>(ExecutionJobVertex.java:156)
  ... 25 more
Caused by: java.lang.RuntimeException: java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3native.NativeS3FileSystem not found
  at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2154)
  at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2178)
  ... 32 more
Caused by: java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3native.NativeS3FileSystem not found
  at org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:2060)
  at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2152)
  ... 33 more

IOException异常： `400: Bad Request`

如果您已正确配置所有内容但获得Bad Request异常且 S3存储桶位于区域中eu-central-1，则可能正在运行S3客户端，该客户端不支持Amazon的签名版本4。

[...]
Caused by: java.io.IOException: s3://<bucket-in-eu-central-1>/<endpoint> : 400 : Bad Request [...]
Caused by: org.jets3t.service.impl.rest.HttpException [...]

要么

com.amazonaws.services.s3.model.AmazonS3Exception: Status Code: 400, AWS Service: Amazon S3, AWS Request ID: [...], AWS Error Code: null, AWS Error Message: Bad Request, S3 Extended Request ID: [...]

这不应该适用于我们的阴影Hadoop / Presto S3文件系统，但可以适用于Hadoop提供的S3文件系统。特别是，所有运行高达2.7.2的Hadoop版本NativeS3FileSystem（取决于JetS3t 0.9.0版本> = 0.9.4）都会受到影响，但用户也会报告这种情况S3AFileSystem。

除了更改存储区域之外，您还可以通过请求用于请求身份验证的签名版本4来解决此问题，例如将其添加到Flink的JVM选项中flink-conf.yaml（请参阅配置）：

env.java.opts: -Dcom.amazonaws.services.s3.enableV4

org.apache.hadoop.fs.LocalDirAllocator中的NullPointerException

此异常通常是由跳过本地缓冲区目录配置引起fs.s3a.buffer.dir的S3AFileSystem。请参阅S3AFileSystem配置部分以了解如何S3AFileSystem正确配置。

[...]
Caused by: java.lang.NullPointerException at
o.a.h.fs.LocalDirAllocator$AllocatorPerContext.confChanged(LocalDirAllocator.java:268) at
o.a.h.fs.LocalDirAllocator$AllocatorPerContext.getLocalPathForWrite(LocalDirAllocator.java:344) at
o.a.h.fs.LocalDirAllocator$AllocatorPerContext.createTmpFileForWrite(LocalDirAllocator.java:416) at
o.a.h.fs.LocalDirAllocator.createTmpFileForWrite(LocalDirAllocator.java:198) at
o.a.h.fs.s3a.S3AOutputStream.<init>(S3AOutputStream.java:87) at
o.a.h.fs.s3a.S3AFileSystem.create(S3AFileSystem.java:410) at
o.a.h.fs.FileSystem.create(FileSystem.java:907) at
o.a.h.fs.FileSystem.create(FileSystem.java:888) at
o.a.h.fs.FileSystem.create(FileSystem.java:785) at
o.a.f.runtime.fs.hdfs.HadoopFileSystem.create(HadoopFileSystem.java:404) at
o.a.f.runtime.fs.hdfs.HadoopFileSystem.create(HadoopFileSystem.java:48) at
... 25 more

EMR：弹性MapReduce

标准EMR安装

自定义EMR安装

S3：简单存储服务

Shaded Hadoop / Presto S3文件系统（推荐）

配置访问凭据

身份和访问管理（IAM）（推荐）

访问Keys（气馁）

Hadoop提供的S3文件系统 - 手动设置

设置S3 FileSystem

S3AFileSystem （推荐的）

NativeS3FileSystem

Hadoop配置

配置访问凭据

身份和访问管理（IAM）（推荐）

访问KeysS3AFileSystem（不鼓励）

访问KeysNativeS3FileSystem（不鼓励）

提供S3 FileSystem依赖关系

Flink for Hadoop 2.7

Flink for Hadoop 2.6

Flink for Hadoop 2.4及更早版本

常见问题

缺少S3文件系统配置

未指定AWS访问KeysID和密钥访问Keys

ClassNotFoundException：找不到NativeS3FileSystem / S3AFileSystem

IOException异常： 400: Bad Request

org.apache.hadoop.fs.LocalDirAllocator中的NullPointerException

`S3AFileSystem` （推荐的）

`NativeS3FileSystem`

访问Keys`S3AFileSystem`（不鼓励）

访问Keys`NativeS3FileSystem`（不鼓励）

IOException异常： `400: Bad Request`