Skip to content

Commit b1a4048

Browse files
terryyylimpyalex
terryyylim
and
pyalex
authored
Add support for BigTable Online Storage (#17)
Signed-off-by: Terence Lim <terencelimxp@gmail.com> Co-authored-by: Oleksii Moskalenko <moskalenko.alexey@gmail.com>
1 parent 8541f36 commit b1a4048

22 files changed

Lines changed: 1520 additions & 219 deletions

File tree

‎common-test/src/main/java/feast/common/it/DataGenerator.java‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,28 @@ public static Triple<String, String, Boolean> getDefaultSubscription() {
5353
return defaultSubscription;
5454
}
5555

56+
public static String valueToString(ValueProto.Value v) {
57+
String stringRepr;
58+
switch (v.getValCase()) {
59+
case STRING_VAL:
60+
stringRepr = v.getStringVal();
61+
break;
62+
case INT64_VAL:
63+
stringRepr = String.valueOf(v.getInt64Val());
64+
break;
65+
case INT32_VAL:
66+
stringRepr = String.valueOf(v.getInt32Val());
67+
break;
68+
case BYTES_VAL:
69+
stringRepr = v.getBytesVal().toString();
70+
break;
71+
default:
72+
throw new RuntimeException("Type is not supported to be entity");
73+
}
74+
75+
return stringRepr;
76+
}
77+
5678
public static StoreProto.Store getDefaultStore() {
5779
return defaultStore;
5880
}
@@ -247,6 +269,18 @@ public static ServingAPIProto.GetOnlineFeaturesRequestV2.EntityRow createEntityR
247269
.build();
248270
}
249271

272+
public static ServingAPIProto.GetOnlineFeaturesRequestV2.EntityRow createCompoundEntityRow(
273+
ImmutableMap<String, ValueProto.Value> entityNameValues, long seconds) {
274+
ServingAPIProto.GetOnlineFeaturesRequestV2.EntityRow.Builder entityRow =
275+
ServingAPIProto.GetOnlineFeaturesRequestV2.EntityRow.newBuilder()
276+
.setTimestamp(Timestamp.newBuilder().setSeconds(seconds));
277+
278+
entityNameValues.entrySet().stream()
279+
.forEach(entry -> entityRow.putFields(entry.getKey(), entry.getValue()));
280+
281+
return entityRow.build();
282+
}
283+
250284
public static DataSource createKinesisDataSourceSpec(
251285
String region, String streamName, String classPath, String timestampColumn) {
252286
return DataSource.newBuilder()

‎serving/pom.xml‎

Lines changed: 36 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -84,12 +84,26 @@
8484
<version>${project.version}</version>
8585
</dependency>
8686

87+
<dependency>
88+
<groupId>dev.feast</groupId>
89+
<artifactId>feast-storage-connector-bigtable</artifactId>
90+
<version>${project.version}</version>
91+
</dependency>
92+
8793
<dependency>
8894
<groupId>dev.feast</groupId>
8995
<artifactId>feast-common</artifactId>
9096
<version>${project.version}</version>
9197
</dependency>
92-
98+
99+
<dependency>
100+
<groupId>com.google.cloud</groupId>
101+
<artifactId>google-cloud-bigtable-emulator</artifactId>
102+
<version>0.130.2</version>
103+
<scope>test</scope>
104+
</dependency>
105+
106+
93107
<!-- TODO: SLF4J is being used via Lombok, but also jog4j - pick one -->
94108
<dependency>
95109
<groupId>org.slf4j</groupId>
@@ -129,6 +143,14 @@
129143
<artifactId>spring-boot-starter-actuator</artifactId>
130144
</dependency>
131145

146+
<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-test -->
147+
<dependency>
148+
<groupId>org.springframework.boot</groupId>
149+
<artifactId>spring-boot-test</artifactId>
150+
<version>2.3.1.RELEASE</version>
151+
<scope>test</scope>
152+
</dependency>
153+
132154
<!--compile "io.grpc:grpc-services:${grpc.version}"-->
133155
<dependency>
134156
<groupId>io.grpc</groupId>
@@ -234,6 +256,13 @@
234256
<scope>test</scope>
235257
</dependency>
236258

259+
<!-- https://mvnrepository.com/artifact/org.apache.avro/avro -->
260+
<dependency>
261+
<groupId>org.apache.avro</groupId>
262+
<artifactId>avro</artifactId>
263+
<version>1.10.2</version>
264+
</dependency>
265+
237266
<!-- Utilities -->
238267
<dependency>
239268
<groupId>com.fasterxml.jackson.dataformat</groupId>
@@ -307,6 +336,12 @@
307336
<version>1.15.1</version>
308337
<scope>test</scope>
309338
</dependency>
339+
<dependency>
340+
<groupId>org.testcontainers</groupId>
341+
<artifactId>gcloud</artifactId>
342+
<version>1.15.2</version>
343+
<scope>test</scope>
344+
</dependency>
310345
<dependency>
311346
<groupId>org.awaitility</groupId>
312347
<artifactId>awaitility</artifactId>

‎serving/src/main/java/feast/serving/config/FeastProperties.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import feast.common.auth.config.SecurityProperties.AuthorizationProperties;
2727
import feast.common.auth.credentials.CoreAuthenticationProperties;
2828
import feast.common.logging.config.LoggingProperties;
29+
import feast.storage.connectors.bigtable.retriever.BigTableStoreConfig;
2930
import feast.storage.connectors.redis.retriever.RedisClusterStoreConfig;
3031
import feast.storage.connectors.redis.retriever.RedisStoreConfig;
3132
import io.lettuce.core.ReadFrom;
@@ -269,7 +270,7 @@ public void setName(String name) {
269270
}
270271

271272
/**
272-
* Gets the store type. Example are REDIS or REDIS_CLUSTER
273+
* Gets the store type. Example are REDIS, REDIS_CLUSTER or BIGTABLE
273274
*
274275
* @return the store type as a String.
275276
*/
@@ -311,6 +312,10 @@ public RedisStoreConfig getRedisConfig() {
311312
Boolean.valueOf(this.config.getOrDefault("ssl", "false")));
312313
}
313314

315+
public BigTableStoreConfig getBigtableConfig() {
316+
return new BigTableStoreConfig(this.config.get("project_id"), this.config.get("instance_id"));
317+
}
318+
314319
/**
315320
* Sets the store config. Please protos/feast/core/Store.proto for the specific options for each
316321
* store.
@@ -323,6 +328,7 @@ public void setConfig(Map<String, String> config) {
323328
}
324329

325330
public enum StoreType {
331+
BIGTABLE,
326332
REDIS,
327333
REDIS_CLUSTER;
328334
}

‎serving/src/main/java/feast/serving/config/ServingServiceConfigV2.java‎

Lines changed: 30 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -16,26 +16,47 @@
1616
*/
1717
package feast.serving.config;
1818

19-
import com.fasterxml.jackson.core.JsonProcessingException;
20-
import com.google.protobuf.InvalidProtocolBufferException;
19+
import com.google.cloud.bigtable.data.v2.BigtableDataClient;
20+
import com.google.cloud.bigtable.data.v2.BigtableDataSettings;
2121
import feast.serving.service.OnlineServingServiceV2;
2222
import feast.serving.service.ServingServiceV2;
2323
import feast.serving.specs.CachedSpecService;
2424
import feast.storage.api.retriever.OnlineRetrieverV2;
25+
import feast.storage.connectors.bigtable.retriever.BigTableOnlineRetriever;
26+
import feast.storage.connectors.bigtable.retriever.BigTableStoreConfig;
2527
import feast.storage.connectors.redis.retriever.*;
2628
import io.opentracing.Tracer;
29+
import java.io.IOException;
2730
import org.slf4j.Logger;
31+
import org.springframework.beans.factory.annotation.Autowired;
32+
import org.springframework.context.ApplicationContext;
2833
import org.springframework.context.annotation.Bean;
2934
import org.springframework.context.annotation.Configuration;
35+
import org.springframework.context.annotation.Lazy;
3036

3137
@Configuration
3238
public class ServingServiceConfigV2 {
3339
private static final Logger log = org.slf4j.LoggerFactory.getLogger(ServingServiceConfigV2.class);
3440

41+
@Autowired private ApplicationContext context;
42+
43+
@Bean
44+
@Lazy(true)
45+
public BigtableDataClient bigtableClient(FeastProperties feastProperties) throws IOException {
46+
BigTableStoreConfig config = feastProperties.getActiveStore().getBigtableConfig();
47+
String projectId = config.getProjectId();
48+
String instanceId = config.getInstanceId();
49+
50+
return BigtableDataClient.create(
51+
BigtableDataSettings.newBuilder()
52+
.setProjectId(projectId)
53+
.setInstanceId(instanceId)
54+
.build());
55+
}
56+
3557
@Bean
3658
public ServingServiceV2 servingServiceV2(
37-
FeastProperties feastProperties, CachedSpecService specService, Tracer tracer)
38-
throws InvalidProtocolBufferException, JsonProcessingException {
59+
FeastProperties feastProperties, CachedSpecService specService, Tracer tracer) {
3960
ServingServiceV2 servingService = null;
4061
FeastProperties.Store store = feastProperties.getActiveStore();
4162

@@ -51,6 +72,11 @@ public ServingServiceV2 servingServiceV2(
5172
OnlineRetrieverV2 redisRetriever = new OnlineRetriever(redisClient);
5273
servingService = new OnlineServingServiceV2(redisRetriever, specService, tracer);
5374
break;
75+
case BIGTABLE:
76+
BigtableDataClient bigtableClient = context.getBean(BigtableDataClient.class);
77+
OnlineRetrieverV2 bigtableRetriever = new BigTableOnlineRetriever(bigtableClient);
78+
servingService = new OnlineServingServiceV2(bigtableRetriever, specService, tracer);
79+
break;
5480
}
5581

5682
return servingService;

0 commit comments

Comments
 (0)