MiguelAnzoWizeline commented on a change in pull request #15549:
URL: https://github.com/apache/beam/pull/15549#discussion_r732093376
##########
File path:
sdks/java/io/redis/src/main/java/org/apache/beam/sdk/io/redis/RedisIO.java
##########
@@ -334,94 +336,85 @@ public void setup() {
public void teardown() {
jedis.close();
}
- }
-
- private static class ReadKeysWithPattern extends BaseReadFn<String> {
- ReadKeysWithPattern(RedisConnectionConfiguration connectionConfiguration) {
- super(connectionConfiguration);
+ @GetInitialRestriction
+ public OffsetRange getInitialRestriction(@Element String pattern) {
+ return new OffsetRange(0, getKeyPatters(pattern).size());
}
@ProcessElement
- public void processElement(ProcessContext c) {
- ScanParams scanParams = new ScanParams();
- scanParams.match(c.element());
-
- String cursor = ScanParams.SCAN_POINTER_START;
- boolean finished = false;
- while (!finished) {
- ScanResult<String> scanResult = jedis.scan(cursor, scanParams);
- List<String> keys = scanResult.getResult();
- for (String k : keys) {
- c.output(k);
+ public void processElement(ProcessContext c,
RestrictionTracker<OffsetRange, Long> tracker) {
Review comment:
Hi @lukecwik after reading the previous implementation again now I see
how it didn’t load all keys into memory using two DoFn's and emiting the keys.
I'm still having problems implementing the proposed solution, if I understand
correctly, you mean tracking the cursor pointer as the restriction starting on
0 (SCAN_POINTER_START) and ending on 0 (as 0 is returned when the pointer
reaches the end of the cursor), is that right? The problem I see and don’t yet
understand there is on splitting, as we don’t know the values of the cursors
until we receive them in the iterations and they do not seem to be sequential
I'm not being able to define the ranges of the splits.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
