如何在 Flink 中为 Google Cloud Storage 创建 RecoverableWriter答案

【问题标题】：How to create a RecoverableWriter in Flink for Google Cloud Storage如何在 Flink 中为 Google Cloud Storage 创建 RecoverableWriter
【发布时间】：2018-09-16 13:52:39
【问题描述】：

我想使用Google Cloud Storage 从我使用StreamingFileSink 的流式作业中写入（sink）DataStream 的元素。

为此，我将 Google Cloud Storage connector 用于 Hadoop，作为 org.apache.hadoop.fs.FileSystem 的实现，并使用 HadoopFileSystem as an implementation of org.apache.flink.core.fs.FileSystem 包装了 Flink 的 hadoop FileSystem 类。

我在我的 gradle 文件中包含了以下依赖项：

compile("com.google.cloud.bigdataoss:gcs-connector:1.9.4-hadoop2")
compile("org.apache.flink:flink-connector-filesystem_2.11:1.6.0")
provided("org.apache.flink:flink-shaded-hadoop2:1.6.0")

现在，据我了解，查看源代码 ^[1]^[2]^[3]，Flink 会动态加载FileSystemFactory 在运行时（通过java.util.ServiceLoader）并在运行时加载HadoopFsFactory（通过reflection，如果它在类路径中找到Hadoop），然后它使用它来创建FileSystem 的实例。

我遇到的问题是 Hadoop 兼容包的默认 RecoverableWriter 仅支持 hdfs 文件方案（我使用 gs），因此支持 throws an error at runtime。

所以，我extendedHadoopFileSystem（我叫GCSFileSystem）和@overridedFileSystem#createRecoverableWriter()返回一个RecoverableWriter的自定义实现，然后处理恢复等细节并创建一个对应的FileSystemFactory 类（该类用@AutoService 装饰，因此应该可以被ServiceLoader 发现）。

该设置在本地和本地 docker 集群上运行良好（实际上 GCS 连接器由于缺乏授权而引发错误，但这很好，因为这意味着 FileSystem 已加载并正在运行）但在我部署时失败它到在 Google Compute Engine 上运行的 docker 集群。

在 GCE 上，默认的 HadoopFileSystem 被加载并抛出异常，因为方案是 gs 而不是 hdfs，但我的假设是它应该已经加载了我的工厂实现，因此这个错误应该t已经出现了。

我在 Flink v1.6.0 上运行 long running session cluster on Docker using docker-flink

【问题讨论】：

标签： java hadoop google-cloud-storage google-compute-engine apache-flink

【解决方案1】：

答案在 OP 的最后一行！！

我在一个长寿的会话集群上运行，当我的job.jar 被执行时，FileSystem 初始化已经完成并且工厂已经加载！因此，当我添加作业时，没有进行 初始化 调用。

解决方案？根据您部署工作的方式，有几种方法：

独立：将包含FileSystem 实现的jar 添加到lib/ 目录
集群 (manual)： 将包含 FileSystem 实现的 jar 添加到您的 lib/ 目录中的 lib/ 或图像或其他任何东西。
李>
Cluster (docker)(long-living)：创建自定义容器镜像并将 jar 添加到该镜像的lib/ 目录。
李>
Cluster (docker)(per-job-session)： 创建自定义容器镜像并将所有 jar（包含 FileSystem 和您的工作等）添加到 @ 987654336@目录，read more about per-job session here.

【讨论】：