如何从 SBT（本地）在 Dataflow 上运行 Scio 管道

Question

我正在尝试在 Dataflow 上运行我的第一个 Scio 管道。

可以在here 找到相关代码。但是我认为这不是太重要。
我的第一个实验是使用 DirecRunner 读取一些本地 CSV 文件并编写另一个本地 CSV 文件。这按预期工作。

现在，我正在尝试从 GCS 读取文件，将输出写入 BigQuery 并使用 DataflowRunner 运行管道。我已经进行了所有必要的更改（或者这就是我所相信的）。但我无法让它运行。

我已经gcloud auth application-default login 并且当我这样做时

sbt run --runner=DataflowRunner --project=project-id --input-path=gs://path/to/data --output-table=dataset.table

我可以看到 Jb 在 Dataflow 中提交。但是，一小时后作业失败并显示以下错误消息。

工作流程失败。原因：Dataflow 作业似乎卡住了，因为在过去 1 小时内没有看到任何工作人员活动。

（请注意，这项工作在那段时间里什么也没做，因为这是一个实验，所以数据太简单了，需要几分钟时间）。

检查StackDriver可以发现如下错误：

java.lang.ClassNotFoundException: scala.collection.Seq

与一些 jackson 的事情有关：

java.util.ServiceConfigurationError: com.fasterxml.jackson.databind.Module: 提供者 com.fasterxml.jackson.module.scala.DefaultScalaModule 无法实例化

这就是一开始就杀死每个执行者的原因。我真的不明白为什么我找不到 Scala 标准库。

我也尝试先创建一个模板，然后用：

sbt run --runner=DataflowRunner --project=project-id --input-path=gs://path/to/data --output-table=dataset.table --stagingLocation=gs://path/to/staging --templateLocation=gs://path/to/templates/template-1

但是，运行模板后，我得到了同样的错误。
另外，我注意到 staging 文件夹中有很多 jar，但 scala-library.jar 不在其中。

我遗漏了一些明显的东西？

Answer 1

这是 sbt 1.3.0 的一个已知问题，它引入了一些重大更改 w.r.t。类加载器。试试 1.2.8？

Jackson 问题也可能与 Java 11 或更高版本有关。现在继续使用 Java 8。

Answer 2

通过设置 sbt classLoaderLayeringStrategy 来修复：

run / classLoaderLayeringStrategy := ClassLoaderLayeringStrategy.Flat

sbt 为使用run 运行的应用程序使用新的类加载器。这会导致 JVM 已加载的其他类（例如 Predef）被重用，从而减少启动时间。详情请见in-process classloaders。

这不适用于 Beam DataflowRunner，因为它明确不从父类加载器暂存类，请参阅 PipelineResources.java#L51：

尝试检测类加载器可以访问的所有资源。这不会递归到类加载器父级阻止它从系统类加载器中提取资源。

因此，解决方法是强制将应用程序使用的所有类加载到同一个类加载器中，以便 DataflowRunner 暂存所有内容。

希望有帮助