import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
private static Map<TupleTag<?>, BroadcastHelper<?>> getSideInputs(
    List<PCollectionView<?>> views,
    EvaluationContext context) {
  if (views == null) {
    return ImmutableMap.of();
  } else {
    Map<TupleTag<?>, BroadcastHelper<?>> sideInputs = Maps.newHashMap();
    for (PCollectionView<?> view : views) {
      Iterable<? extends WindowedValue<?>> collectionView = context.getPCollectionView(view);
      Coder<Iterable<WindowedValue<?>>> coderInternal = view.getCoderInternal();
      @SuppressWarnings("unchecked")
      BroadcastHelper<?> helper =
          BroadcastHelper.create((Iterable<WindowedValue<?>>) collectionView, coderInternal);
      //broadcast side inputs
      helper.broadcast(context.getSparkContext());
      sideInputs.put(view.getTagInternal(), helper);
    }
    return sideInputs;
  }
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
public static PCollection<KV<GATKRead, Variant>> pair(PCollection<GATKRead> pRead, PCollection<Variant> pVariant) {

        PCollection<KV<VariantShard, GATKRead>> vkReads = pRead.apply(new KeyReadsByOverlappingVariantShard());
        PCollection<KV<VariantShard, Variant>> vkVariants =
                pVariant.apply(new KeyVariantByOverlappingVariantShard());

        // GroupBy VariantShard
        final TupleTag<Variant> variantTag = new TupleTag<>();
        final TupleTag<GATKRead> readTag = new TupleTag<>();
        PCollection<KV<VariantShard, CoGbkResult>> coGbkInput = KeyedPCollectionTuple
                .of(variantTag, vkVariants)
                .and(readTag, vkReads).apply(CoGroupByKey.<VariantShard>create());

        // GroupBy Read
        return coGbkInput.apply(ParDo.of(
                new DoFn<KV<VariantShard, CoGbkResult>, KV<GATKRead, Variant>>() {
                    private static final long serialVersionUID = 1L;
                    @Override
                    public void processElement(ProcessContext c) throws Exception {
                        Iterable<Variant> kVariants = c.element().getValue().getAll(variantTag);
                        Iterable<GATKRead> kReads = c.element().getValue().getAll(readTag);
                        // Compute overlap.
                        for (GATKRead r : kReads) {
                            SimpleInterval readInterval = new SimpleInterval(r);
                            for (Variant v : kVariants) {
                                if (readInterval.overlaps(v)) {
                                    c.output(KV.of(r, v));
                                }
                            }
                        }
                    }
                })).setName("PairReadsAndVariants_GroupByRead");

    }
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
MultiDoFnFunction(
    DoFn<I, O> fn,
    SparkRuntimeContext runtimeContext,
    TupleTag<O> mainOutputTag,
    Map<TupleTag<?>, BroadcastHelper<?>> sideInputs) {
  this.mFunction = fn;
  this.mRuntimeContext = runtimeContext;
  this.mMainOutputTag = mainOutputTag;
  this.mSideInputs = sideInputs;
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
@Override
public Iterable<Tuple2<TupleTag<?>, WindowedValue<?>>>
    call(Iterator<WindowedValue<I>> iter) throws Exception {
  ProcCtxt ctxt = new ProcCtxt(mFunction, mRuntimeContext, mSideInputs);
  mFunction.startBundle(ctxt);
  ctxt.setup();
  return ctxt.getOutputIterable(iter, mFunction);
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
@Override
protected Iterator<Tuple2<TupleTag<?>, WindowedValue<?>>> getOutputIterator() {
  return Iterators.transform(outputs.entries().iterator(),
      new Function<Map.Entry<TupleTag<?>, WindowedValue<?>>,
          Tuple2<TupleTag<?>, WindowedValue<?>>>() {
    @Override
    public Tuple2<TupleTag<?>, WindowedValue<?>> apply(Map.Entry<TupleTag<?>,
        WindowedValue<?>> input) {
      return new Tuple2<TupleTag<?>, WindowedValue<?>>(input.getKey(), input.getValue());
    }
  });
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
SparkProcessContext(DoFn<I, O> fn,
    SparkRuntimeContext runtime,
    Map<TupleTag<?>, BroadcastHelper<?>> sideInputs) {
  fn.super();
  this.fn = fn;
  this.mRuntimeContext = runtime;
  this.mSideInputs = sideInputs;
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
@Override
public <T> void sideOutput(TupleTag<T> tupleTag, T t) {
  String message = "sideOutput is an unsupported operation for doFunctions, use a " +
      "MultiDoFunction instead.";
  LOG.warn(message);
  throw new UnsupportedOperationException(message);
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
@Override
public <T> void sideOutputWithTimestamp(TupleTag<T> tupleTag, T t, Instant instant) {
  String message =
      "sideOutputWithTimestamp is an unsupported operation for doFunctions, use a " +
          "MultiDoFunction instead.";
  LOG.warn(message);
  throw new UnsupportedOperationException(message);
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
private static <I, O> TransformEvaluator<ParDo.BoundMulti<I, O>> multiDo() {
  return new TransformEvaluator<ParDo.BoundMulti<I, O>>() {
    @Override
    public void evaluate(ParDo.BoundMulti<I, O> transform, EvaluationContext context) {
      TupleTag<O> mainOutputTag = MULTIDO_FG.get("mainOutputTag", transform);
      MultiDoFnFunction<I, O> multifn = new MultiDoFnFunction<>(
          transform.getFn(),
          context.getRuntimeContext(),
          mainOutputTag,
          getSideInputs(transform.getSideInputs(), context));

      @SuppressWarnings("unchecked")
      JavaRDDLike<WindowedValue<I>, ?> inRDD =
          (JavaRDDLike<WindowedValue<I>, ?>) context.getInputRDD(transform);
      JavaPairRDD<TupleTag<?>, WindowedValue<?>> all = inRDD
          .mapPartitionsToPair(multifn)
          .cache();

      PCollectionTuple pct = context.getOutput(transform);
      for (Map.Entry<TupleTag<?>, PCollection<?>> e : pct.getAll().entrySet()) {
        @SuppressWarnings("unchecked")
        JavaPairRDD<TupleTag<?>, WindowedValue<?>> filtered =
            all.filter(new TupleTagFilter(e.getKey()));
        @SuppressWarnings("unchecked")
        // Object is the best we can do since different outputs can have different tags
        JavaRDD<WindowedValue<Object>> values =
            (JavaRDD<WindowedValue<Object>>) (JavaRDD<?>) filtered.values();
        context.setRDD(e.getValue(), values);
      }
    }
  };
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
/**
 * @param fn         DoFunction to be wrapped.
 * @param runtime    Runtime to apply function in.
 * @param sideInputs Side inputs used in DoFunction.
 */
public DoFnFunction(DoFn<I, O> fn,
             SparkRuntimeContext runtime,
             Map<TupleTag<?>, BroadcastHelper<?>> sideInputs) {
  this.mFunction = fn;
  this.mRuntimeContext = runtime;
  this.mSideInputs = sideInputs;
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
public FileToState(TupleTag<GCPResourceErrorInfo> tag) {
  errorOutputTag = tag;
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
public ExtractState(TupleTag<GCPResourceErrorInfo> tag) {
  errorOutputTag = tag;
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
public CreateWorkPacketsDoFn(WorkPacketConfig workPacketView, TupleTag<Integer> counter) {
  this.workPacketView = workPacketView;
  this.counter = counter;
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
public DistributeWorkDataDoFn(WorkPacketConfig workPacketView, TupleTag<Integer> tag) {
  this.workPacketView = workPacketView;
  this.tag  = tag;
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
ProcCtxt(DoFn<I, O> fn, SparkRuntimeContext runtimeContext, Map<TupleTag<?>,
    BroadcastHelper<?>> sideInputs) {
  super(fn, runtimeContext, sideInputs);
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
@Override
public synchronized <T> void sideOutput(TupleTag<T> tag, T t) {
  outputs.put(tag, windowedValue.withValue(t));
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
@Override
public <T> void sideOutputWithTimestamp(TupleTag<T> tupleTag, T t, Instant instant) {
  outputs.put(tupleTag, WindowedValue.of(t, instant,
      windowedValue.getWindows(), windowedValue.getPane()));
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
private TupleTagFilter(TupleTag<V> tag) {
  this.tag = tag;
}
 

import com.google.cloud.dataflow.sdk.values.TupleTag; //导入依赖的package包/类
@Override
public Boolean call(Tuple2<TupleTag<V>, WindowedValue<?>> input) {
  return tag.equals(input._1());
}