public static PTransform<PCollection<? extends KV<?, ?>>, PCollection<String>> kvs() {
    return kvs(",");
  }
@Test
  @Category(NeedsRunner.class)
  public void testToStringKV() {
    ArrayList<KV<String, Integer>> kvs = new ArrayList<>();
    kvs.add(KV.of("one", 1));
    kvs.add(KV.of("two", 2));

    ArrayList<String> expected = new ArrayList<>();
    expected.add("one,1");
    expected.add("two,2");

    PCollection<KV<String, Integer>> input = p.apply(Create.of(kvs));
    PCollection<String> output = input.apply(ToString.kvs());
    PAssert.that(output).containsInAnyOrder(expected);
    p.run();
  }