java - 将外部类与数据流结合使用

Question

我正在尝试创建一个接受侧面输入的自定义转换类，并且我正在使用 PipelineTester 和 PAssert 来尝试测试它，但我一直在尝试将其引入转换的方法中遇到 no such method 异常来自其他类(class)。

Caused by: java.lang.NoSuchMethodError: com.org.utils.MyUtils.createMap(Ljava/lang/Iterable;)Ljava/util/Map; at com.org.beam.MyTransform.ProcessElement(MyTransform.java:51)

我尝试使用@Autowired注释来引入类

@Autowired
private MyUtils myutils;

以及在代码中创建一个静态实例，例如

private static MyUtils myUtils = new MyUtils();

然后调用

this.myUtils.createMap(mapThisToThat(inputCollection, this.myMap));

我还尝试将方法设为静态并像这样调用它们

MyUtils.createMap(mapThisToThat(inputCollection, this.myMap));

mapThisToThat 的签名是

private Iterable<MyObject> mapThisToThat(Iterable<MyObject> objectIterator, Map<String, Integer> myMap) {

它被传递到具有此签名的 createMap 方法 -

public Map<String, MyObject> createMap(Iterable<MyObject> inputCollection){

因此它正确地传入了 MyObjects 的 Iterable，但它表示该方法由于某种原因不存在。这是否意味着光束变换不能有外部方法或者我做错了什么？

Answer 1

对于我来说，Python 需要做很多事情才能使其正常工作:

https://cloud.google.com/dataflow/faq#how-do-i-handle-nameerrors https://beam.apache.org/documentation/sdks/python-pipeline-dependencies/

对于java中的你来说，由于某种原因他们没有相互的文档，但是在这里https://beam.apache.org/documentation/runners/dataflow/他们这样说:

In some cases, such as starting a pipeline using a scheduler such as Apache AirFlow, you must have a self-contained application. You can pack a self-executing JAR by explicitly adding the following dependency on the Project section of your pom.xml, in addition to the adding existing dependency shown in the previous section.

在他们的示例中，自述文件 https://github.com/mbrukman/apache-beam/tree/master/examples/java他们这样说

Alternatively, you may choose to bundle all dependencies into a single JAR and execute it outside of the Maven environment. For example, you can execute the following commands to create the bundled JAR of the examples and execute it both locally and in Cloud Platform

如果您继续浏览该示例存储库，则会出现一个包含 utils 的 common 文件夹。希望您可以复制他们是如何做到的。