Skip to content

Commit 12e7dea

Browse files
davidheryantodavidheryanto
committed
Add kafka dependency in ingestion (required for import job integration test)
- Remove extra 1 space indentation in FeatureRowToRedisMutationDoFn.java - Update test log level to INFO for root so important logs are not omitted
1 parent 4c92699 commit 12e7dea

3 files changed

Lines changed: 78 additions & 69 deletions

File tree

‎ingestion/pom.xml‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -203,7 +203,6 @@
203203
<groupId>org.apache.kafka</groupId>
204204
<artifactId>kafka-clients</artifactId>
205205
<version>2.3.0</version>
206-
<scope>test</scope>
207206
</dependency>
208207

209208
<dependency>
@@ -257,14 +256,23 @@
257256
<scope>runtime</scope>
258257
</dependency>
259258

260-
<!-- To simulate actual Redis for ingestion integration test -->
259+
<!-- To run actual Redis for ingestion integration test -->
261260
<dependency>
262261
<groupId>com.github.kstyrc</groupId>
263262
<artifactId>embedded-redis</artifactId>
264263
<version>0.6</version>
265264
<scope>test</scope>
266265
</dependency>
267266

267+
<!-- To run actual Kafka for ingestion integration test -->
268+
<dependency>
269+
<groupId>org.apache.kafka</groupId>
270+
<artifactId>kafka_2.12</artifactId>
271+
<version>2.3.0</version>
272+
<scope>test</scope>
273+
</dependency>
274+
275+
268276
<dependency>
269277
<groupId>com.google.guava</groupId>
270278
<artifactId>guava</artifactId>

‎ingestion/src/main/java/feast/store/serving/redis/FeatureRowToRedisMutationDoFn.java‎

Lines changed: 66 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -1,74 +1,74 @@
1-
/*
2-
* Copyright 2018 The Feast Authors
3-
*
4-
* Licensed under the Apache License, Version 2.0 (the "License");
5-
* you may not use this file except in compliance with the License.
6-
* You may obtain a copy of the License at
7-
*
8-
* https://www.apache.org/licenses/LICENSE-2.0
9-
*
10-
* Unless required by applicable law or agreed to in writing, software
11-
* distributed under the License is distributed on an "AS IS" BASIS,
12-
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13-
* See the License for the specific language governing permissions and
14-
* limitations under the License.
15-
*
16-
*/
1+
/*
2+
* Copyright 2018 The Feast Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*
16+
*/
1717

18-
package feast.store.serving.redis;
18+
package feast.store.serving.redis;
1919

20-
import feast.core.FeatureSetProto.EntitySpec;
21-
import feast.core.FeatureSetProto.FeatureSetSpec;
22-
import feast.storage.RedisProto.RedisKey;
23-
import feast.storage.RedisProto.RedisKey.Builder;
24-
import feast.store.serving.redis.RedisCustomIO.Method;
25-
import feast.store.serving.redis.RedisCustomIO.RedisMutation;
26-
import feast.types.FieldProto.Field;
27-
import feast.types.FeatureRowExtendedProto.FeatureRowExtended;
28-
import feast.types.FeatureRowProto.FeatureRow;
29-
import java.util.Set;
30-
import java.util.stream.Collectors;
31-
import lombok.AllArgsConstructor;
32-
import lombok.extern.slf4j.Slf4j;
33-
import org.apache.beam.sdk.transforms.DoFn;
20+
import feast.core.FeatureSetProto.EntitySpec;
21+
import feast.core.FeatureSetProto.FeatureSetSpec;
22+
import feast.storage.RedisProto.RedisKey;
23+
import feast.storage.RedisProto.RedisKey.Builder;
24+
import feast.store.serving.redis.RedisCustomIO.Method;
25+
import feast.store.serving.redis.RedisCustomIO.RedisMutation;
26+
import feast.types.FeatureRowProto.FeatureRow;
27+
import feast.types.FieldProto.Field;
28+
import java.util.Set;
29+
import java.util.stream.Collectors;
30+
import lombok.AllArgsConstructor;
31+
import lombok.NoArgsConstructor;
32+
import lombok.extern.slf4j.Slf4j;
33+
import org.apache.beam.sdk.transforms.DoFn;
3434

35-
@Slf4j
36-
@AllArgsConstructor
37-
public class FeatureRowToRedisMutationDoFn extends DoFn<FeatureRow, RedisMutation> {
35+
@Slf4j
36+
@AllArgsConstructor
37+
public class FeatureRowToRedisMutationDoFn extends DoFn<FeatureRow, RedisMutation> {
3838

39-
private FeatureSetSpec featureSetSpec;
39+
private FeatureSetSpec featureSetSpec;
4040

41-
// TODO: type and completeness checking
42-
private RedisKey getKey(FeatureRow featureRow) {
43-
Set<String> entityNames = featureSetSpec.getEntitiesList().stream()
44-
.map(EntitySpec::getName).collect(Collectors.toSet());
41+
// TODO: type and completeness checking
42+
private RedisKey getKey(FeatureRow featureRow) {
43+
Set<String> entityNames = featureSetSpec.getEntitiesList().stream()
44+
.map(EntitySpec::getName).collect(Collectors.toSet());
4545

46-
Builder redisKeyBuilder = RedisKey.newBuilder()
47-
.setFeatureSet(featureRow.getFeatureSet());
48-
for (Field field : featureRow.getFieldsList()) {
49-
if (entityNames.contains(field.getName())) {
50-
redisKeyBuilder.addEntities(field);
51-
}
52-
}
46+
Builder redisKeyBuilder = RedisKey.newBuilder()
47+
.setFeatureSet(featureRow.getFeatureSet());
48+
for (Field field : featureRow.getFieldsList()) {
49+
if (entityNames.contains(field.getName())) {
50+
redisKeyBuilder.addEntities(field);
51+
}
52+
}
5353

54-
return redisKeyBuilder.build();
55-
}
54+
return redisKeyBuilder.build();
55+
}
5656

57-
/**
58-
* Output a redis mutation object for every feature in the feature row.
59-
*/
60-
@ProcessElement
61-
public void processElement(ProcessContext context) {
62-
FeatureRow featureRow = context.element();
63-
RedisKey key = getKey(featureRow);
64-
// Duration expiry = options.getExpiryDuration();
65-
// // Add randomness to expiry so that it won't expire in the same time.
66-
// long expiryMillis = (long) (expiry.getMillis() * (1 + random.nextFloat()));
67-
context.output(
68-
RedisMutation.builder()
69-
.key(key.toByteArray())
70-
.value(featureRow.toByteArray())
71-
.method(Method.SET)
72-
.build());
73-
}
74-
}
57+
/**
58+
* Output a redis mutation object for every feature in the feature row.
59+
*/
60+
@ProcessElement
61+
public void processElement(ProcessContext context) {
62+
FeatureRow featureRow = context.element();
63+
RedisKey key = getKey(featureRow);
64+
// Duration expiry = options.getExpiryDuration();
65+
// Add randomness to expiry so that it won't expire in the same time.
66+
// long expiryMillis = (long) (expiry.getMillis() * (1 + random.nextFloat()));
67+
context.output(
68+
RedisMutation.builder()
69+
.key(key.toByteArray())
70+
.value(featureRow.toByteArray())
71+
.method(Method.SET)
72+
.build());
73+
}
74+
}

‎ingestion/src/test/resources/logback-test.xml‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,8 @@
2828
<pattern>%d{HH:mm:ss} [%thread] %-5level %logger{50} - %msg%n</pattern>
2929
</encoder>
3030
</appender>
31-
<root level="WARN">
31+
32+
<root level="INFO">
3233
<appender-ref ref="STDOUT"/>
3334
</root>
3435
</configuration>

0 commit comments

Comments
 (0)