Skip to content

Commit bda8fb4

Browse files
pyalexpyalex
committed
configurable flush frequncy in redis sink
1 parent 3200109 commit bda8fb4

5 files changed

Lines changed: 44 additions & 6 deletions

File tree

‎infra/scripts/setup-common-functions.sh‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -98,7 +98,7 @@ start_feast_serving() {
9898

9999
if [ -n "$1" ]; then
100100
echo "Custom Spring application.yml location provided: $1"
101-
export CONFIG_ARG="--spring.config.location=file://$1"
101+
export CONFIG_ARG="--spring.config.location=classpath:/application.yml,file://$1"
102102
fi
103103

104104
nohup java -jar serving/target/feast-serving-$FEAST_BUILD_VERSION.jar $CONFIG_ARG &>/var/log/feast-serving-online.log &

‎infra/scripts/test-end-to-end-redis-cluster.sh‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@ feast:
5656
config:
5757
# Connection string specifies the IP and ports of Redis instances in Redis cluster
5858
connection_string: "localhost:7000,localhost:7001,localhost:7002,localhost:7003,localhost:7004,localhost:7005"
59+
flush_frequency_seconds: 1
5960
# Subscriptions indicate which feature sets needs to be retrieved and used to populate this store
6061
subscriptions:
6162
# Wildcards match all options. No filtering is done.

‎infra/scripts/test-end-to-end.sh‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -74,7 +74,21 @@ if [[ ${ENABLE_AUTH} = "True" ]];
7474
start_feast_core
7575
fi
7676

77-
start_feast_serving
77+
cat <<EOF > /tmp/serving.warehouse.application.yml
78+
feast:
79+
stores:
80+
- name: online
81+
type: REDIS
82+
config:
83+
host: localhost
84+
port: 6379
85+
flush_frequency_seconds: 1
86+
subscriptions:
87+
- name: "*"
88+
project: "*"
89+
EOF
90+
91+
start_feast_serving /tmp/serving.warehouse.application.yml
7892
install_python_with_miniconda_and_feast_sdk
7993

8094
print_banner "Running end-to-end tests with pytest at 'tests/e2e'"

‎protos/feast/core/Store.proto‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,8 @@ message Store {
108108
int32 initial_backoff_ms = 3;
109109
// Optional. Maximum total number of retries for connecting to Redis. Default to zero retries.
110110
int32 max_retries = 4;
111+
// Optional. how often flush data to redis
112+
int32 flush_frequency_seconds = 5;
111113
}
112114

113115
message BigQueryConfig {
@@ -129,6 +131,8 @@ message Store {
129131
string connection_string = 1;
130132
int32 initial_backoff_ms = 2;
131133
int32 max_retries = 3;
134+
// Optional. how often flush data to redis
135+
int32 flush_frequency_seconds = 4;
132136
}
133137

134138
message Subscription {

‎storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisFeatureSink.java‎

Lines changed: 23 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
import org.apache.beam.sdk.values.KV;
3636
import org.apache.beam.sdk.values.PCollection;
3737
import org.apache.beam.sdk.values.PCollectionView;
38+
import org.joda.time.Duration;
3839

3940
@AutoValue
4041
public abstract class RedisFeatureSink implements FeatureSink {
@@ -113,11 +114,29 @@ public PCollection<FeatureSetReference> prepareWrite(
113114
@Override
114115
public PTransform<PCollection<FeatureRow>, WriteResult> writer() {
115116
if (getRedisClusterConfig() != null) {
116-
return new RedisCustomIO.Write(
117-
new RedisClusterIngestionClient(getRedisClusterConfig()), getSpecsView());
117+
RedisCustomIO.Write writer =
118+
new RedisCustomIO.Write(
119+
new RedisClusterIngestionClient(getRedisClusterConfig()), getSpecsView());
120+
121+
if (getRedisClusterConfig().getFlushFrequencySeconds() > 0) {
122+
writer =
123+
writer.withFlushFrequency(
124+
Duration.standardSeconds(getRedisClusterConfig().getFlushFrequencySeconds()));
125+
}
126+
127+
return writer;
118128
} else if (getRedisConfig() != null) {
119-
return new RedisCustomIO.Write(
120-
new RedisStandaloneIngestionClient(getRedisConfig()), getSpecsView());
129+
RedisCustomIO.Write writer =
130+
new RedisCustomIO.Write(
131+
new RedisStandaloneIngestionClient(getRedisConfig()), getSpecsView());
132+
133+
if (getRedisConfig().getFlushFrequencySeconds() > 0) {
134+
writer =
135+
writer.withFlushFrequency(
136+
Duration.standardSeconds(getRedisConfig().getFlushFrequencySeconds()));
137+
}
138+
139+
return writer;
121140
} else {
122141
throw new RuntimeException(
123142
"At least one RedisConfig or RedisClusterConfig must be provided to Redis Sink");

0 commit comments

Comments
 (0)