在 EMR 上运行带有 flink 纱线会话的光束管道答案

【问题标题】：Run beam pipeline with flink yarn session on EMR在 EMR 上运行带有 flink 纱线会话的光束管道
【发布时间】：2020-10-07 19:49:16
【问题描述】：

我正在尝试在 AWS EMR 上使用 flink yarn 会话从 python SDK 运行基本的 wordcount 光束管道。我同时使用了 flink 运行器和便携式运行器，并得到下面列出的两个不同的错误。两种类型的运行器的作业都显示在 flink UI 中，并在我的笔记本电脑上通过本地 flink 会话成功运行。

使用 FlinkRunner，作业作为 BeamApp-hadoop-0617202523-14894e58 运行并给出错误：

ERROR:root:java.lang.NoClassDefFoundError: Could not initialize class org.apache.beam.runners.core.construction.SerializablePipelineOptions

使用 PortableRunner，作业以 BeamApp-root-0617202248-36b0d306 运行（我相信这意味着它已成功从 Beam 便携式运行器 docker 映像提交作业）并给出错误：

ERROR:root:java.util.ServiceConfigurationError: com.fasterxml.jackson.databind.Module: Provider com.fasterxml.jackson.module.jaxb.JaxbAnnotationModule not a subtype

我认为这些是依赖错误，并尝试在 /usr/lib/flink/lib 目录中获取提到的 jar。在应用程序启动时记录类路径时，纱线容器日志列出了正确的 jar，但错误仍然存在。

Apache Beam 版本 2.22.0，flink 版本 1.10.0，emr 版本 5.30.0。

【问题讨论】：

标签： apache-flink hadoop-yarn apache-beam amazon-emr

【解决方案1】：

我在使用 Apache beam + AWS EMR + Flink 时遇到了类似的问题，我通过从 FlinkPipelineOptions.filesToStage 中排除 jackson-core、jackson-annotation 和 jackson-databind 依赖项解决了这个问题。

options.setFilesToStage(Arrays.stream(System.getProperty("java.class.path").split(":"))
   .filter(d -> !d.contains("com.fasterxml.jackson.core"))
   .filter(d -> Files.exists(Paths.get(d)))
   .collect(Collectors.toList()));

【讨论】：