java - 如何在 Flink 中为 Google Cloud Storage 创建 RecoverableWriter

Question

我想使用 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 文件中包含了以下依赖项:

• 编译("com.google.cloud.bigdataoss:gcs-connector:1.9.4-hadoop2")
• 编译("org.apache.flink:flink-connector-filesystem_2.11:1.6.0")
• 提供("org.apache.flink:flink-shaded-hadoop2:1.6.0")

现在，根据我对资源的理解 ^{[1] [2] [3]} ，Flink动态加载FileSystemFactory的实现在运行时(通过 java.util.ServiceLoader)并加载 HadoopFsFactory在运行时(通过反射，如果它在类路径中找到 Hadoop)然后使用它来创建 FileSystem 的实例。

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

因此，我扩展 HadoopFileSystem(我调用GCSFileSystem)和@overrided FileSystem#createRecoverableWriter()返回 RecoverableWriter 的自定义实现然后处理恢复等细节，并创建相应的 FileSystemFactory 类(该类用 @AutoService 修饰，因此应该可以被 ServiceLoader< 发现)。

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

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

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

Answer 1

答案在 OP 的最后一行!!

我在一个长期存在的 session 集群上运行，当我的 job.jar 被执行时，FileSystem 初始化已经完成完成了，工厂已经装满了!因此，当我添加作业时，没有进行任何初始化调用。

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

• 独立将包含FileSystem实现的jar添加到lib/目录

• Cluster (manual):将包含FileSystem实现的jar添加到lib/目录您的 zip 或图像或其他内容。

• Cluster (docker)(long-living):创建一个自定义容器镜像并将jar添加到lib/ 该图像的目录。

• 集群 (docker)(per-job-session): 创建自定义容器镜像并添加所有 jar (包含 FileSystem 和您的工作等)到 lib/ 目录，read more about per-job session here.