public static <K, V> KvSwap<K, V> create() {
    return new KvSwap<>();
  }
@Test
  @Category(NeedsRunner.class)
  public void testKvSwapEmpty() {
    PCollection<KV<String, Integer>> input =
        p.apply(
            Create.of(Arrays.asList(EMPTY_TABLE))
                .withCoder(KvCoder.of(StringUtf8Coder.of(), BigEndianIntegerCoder.of())));

    PCollection<KV<Integer, String>> output = input.apply(KvSwap.create());

    PAssert.that(output).empty();
    p.run();
  }