答:如何加入 Apache Flink 和 Riak CS?
Riak CS 具有 S3(版本 2)兼容接口。因此,可以使用 Hadoop 的 S3 文件系统适配器与 Riak CS 配合使用。
我不知道为什么,但 Apache Flink 在 fat jar 中只有部分 Hadoop 文件系统适配器(lib/flink-dist_2.11-1.0.1.jar
)即它有 FTP 文件系统(org.apache.hadoop.fs.ftp.FTPFileSystem
)但没有 S3 文件系统(即org.apache.hadoop.fs.s3a.S3AFileSystem
)。所以,你有两种方法来解决这个问题:
- 使用 Hadoop 安装中的这些适配器。我没有尝试这个,但似乎你应该只配置 HADOOP_CLASSPATH 或 HADOOP_HOME evn 变量。
- Monky 修补 Apache Flink 并下载所需的 JAR
<flink home>/lib
目录
因此,我选择第二种方式,因为不想在我的环境中配置 Hadoop。您可以从 Hadoop dist 或互联网复制 JAR:
curl http://central.maven.org/maven2/org/apache/hadoop/hadoop-aws/2.7.2/hadoop-aws-2.7.2.jar -o /flink/lib/hadoop-aws-2.7.2.jar
curl http://central.maven.org/maven2/com/amazonaws/aws-java-sdk/1.7.4/aws-java-sdk-1.7.4.jar -o /flink/lib/aws-java-sdk-1.7.4.jar
curl http://central.maven.org/maven2/org/apache/httpcomponents/httpcore/4.2.5/httpcore-4.2.5.jar -o /flink/lib/httpcore-4.2.5.jar
curl http://central.maven.org/maven2/org/apache/httpcomponents/httpclient/4.2.5/httpclient-4.2.5.jar -o /flink/lib/httpclient-4.2.5.jar
正如你所看到的,我使用的是旧版本,因为这个版本在 Hadoop 2.7.2 中使用,并且我使用与这个版本的 Hadoop 兼容的 Flink。
仅供参考:如果您在自己的流程中使用这些 JAR 的最新版本,此类 hack 可能会导致问题。为了避免与不同版本相关的问题,您可以在使用流构建 fat jar 时重新定位包,使用类似的东西(我正在使用 Gradle):
// Relocate org.apache.http packages because Apache Flink include old version of this library (we place them for using S3 compatible FS)
shadowJar {
dependencies {
include(dependency('.*:.*:.*'))
}
relocate 'org.apache.http', 'relocated.org.apache.http'
relocate 'org.apache.commons', 'relocated.org.apache.commons'
}
然后你应该指定路径core-site.xml
in flink-conf.yaml
因为 Hadoop 兼容文件系统使用此配置来加载设置:
...
fs.hdfs.hadoopconf: /flink/conf
...
正如你所看到的,我只是把它放在<fink home>/conf
目录。它有以下设置:
<?xml version="1.0" encoding="UTF-8" ?>
<configuration>
<property>
<name>fs.s3a.impl</name>
<value>org.apache.hadoop.fs.s3a.S3AFileSystem</value> // because S3A better then other: https://wiki.apache.org/hadoop/AmazonS3
</property>
<property>
<name>fs.s3a.endpoint</name>
<value>my-riak-cs.stage.local</value> // this is my Riak CS host
</property>
<property>
<name>fs.s3a.connection.ssl.enabled</name> // my Riak CS in staging doesn't support SSL
<value>false</value>
</property>
<property>
<name>fs.s3a.access.key</name>
<value>????</value> // this is my access key for Riak CS
</property>
<property>
<name>fs.s3a.secret.key</name>
<value>????</value> // this is my secret key for Riak CS
</property>
</configuration>
然后你应该配置 Riak CS 存储桶flink-conf.yaml
作为推荐人here https://ci.apache.org/projects/flink/flink-docs-release-1.0/setup/jobmanager_high_availability.html#standalone-cluster-high-availability:
...
state.backend.fs.checkpointdir: s3a://example-staging-flink/checkpoints
...
recovery.zookeeper.storageDir: s3a://example-staging-flink/recovery
...
并在 Riak CS 中创建存储桶。我在用s3cmd
(安装在brew
在我的 OS X 开发环境中):
s3cmd mb s3://example-staging-flink
仅供参考:使用前s3cmd
你应该配置它使用s3cmd --configure
然后修复一些设置~/.s3cmd
file:
signature_v2 = True // because Riak CS using S3 V2 interface
use_https = False // if your don't use SSL
access_key = ???
secret_key = ???
host_base = my-riak-cs.stage.local // your Riak CS host
host_bucket = %(bucket).my-riak-cs.stage.local // format of bucket used by Riak CS
因此,这就是您在 Riak CS 中为独立 HA Apache Flink 集群的保存/恢复状态应该配置的所有内容。