亚马逊网络服务(AWS)

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:

  1. 转到下载页面下载与您的EMR集群的Hadoop版本匹配的Flink二进制版本,例如Hadoop 2.7 for EMR版本4.3.0,4.4.0或4.5.0。
  2. 解压缩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-hadoopflink-s3-fs-prestoS3文件系统打包器之一,这些打包器非常容易设置。但是,对于某些情况,例如,使用S3作为YARN的资源存储目录,可能需要设置特定的Hadoop S3 FileSystem实现。两种方式如下所述。

注意:如果在EMR上运行Flink,则无需手动配置

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

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

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

配置访问凭据

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

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

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

访问Keys(气馁)

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

您需要同时配置s3.access-keys3.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客户端之一进行

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

这是推荐使用的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.hadoopconfflink-conf.yaml以下位置设置配置选项
fs.hdfs.hadoopconf: /path/to/etc/hadoop

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

回到顶部

配置访问凭据

注意:如果在EMR上运行Flink,则无需手动配置

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

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

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

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

回到顶部

访问KeysS3AFileSystem(不鼓励)

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

对于S3AFileSystem您需要配置fs.s3a.access.keyfs.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>

回到顶部

访问KeysNativeS3FileSystem(不鼓励)

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

对于NativeS3FileSystem您需要配置fs.s3.awsAccessKeyIdfs.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的目录导出为环境变量的一部分

根据您使用的文件系统,请添加以下依赖项。您可以在以下位置找到这些作为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遮蔽。

根据您使用的文件系统,请添加以下依赖项。您可以在以下位置找到这些作为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遮蔽。

这些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.dirS3AFileSystem请参阅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

回到顶部