Skip to content

Add topology stream-awareness to storm-redis #1760

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
mo-getter wants to merge 2 commits into from
Closed

Add topology stream-awareness to storm-redis #1760

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
28e31f3
Add topology stream-awareness to storm-redis.
Nov 3, 2016
ab1cb55
Merge remote-tracking branch 'upstream/master'
Nov 4, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 55 additions & 5 deletions external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisFilterBolt.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,17 +17,20 @@
*/
package org.apache.storm.redis.bolt;

import java.util.List;
import org.apache.storm.redis.common.config.JedisClusterConfig;
import org.apache.storm.redis.common.config.JedisPoolConfig;
import org.apache.storm.redis.common.mapper.BasicStreamMapper;
import org.apache.storm.redis.common.mapper.DefaultStreamMapper;
import org.apache.storm.redis.common.mapper.RedisDataTypeDescription;
import org.apache.storm.redis.common.mapper.RedisFilterMapper;
import org.apache.storm.redis.common.mapper.StreamMapper;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import redis.clients.jedis.GeoCoordinate;
import redis.clients.jedis.JedisCommands;

import java.util.List;

/**
* Basic bolt for querying from Redis and filters out if key/field doesn't exist.
* If key/field exists on Redis, this bolt just forwards input tuple to default stream.
Expand All @@ -45,18 +48,41 @@
*/
public class RedisFilterBolt extends AbstractRedisBolt {
private final RedisFilterMapper filterMapper;
private final StreamMapper streamMapper;
private final RedisDataTypeDescription.RedisDataType dataType;
private final String additionalKey;

/**
* Constructor for single Redis environment (JedisPool)
* Constructor for single Redis environment (JedisPool).
* Tuples will be emitted to Storm's default streamId.
* @param config configuration for initializing JedisPool
* @param filterMapper mapper containing which datatype, query key that Bolt uses
*/
public RedisFilterBolt(JedisPoolConfig config, RedisFilterMapper filterMapper) {

@revans2 revans2 Nov 3, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This now creates a lot of coupling between the filter mapper and the stream mapper. Simply because the Filter Mapper is the one that declares the output fields.

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        filterMapper.declareOutputFields(declarer);
    }

So either we need to embrace the coupling and have StreamMapper also be a FilterMapper. (which would require some documentation) or we find a way to fake out FilterMapper and have it declare multiple outputs for what the StreamMapper wants.

I prefer the first one, because it seems like it would be more flexible.

@mo-getter mo-getter Nov 4, 2016

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for taking a look at this. I see your point, and I agree it makes the most sense for both to be done in the same interface. TL;DR is at the bottom ;-)

The same problem also applies to RedisLookupMapper, which also defines declareOutputFields separately. I just came across STORM-1953.

About your first option, would it make more sense for a FilterMapper to be a StreamMapper, rather than a StreamMapper be a FilterMapper? I'm afraid that making StreamMapper a FilterMapper would introduce too much ambiguity in the lookup and filter bolts (if the constructors accepted both), since we'd have to rely on only the docs to define which object's declareOutputFields would be called. It also would make STORM-1953 worse. Or did you mean the bolts only accept FilterMapper, and have something like this in execute:

if (filterMapper instanceof StreamMapper) {
    String streamId = ((StreamMapper) filterMapper).getStreamId(input, value);
    collector.emit(streamId, input, value);
} else {
    collector.emit(input, value);
}

Either way, if you combine them, one downside is that the provided convenience StreamMapper implementations would have to be sacrificed. Making them abstract probably wouldn't be worth it for something like just specifying the stream.

In case you want to see what having FilterMapper and LookupMapper also extend StreamMapper looks like, I implemented that in a branch here. The flexibility to dynamically choose a stream is there, but the problem is that the trident-related classes also use LookupMapper, and have no need to declare a streamId, yet users will have to implement this method in their LookupMappers. Just returning null is one [not so good] option here, and is also an option when using LookupMapper for bolts (in which case, the existing behavior of emitting to the default stream is maintained).

TL;DR: I can't think of a great solution for what you mentioned, while maintaining user-friendliness of the API, without totally redoing the Mapper interfaces, i.e. STORM-1953. On the other hand, the above commit does maintain full backward compatibility and is probably most convenient for users!

@revans2 revans2 Nov 4, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you want to target this for master? Or do you also want this in 1.x? If it is just master we can play some games with a default method implementation in the FilterMapper interface.

If you want it in 1.x I would suggest that we leave FilterMapper untouched and create a LookupMapper that also has the same, or similar methods to FilterMapper, but is not a FilterMapper. Then you can have a wrapper class that is a LookupMapper, but takes a FilterMapper. The code could then wrap any FilterMapper passed in, and just use the LookupMapper interface.

I prefer the first one with the default methods because it reduce the number of classes and interfaces but is also binary compatible. If we are not on java 8 like 1.x then we cannot use default methods.

this(config, filterMapper, new DefaultStreamMapper());
}

/**
* Constructor for single Redis environment (JedisPool).
* @param config configuration for initializing JedisCluster
* @param filterMapper mapper containing which datatype, query key that Bolt uses
* @param streamId the stream to which tuples that make it through the filter should be emitted.
*/
public RedisFilterBolt(JedisPoolConfig config, RedisFilterMapper filterMapper, String streamId) {
this(config, filterMapper, new BasicStreamMapper(streamId));
}

/**
* Constructor for single Redis environment (JedisPool).
* @param config configuration for initializing JedisPool
* @param filterMapper mapper containing which datatype, query key that Bolt uses
* @param streamMapper mapper to which stream a given Tuple/Values pair should be emitted.
*/
public RedisFilterBolt(JedisPoolConfig config, RedisFilterMapper filterMapper, StreamMapper streamMapper) {
super(config);

this.filterMapper = filterMapper;
this.streamMapper = streamMapper;

RedisDataTypeDescription dataTypeDescription = filterMapper.getDataTypeDescription();
this.dataType = dataTypeDescription.getDataType();
Expand All @@ -69,14 +95,36 @@ public RedisFilterBolt(JedisPoolConfig config, RedisFilterMapper filterMapper) {
}

/**
* Constructor for Redis Cluster environment (JedisCluster)
* Constructor for Redis Cluster environment (JedisCluster).
* Tuples will be emitted to Storm's default streamId.
* @param config configuration for initializing JedisCluster
* @param filterMapper mapper containing which datatype, query key that Bolt uses
*/
public RedisFilterBolt(JedisClusterConfig config, RedisFilterMapper filterMapper) {
this(config, filterMapper, new DefaultStreamMapper());
}

/**
* Constructor for Redis Cluster environment (JedisCluster).
* @param config configuration for initializing JedisCluster
* @param filterMapper mapper containing which datatype, query key that Bolt uses
* @param streamId the stream to which tuples that make it through the filter should be emitted.
*/
public RedisFilterBolt(JedisClusterConfig config, RedisFilterMapper filterMapper, String streamId) {
this(config, filterMapper, new BasicStreamMapper(streamId));
}

/**
* Constructor for Redis Cluster environment (JedisCluster).
* @param config configuration for initializing JedisCluster
* @param filterMapper mapper containing which datatype, query key that Bolt uses
* @param streamMapper mapper to which stream a given Tuple/Values pair should be emitted.
*/
public RedisFilterBolt(JedisClusterConfig config, RedisFilterMapper filterMapper, StreamMapper streamMapper) {
super(config);

this.filterMapper = filterMapper;
this.streamMapper = streamMapper;

RedisDataTypeDescription dataTypeDescription = filterMapper.getDataTypeDescription();
this.dataType = dataTypeDescription.getDataType();
Expand Down Expand Up @@ -126,7 +174,9 @@ public void execute(Tuple input) {
}

if (found) {
collector.emit(input, input.getValues());
Values values = new Values(input.getValues().toArray());
String streamId = streamMapper.getStreamId(input, values);
collector.emit(streamId, input, values);
}

collector.ack(input);
Expand Down
66 changes: 57 additions & 9 deletions external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisLookupBolt.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,51 +17,98 @@
*/
package org.apache.storm.redis.bolt;

import java.util.List;
import org.apache.storm.redis.common.config.JedisClusterConfig;
import org.apache.storm.redis.common.config.JedisPoolConfig;
import org.apache.storm.redis.common.mapper.BasicStreamMapper;
import org.apache.storm.redis.common.mapper.DefaultStreamMapper;
import org.apache.storm.redis.common.mapper.RedisDataTypeDescription;
import org.apache.storm.redis.common.mapper.RedisLookupMapper;
import org.apache.storm.redis.common.mapper.StreamMapper;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.apache.storm.redis.common.mapper.RedisDataTypeDescription;
import org.apache.storm.redis.common.mapper.RedisLookupMapper;
import org.apache.storm.redis.common.config.JedisClusterConfig;
import org.apache.storm.redis.common.config.JedisPoolConfig;
import redis.clients.jedis.JedisCommands;

import java.util.List;

/**
* Basic bolt for querying from Redis and emits response as tuple.
* <p/>
* Various data types are supported: STRING, LIST, HASH, SET, SORTED_SET, HYPER_LOG_LOG, GEO
*/
public class RedisLookupBolt extends AbstractRedisBolt {
private final RedisLookupMapper lookupMapper;
private final StreamMapper streamMapper;
private final RedisDataTypeDescription.RedisDataType dataType;
private final String additionalKey;

/**
* Constructor for single Redis environment (JedisPool)
* Constructor for single Redis environment (JedisPool).
* Emits tuples to Storm's default stream.
* @param config configuration for initializing JedisPool
* @param lookupMapper mapper containing which datatype, query key, output key that Bolt uses
*/
public RedisLookupBolt(JedisPoolConfig config, RedisLookupMapper lookupMapper) {
this(config, lookupMapper, new DefaultStreamMapper());
}

/**
* Constructor for single Redis environment (JedisPool).
* @param config configuration for initializing JedisPool
* @param lookupMapper mapper containing which datatype, query key, output key that Bolt uses
* @param streamId the streamId to which this bolt should emit tuples
*/
public RedisLookupBolt(JedisPoolConfig config, RedisLookupMapper lookupMapper, String streamId) {
this(config, lookupMapper, new BasicStreamMapper(streamId));
}

/**
* Constructor for single Redis environment (JedisPool).
* @param config configuration for initializing JedisPool
* @param lookupMapper mapper containing which datatype, query key, output key that Bolt uses
* @param streamMapper mapper to which stream a given Tuple/Values pair should be emitted.
*/
public RedisLookupBolt(JedisPoolConfig config, RedisLookupMapper lookupMapper, StreamMapper streamMapper) {
super(config);

this.lookupMapper = lookupMapper;
this.streamMapper = streamMapper;

RedisDataTypeDescription dataTypeDescription = lookupMapper.getDataTypeDescription();
this.dataType = dataTypeDescription.getDataType();
this.additionalKey = dataTypeDescription.getAdditionalKey();
}

/**
* Constructor for Redis Cluster environment (JedisCluster)
* Constructor for Redis Cluster environment (JedisCluster).
* Emits tuples to Storm's default stream.
* @param config configuration for initializing JedisCluster
* @param lookupMapper mapper containing which datatype, query key, output key that Bolt uses
*/
public RedisLookupBolt(JedisClusterConfig config, RedisLookupMapper lookupMapper) {
this(config, lookupMapper, new DefaultStreamMapper());
}

/**
* Constructor for single Redis environment (JedisPool).
* @param config configuration for initializing JedisPool
* @param lookupMapper mapper containing which datatype, query key, output key that Bolt uses
* @param streamId the streamId to which this bolt should emit tuples
*/
public RedisLookupBolt(JedisClusterConfig config, RedisLookupMapper lookupMapper, String streamId) {
this(config, lookupMapper, new BasicStreamMapper(streamId));
}

/**
* Constructor for Redis Cluster environment (JedisCluster).
* @param config configuration for initializing JedisCluster
* @param lookupMapper mapper containing which datatype, query key, output key that Bolt uses
* @param streamMapper mapper to which stream a given Tuple/Values pair should be emitted.
*/
public RedisLookupBolt(JedisClusterConfig config, RedisLookupMapper lookupMapper, StreamMapper streamMapper) {
super(config);

this.lookupMapper = lookupMapper;
this.streamMapper = streamMapper;

RedisDataTypeDescription dataTypeDescription = lookupMapper.getDataTypeDescription();
this.dataType = dataTypeDescription.getDataType();
Expand Down Expand Up @@ -115,7 +162,8 @@ public void execute(Tuple input) {

List<Values> values = lookupMapper.toTuple(input, lookupValue);
for (Values value : values) {
collector.emit(input, value);
String streamId = streamMapper.getStreamId(input, value);
collector.emit(streamId, input, value);
}

collector.ack(input);
Expand Down
38 changes: 38 additions & 0 deletions ...nal/storm-redis/src/main/java/org/apache/storm/redis/common/mapper/BasicStreamMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
/*
* Copyright 2016 The Apache Software Foundation.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.storm.redis.common.mapper;

import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
* A StreamMapper implementation which always returns the streamId with
* which it was constructed.
*/
public class BasicStreamMapper implements StreamMapper {

private final String streamId;

public BasicStreamMapper(String streamId) {
this.streamId = streamId;
}

@Override
public String getStreamId(Tuple input, Values values) {
return streamId;
}

}
29 changes: 29 additions & 0 deletions ...l/storm-redis/src/main/java/org/apache/storm/redis/common/mapper/DefaultStreamMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
/*
* Copyright 2016 The Apache Software Foundation.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.storm.redis.common.mapper;

import org.apache.storm.utils.Utils;

/**
* A RedisStreamMapper implementation which always returns Storm's default streamId.
*/
public final class DefaultStreamMapper extends BasicStreamMapper {

public DefaultStreamMapper() {
super(Utils.DEFAULT_STREAM_ID);
}

}
32 changes: 32 additions & 0 deletions ...orm-redis/src/main/java/org/apache/storm/redis/common/mapper/InputSourceStreamMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
/*
* Copyright 2016 The Apache Software Foundation.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.storm.redis.common.mapper;

import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
* Returns the streamId of the input tuple's source stream. In other words,
* tuples will be emitted to the same streamId from which they came.
*/
public class InputSourceStreamMapper implements StreamMapper {

@Override
public String getStreamId(Tuple input, Values values) {
return input.getSourceStreamId();
}

}
38 changes: 38 additions & 0 deletions external/storm-redis/src/main/java/org/apache/storm/redis/common/mapper/StreamMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
/*
* Copyright 2016 The Apache Software Foundation.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.storm.redis.common.mapper;

import java.io.Serializable;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
* StreamMapper is for specifying the stream to which Values should be
* emitted, based on the input tuple and/or the already mapped output values
* (about to be emitted).
*/
public interface StreamMapper extends Serializable {

/**
* Gets the streamId based on the input Tuple and/or the values to be
* emitted.
* @param input the original source input tuple
* @param values the Values which were generated by a bolt, based on the input tuple.
* @return the stream id to use for emitting tuples
*/
String getStreamId(Tuple input, Values values);

}