Skip to content

Commit 26c8070

Browse files
author
Oleksii Moskalenko
authored
Allow ingestion job grouping/consolidation to be configurable (#825)
* job grouping strategy * apidocs & tests * fix some old docs * adding apidoc in Spring Config
1 parent 89883d4 commit 26c8070

11 files changed

Lines changed: 388 additions & 107 deletions

File tree

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,9 @@ public static class JobProperties {
8181
/* The active Apache Beam runner name. This name references one instance of the Runner class */
8282
private String activeRunner;
8383

84+
/* If true only one IngestionJob would be created per source with all subscribed stores in it */
85+
private Boolean consolidateJobsPerSource = false;
86+
8487
/** List of configured job runners. */
8588
private List<Runner> runners = new ArrayList<>();
8689

‎core/src/main/java/feast/core/config/JobConfig.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,11 @@
2020
import com.google.protobuf.InvalidProtocolBufferException;
2121
import com.google.protobuf.util.JsonFormat;
2222
import feast.core.config.FeastProperties.JobProperties;
23+
import feast.core.dao.JobRepository;
24+
import feast.core.job.ConsolidatedJobStrategy;
25+
import feast.core.job.JobGroupingStrategy;
2326
import feast.core.job.JobManager;
27+
import feast.core.job.JobPerStoreStrategy;
2428
import feast.core.job.dataflow.DataflowJobManager;
2529
import feast.core.job.direct.DirectJobRegistry;
2630
import feast.core.job.direct.DirectRunnerJobManager;
@@ -64,6 +68,26 @@ public IngestionJobProto.SpecsStreamingUpdateConfig createSpecsStreamingUpdateCo
6468
.build();
6569
}
6670

71+
/**
72+
* Returns Grouping Strategy which is responsible for how Ingestion would be split across job
73+
* instances (or how Sources and Stores would be grouped together). Choosing strategy depends on
74+
* FeastProperties config "feast.jobs.consolidate-jobs-per-source".
75+
*
76+
* @param feastProperties feast config properties
77+
* @param jobRepository repository required by strategy
78+
* @return JobGroupingStrategy
79+
*/
80+
@Bean
81+
public JobGroupingStrategy getJobGroupingStrategy(
82+
FeastProperties feastProperties, JobRepository jobRepository) {
83+
Boolean shouldConsolidateJobs = feastProperties.getJobs().getConsolidateJobsPerSource();
84+
if (shouldConsolidateJobs) {
85+
return new ConsolidatedJobStrategy(jobRepository);
86+
} else {
87+
return new JobPerStoreStrategy(jobRepository);
88+
}
89+
}
90+
6791
/**
6892
* Get a JobManager according to the runner type and Dataflow configuration.
6993
*

‎core/src/main/java/feast/core/job/ConsolidatedJobStrategy.java‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
* Copyright 2018-2020 The Feast Authors
4+
*
5+
* Licensed under the Apache License, Version 2.0 (the "License");
6+
* you may not use this file except in compliance with the License.
7+
* You may obtain a copy of the License at
8+
*
9+
* https://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package feast.core.job;
18+
19+
import feast.core.dao.JobRepository;
20+
import feast.core.model.Job;
21+
import feast.core.model.JobStatus;
22+
import feast.core.model.Source;
23+
import feast.core.model.Store;
24+
import java.time.Instant;
25+
import java.util.HashSet;
26+
import java.util.Map;
27+
import java.util.Objects;
28+
import java.util.Set;
29+
import java.util.stream.Collectors;
30+
import java.util.stream.Stream;
31+
import org.apache.commons.lang3.tuple.Pair;
32+
33+
/**
34+
* In this strategy one Ingestion Job per source is created. All stores that subscribed to
35+
* FeatureSets from this source will be included as sinks in this consolidated Job.
36+
*
37+
* <p>JobId will contain only source parameters (type + config). StoreName will remain empty in Job
38+
* table.
39+
*/
40+
public class ConsolidatedJobStrategy implements JobGroupingStrategy {
41+
private final JobRepository jobRepository;
42+
43+
public ConsolidatedJobStrategy(JobRepository jobRepository) {
44+
this.jobRepository = jobRepository;
45+
}
46+
47+
@Override
48+
public Job getOrCreateJob(Source source, Set<Store> stores) {
49+
return jobRepository
50+
.findFirstBySourceTypeAndSourceConfigAndStoreNameAndStatusNotInOrderByLastUpdatedDesc(
51+
source.getType(), source.getConfig(), null, JobStatus.getTerminalStates())
52+
.orElseGet(
53+
() ->
54+
Job.builder()
55+
.setSource(source)
56+
.setStores(stores)
57+
.setFeatureSetJobStatuses(new HashSet<>())
58+
.build());
59+
}
60+
61+
@Override
62+
public String createJobId(Job job) {
63+
String dateSuffix = String.valueOf(Instant.now().toEpochMilli());
64+
String jobId =
65+
String.format(
66+
"%s-%d-%s",
67+
job.getSource().getTypeString(),
68+
Objects.hashCode(job.getSource().getConfig()),
69+
dateSuffix);
70+
return jobId.replaceAll("_store", "-").toLowerCase();
71+
}
72+
73+
@Override
74+
public Iterable<Pair<Source, Set<Store>>> collectSingleJobInput(
75+
Stream<Pair<Source, Store>> stream) {
76+
Map<Source, Set<Store>> map =
77+
stream.collect(
78+
Collectors.groupingBy(
79+
Pair::getLeft, Collectors.mapping(Pair::getRight, Collectors.toSet())));
80+
81+
return map.entrySet().stream()
82+
.map(e -> Pair.of(e.getKey(), e.getValue()))
83+
.collect(Collectors.toList());
84+
}
85+
}

‎core/src/main/java/feast/core/job/CreateJobTask.java‎

Lines changed: 1 addition & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -19,19 +19,13 @@
1919
import feast.core.log.Action;
2020
import feast.core.model.Job;
2121
import feast.core.model.JobStatus;
22-
import feast.core.model.Source;
23-
import java.time.Instant;
24-
import java.util.Objects;
2522
import lombok.Builder;
2623
import lombok.Getter;
2724
import lombok.Setter;
2825
import org.slf4j.Logger;
2926
import org.slf4j.LoggerFactory;
3027

31-
/**
32-
* Task that starts recently created {@link Job} by using {@link JobManager}. Since it's new job its
33-
* Id being generated from attached {@link Source} and updated accordingly in-place.
34-
*/
28+
/** Task that starts recently created {@link Job} by using {@link JobManager}. */
3529
@Getter
3630
@Setter
3731
@Builder(setterPrefix = "set")
@@ -43,12 +37,10 @@ public class CreateJobTask implements JobTask {
4337

4438
@Override
4539
public Job call() {
46-
String jobId = createJobId(job.getSource());
4740
String runnerName = jobManager.getRunnerType().toString();
4841

4942
job.setRunner(jobManager.getRunnerType());
5043
job.setStatus(JobStatus.PENDING);
51-
job.setId(jobId);
5244

5345
try {
5446
JobTask.logAudit(Action.SUBMIT, job, "Building graph and submitting to %s", runnerName);
@@ -73,12 +65,4 @@ public Job call() {
7365
return job;
7466
}
7567
}
76-
77-
String createJobId(Source source) {
78-
String dateSuffix = String.valueOf(Instant.now().toEpochMilli());
79-
String jobId =
80-
String.format(
81-
"%s-%d-%s", source.getTypeString(), Objects.hashCode(source.getConfig()), dateSuffix);
82-
return jobId.replaceAll("_store", "-").toLowerCase();
83-
}
8468
}

‎core/src/main/java/feast/core/job/JobGroupingStrategy.java‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
* Copyright 2018-2020 The Feast Authors
4+
*
5+
* Licensed under the Apache License, Version 2.0 (the "License");
6+
* you may not use this file except in compliance with the License.
7+
* You may obtain a copy of the License at
8+
*
9+
* https://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package feast.core.job;
18+
19+
import feast.core.model.Job;
20+
import feast.core.model.Source;
21+
import feast.core.model.Store;
22+
import java.util.Set;
23+
import java.util.stream.Stream;
24+
import org.apache.commons.lang3.tuple.Pair;
25+
26+
/**
27+
* Strategy interface that defines how responsibility for sources and stores will be distributed
28+
* across Ingestion Jobs.
29+
*/
30+
public interface JobGroupingStrategy {
31+
/** Get the non terminated ingestion job ingesting for given source and stores. */
32+
public Job getOrCreateJob(Source source, Set<Store> stores);
33+
/** Create unique JobId that would be used as key in communications with JobRunner */
34+
public String createJobId(Job job);
35+
/* Distribute given sources and stores across jobs. One yielded Pair - one created Job **/
36+
public Iterable<Pair<Source, Set<Store>>> collectSingleJobInput(
37+
Stream<Pair<Source, Store>> stream);
38+
}

‎core/src/main/java/feast/core/job/JobPerStoreStrategy.java‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
* Copyright 2018-2020 The Feast Authors
4+
*
5+
* Licensed under the Apache License, Version 2.0 (the "License");
6+
* you may not use this file except in compliance with the License.
7+
* You may obtain a copy of the License at
8+
*
9+
* https://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package feast.core.job;
18+
19+
import com.google.common.collect.Lists;
20+
import feast.core.dao.JobRepository;
21+
import feast.core.model.Job;
22+
import feast.core.model.JobStatus;
23+
import feast.core.model.Source;
24+
import feast.core.model.Store;
25+
import java.time.Instant;
26+
import java.util.ArrayList;
27+
import java.util.HashSet;
28+
import java.util.Objects;
29+
import java.util.Set;
30+
import java.util.stream.Collectors;
31+
import java.util.stream.Stream;
32+
import org.apache.commons.lang3.tuple.Pair;
33+
34+
/**
35+
* In this strategy one job per Source-Store pair is created.
36+
*
37+
* <p>JobId is generated accordingly from Source (type+config) and StoreName.
38+
*/
39+
public class JobPerStoreStrategy implements JobGroupingStrategy {
40+
private final JobRepository jobRepository;
41+
42+
public JobPerStoreStrategy(JobRepository jobRepository) {
43+
this.jobRepository = jobRepository;
44+
}
45+
46+
@Override
47+
public Job getOrCreateJob(Source source, Set<Store> stores) {
48+
ArrayList<Store> storesList = Lists.newArrayList(stores);
49+
if (storesList.size() != 1) {
50+
throw new RuntimeException("Only one store is acceptable in JobPerStore Strategy");
51+
}
52+
Store store = storesList.get(0);
53+
54+
return jobRepository
55+
.findFirstBySourceTypeAndSourceConfigAndStoreNameAndStatusNotInOrderByLastUpdatedDesc(
56+
source.getType(), source.getConfig(), store.getName(), JobStatus.getTerminalStates())
57+
.orElseGet(
58+
() ->
59+
Job.builder()
60+
.setSource(source)
61+
.setStoreName(store.getName())
62+
.setStores(stores)
63+
.setFeatureSetJobStatuses(new HashSet<>())
64+
.build());
65+
}
66+
67+
@Override
68+
public String createJobId(Job job) {
69+
String dateSuffix = String.valueOf(Instant.now().toEpochMilli());
70+
String jobId =
71+
String.format(
72+
"%s-%d-to-%s-%s",
73+
job.getSource().getTypeString(),
74+
Objects.hashCode(job.getSource().getConfig()),
75+
job.getStoreName(),
76+
dateSuffix);
77+
return jobId.replaceAll("_store", "-").toLowerCase();
78+
}
79+
80+
@Override
81+
public Iterable<Pair<Source, Set<Store>>> collectSingleJobInput(
82+
Stream<Pair<Source, Store>> stream) {
83+
return stream.map(p -> Pair.of(p.getLeft(), Set.of(p.getRight()))).collect(Collectors.toList());
84+
}
85+
}

0 commit comments

Comments
 (0)