KvSwap extends PTransform<PCollection<KV<K, V>>, PCollection<KV<V, K>>> { private KvSwap() {} private  KvSwap(); static KvSwap<K, V> create(); @Override PCollection<KV<V, K>> expand(PCollection<KV<K, V>> in);  }
@Test @Category(ValidatesRunner.class) public void testKvSwap() { PCollection<KV<String, Integer>> input = p.apply( Create.of(Arrays.asList(TABLE)) .withCoder( KvCoder.of( StringUtf8Coder.of(), NullableCoder.of(BigEndianIntegerCoder.of())))); PCollection<KV<Integer, String>> output = input.apply(KvSwap.create()); PAssert.that(output) .containsInAnyOrder( KV.of(1, "one"), KV.of(2, "two"), KV.of(3, "three"), KV.of(4, "four"), KV.of(4, "dup"), KV.of(5, "dup"), KV.of((Integer) null, "null")); p.run(); }