java - 在 Beam 中初始化外部服务连接

Question

我正在编写 Dataflow 流式传输管道。在其中一个转换中，DoFn 我想访问外部服务 - 在这种情况下，它是数据存储。

这种初始化步骤有什么最佳实践吗？我不想为每个 processElement 方法调用创建数据存储区连接对象。

Answer 1

在 Dataflow SDK 中，您可以做的最简单的事情是在您的第一个元素中添加一个检查以初始化您的外部服务:

class DatastoreCallingDoFn extends DoFn {

    private ExtServiceHandle handle = null;

    private ExtServiceHandle initializeConnection() {
      // ...
    }

    public void processElement(ProcessContext c) {
      // ... process each element -- setup will have been called
      if (handle == null) {
        handle = initializeConnection();
      }
      // Process elements
    }
}

---

如果您正在使用 Beam，您可以使用 @Setup 装饰器来装饰您的 DoFn 中的一个函数来设置您的 DoFn，例如初始化数据存储连接。

class DatastoreCallingDoFn extends DoFn {
    @Setup
    public void initializeDatastoreConnection() {
      // ...
    }

    @ProcessElement
    public void processElement(ProcessContext c) {
        // ... process each element -- setup will have been called
    }
}

这类似于 answer in this question .