apache-flink - 在 EMR 上使用 flink yarn session 运行 beam pipeline

Question

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

使用 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 portable runner 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 目录中获取提到的 jars。在应用程序启动时记录类路径时，yarn 容器日志列出了正确的 jar，但错误仍然存​​在。

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

Answer 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()));