RedisIO { public static Read read() { return new AutoValue_RedisIO_Read.Builder() .setConnectionConfiguration(RedisConnectionConfiguration.create()) .setKeyPattern("*") .setBatchSize(1000) .setOutputParallelization(true) .build(); } private  RedisIO(); static Read read(); @Deprecated static ReadAll readAll(); static ReadKeyPatterns readKeyPatterns(); static Write write();  }
@Test public void testRead() { List<KV<String, String>> data = buildIncrementalData("bulkread", 10); data.forEach(kv -> client.set(kv.getKey(), kv.getValue())); PCollection<KV<String, String>> read = p.apply( "Read", RedisIO.read() .withEndpoint(REDIS_HOST, port) .withKeyPattern("bulkread*") .withBatchSize(10)); PAssert.that(read).containsInAnyOrder(data); p.run(); }
@Test public void testReadWithKeyPattern() { List<KV<String, String>> data = buildIncrementalData("pattern", 10); data.forEach(kv -> client.set(kv.getKey(), kv.getValue())); PCollection<KV<String, String>> read = p.apply("Read", RedisIO.read().withEndpoint(REDIS_HOST, port).withKeyPattern("pattern*")); PAssert.that(read).containsInAnyOrder(data); PCollection<KV<String, String>> readNotMatch = p.apply( "ReadNotMatch", RedisIO.read().withEndpoint(REDIS_HOST, port).withKeyPattern("foobar*")); PAssert.thatSingleton(readNotMatch.apply(Count.globally())).isEqualTo(0L); p.run(); }