public BeamRowToBigtableMutation(Map<String, Set<String>> familyColumnMapping) {
    columnFamilyMapping =
        familyColumnMapping.entrySet().stream()
            .flatMap(kv -> kv.getValue().stream().map(column -> KV.of(column, kv.getKey())))
            .collect(toMap(KV::getKey, KV::getValue));
  }
@Test
  @SuppressWarnings("unchecked")
  public void testBeamRowToBigtableMutation() {
    Map<String, Set<String>> columnsMapping =
        ImmutableMap.of(
            FAMILY_TEST, ImmutableSet.of(BOOL_COLUMN, LONG_COLUMN, STRING_COLUMN, DOUBLE_COLUMN));

    PCollection<KV<ByteString, Iterable<Mutation>>> rows =
        pipeline
            .apply(Create.of(row(1), row(2)))
            .setRowSchema(TEST_FLAT_SCHEMA)
            .apply(new BeamRowToBigtableMutation(columnsMapping))
            .apply(MapElements.via(new SortMutationsByColumns()));

    PAssert.that(rows).containsInAnyOrder(rowMutation(1), rowMutation(2));
    pipeline.run().waitUntilFinish();
  }