public static Read read() {
    return new AutoValue_RedisIO_Read.Builder()
        .setConnectionConfiguration(RedisConnectionConfiguration.create())
        .setKeyPattern("*")
        .setBatchSize(1000)
        .setOutputParallelization(true)
        .build();
  }
@Test
  public void testReadSplitBig() {
    List<KV<String, String>> data = buildIncrementalData("bigset", 1000);
    data.forEach(kv -> client.set(kv.getKey(), kv.getValue()));

    PCollection<KV<String, String>> read =
        p.apply(
            "Read",
            RedisIO.read()
                .withEndpoint(REDIS_HOST, port)
                .withKeyPattern("bigset*")
                .withBatchSize(8));
    PAssert.that(read).containsInAnyOrder(data);
    p.run();
  }