如何在 Google Dataflow SDK 中使 View 的 asList() 可排序？

Question

我们在使 asList() 方法可排序时遇到问题。

我们认为我们可以通过扩展 View 类并覆盖 asList 方法来做到这一点，但意识到 View 类有一个私有构造函数，所以我们不能这样做。

我们的另一个尝试是在 github 上分叉 Google Dataflow 代码并修改 PCollectionViews 类以使用 Collections.sort 方法返回一个排序列表，如下面的代码 sn-p 所示

@Override
protected List<T> fromElements(Iterable<WindowedValue<T>> contents) {
    Iterable<T> itr = Iterables.transform(
        contents,
        new Function<WindowedValue<T>, T>() {
          @SuppressWarnings("unchecked")
          @Override
          public T apply(WindowedValue<T> input){
            return input.getValue();
          }
        });

    LOG.info("#### About to start sorting the list !");
    List<T> tempList = new ArrayList<T>();
    for (T element : itr) {
      tempList.add(element);
    };
    Collections.sort((List<? extends Comparable>) tempList);
    LOG.info("##### List should now be sorted !");
    return ImmutableList.copyOf(tempList);
}

请注意，我们现在正在对列表进行排序。

这似乎有效，当使用 DirectPipelineRunner 运行时，但当我们尝试使用 BlockingDataflowPipelineRunner 时，似乎没有执行代码更改。

注意：我们实际上重新编译了项目中使用的数据流，但这不起作用。

我们怎样才能做到这一点（作为来自asList 方法调用的排序列表）？

Answer 1

PCollectionViews 中的类不用于扩展。仅支持View.asSingleton、View.asSingleton、View.asIterable、View.asMap 和View.asMultimap 提供的原始视图类型。

要从PCollectionView 获取排序列表，您需要在阅读后对其进行排序。以下代码演示了该模式。

// Assume you have some PCollection
PCollection<MyComparable> myPC = ...;

// Prepare it for side input as a list
final PCollectionView<List<MyComparable> myView = myPC.apply(View.asList());

// Side input the list and sort it
someOtherValue.apply(
    ParDo.withSideInputs(myView).of(
        new DoFn<A, B>() {
          @Override
          public void processElement(ProcessContext ctx) {
            List<MyComparable> tempList =
                Lists.newArrayList(ctx.sideInput(myView));
            Collections.sort(tempList);
            // do whatever you want with sorted list 
          }
        }));

当然，您可能不想重复排序，这取决于排序的成本与将其具体化为新的PCollection 的成本，因此您可以输出此值并将其作为新的侧输入轻松读取：

// Side input the list, sort it, and put it in a PCollection
PCollection<List<MyComparable>> sortedSingleton = Create.<Void>of(null).apply(
    ParDo.withSideInputs(myView).of(
        new DoFn<Void, B>() {
          @Override
          public void processElement(ProcessContext ctx) {
            List<MyComparable> tempList =
                Lists.newArrayList(ctx.sideInput(myView));
            Collections.sort(tempList);
            ctx.output(tempList);
          }
        }));

// Prepare it for side input as a list
final PCollectionView<List<MyComparable>> sortedView =
    sortedSingleton.apply(View.asSingleton());

someOtherValue.apply(
    ParDo.withSideInputs(sortedView).of(
        new DoFn<A, B>() {
          @Override
          public void processElement(ProcessContext ctx) {
            ... ctx.sideInput(sortedView) ...
            // do whatever you want with sorted list 
          }
        }));

您可能还对不受支持的 sorter contrib 模块感兴趣，该模块可以同时使用内存和本地磁盘进行更大的排序。

Answer 2

我们尝试按照 Ken Knowles 建议的方式进行操作。大型数据集存在问题。如果 tempList 很大（因此排序需要一些可测量的时间，因为它与 O(n * log n) 成比例）并且如果“someOtherValue”PCollection 中有数百万个元素，那么我们不必要地重新排序相同的列表数百万次。在将列表传递给 someOtherValue.apply 的 DoFn 之前，我们应该能够排序 ONCE 和 FIRST。