如何在Google Dataflow SDK中使视图的asList（）可排序？

Question

我们在使aslist()方法可排序时遇到了一个问题。

我们认为只需扩展View类并重写asList方法就可以做到这一点，但意识到View类有一个私有构造函数，所以我们无法做到这一点。

java prettyprint-override">@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);
}

我们如何能够实现这一点（从aslist方法调用中获得排序列表）？

Answer 1

pCollectionViews中的类不用于扩展。只支持view.assingleton、view.assingletonview.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 
          }
        }));

您还可能对使用内存和本地磁盘进行较大排序的不受支持的排序器contrib模块感兴趣。