-
Notifications
You must be signed in to change notification settings - Fork 3.7k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
add container orchestrator for normalization (#9020)
* initial commit * improve labeling * make more generic * refactor constants * move port constants * remove flag * some final config fixes * clean up * oops, didn't check this in * add dbt orchestrator (#9114) * respond to pr feedback
- Loading branch information
Showing
12 changed files
with
757 additions
and
221 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
69 changes: 69 additions & 0 deletions
69
...iner-orchestrator/src/main/java/io/airbyte/container_orchestrator/DbtJobOrchestrator.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| /* | ||
| * Copyright (c) 2021 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.container_orchestrator; | ||
|
|
||
| import io.airbyte.config.Configs; | ||
| import io.airbyte.config.OperatorDbtInput; | ||
| import io.airbyte.scheduler.models.IntegrationLauncherConfig; | ||
| import io.airbyte.scheduler.models.JobRunConfig; | ||
| import io.airbyte.workers.DbtTransformationRunner; | ||
| import io.airbyte.workers.DbtTransformationWorker; | ||
| import io.airbyte.workers.WorkerConfigs; | ||
| import io.airbyte.workers.WorkerUtils; | ||
| import io.airbyte.workers.normalization.NormalizationRunnerFactory; | ||
| import io.airbyte.workers.process.ProcessFactory; | ||
| import io.airbyte.workers.temporal.sync.ReplicationLauncherWorker; | ||
| import java.nio.file.Path; | ||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| @Slf4j | ||
| public class DbtJobOrchestrator implements JobOrchestrator<OperatorDbtInput> { | ||
|
|
||
| private final Configs configs; | ||
| private final WorkerConfigs workerConfigs; | ||
| private final ProcessFactory processFactory; | ||
|
|
||
| public DbtJobOrchestrator(final Configs configs, final WorkerConfigs workerConfigs, final ProcessFactory processFactory) { | ||
| this.configs = configs; | ||
| this.workerConfigs = workerConfigs; | ||
| this.processFactory = processFactory; | ||
| } | ||
|
|
||
| @Override | ||
| public String getOrchestratorName() { | ||
| return "DBT Transformation"; | ||
| } | ||
|
|
||
| @Override | ||
| public Class<OperatorDbtInput> getInputClass() { | ||
| return OperatorDbtInput.class; | ||
| } | ||
|
|
||
| @Override | ||
| public void runJob() throws Exception { | ||
| final JobRunConfig jobRunConfig = readJobRunConfig(); | ||
| final OperatorDbtInput dbtInput = readInput(); | ||
|
|
||
| final IntegrationLauncherConfig destinationLauncherConfig = JobOrchestrator.readAndDeserializeFile( | ||
| ReplicationLauncherWorker.INIT_FILE_DESTINATION_LAUNCHER_CONFIG, IntegrationLauncherConfig.class); | ||
|
|
||
| log.info("Setting up dbt worker..."); | ||
| final DbtTransformationWorker worker = new DbtTransformationWorker( | ||
| jobRunConfig.getJobId(), | ||
| Math.toIntExact(jobRunConfig.getAttemptId()), | ||
| workerConfigs.getResourceRequirements(), | ||
| new DbtTransformationRunner( | ||
| workerConfigs, | ||
| processFactory, NormalizationRunnerFactory.create( | ||
| workerConfigs, | ||
| destinationLauncherConfig.getDockerImage(), | ||
| processFactory)));; | ||
|
|
||
| log.info("Running dbt worker..."); | ||
| final Path jobRoot = WorkerUtils.getJobRoot(configs.getWorkspaceRoot(), jobRunConfig.getJobId(), jobRunConfig.getAttemptId()); | ||
| worker.run(dbtInput, jobRoot); | ||
| } | ||
|
|
||
| } |
45 changes: 45 additions & 0 deletions
45
...ntainer-orchestrator/src/main/java/io/airbyte/container_orchestrator/JobOrchestrator.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| /* | ||
| * Copyright (c) 2021 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.container_orchestrator; | ||
|
|
||
| import io.airbyte.commons.json.Jsons; | ||
| import io.airbyte.scheduler.models.JobRunConfig; | ||
| import io.airbyte.workers.temporal.sync.OrchestratorConstants; | ||
| import java.io.IOException; | ||
| import java.nio.file.Files; | ||
| import java.nio.file.Path; | ||
|
|
||
| /** | ||
| * The job orchestrator helps abstract over container launcher application differences across | ||
| * replication, normalization, and custom dbt operators. | ||
| * | ||
| * @param <INPUT> job input type | ||
| */ | ||
| public interface JobOrchestrator<INPUT> { | ||
|
|
||
| // used for logging | ||
| String getOrchestratorName(); | ||
|
|
||
| // used to serialize the loaded input | ||
| Class<INPUT> getInputClass(); | ||
|
|
||
| // reads input from a file that was copied to the container launcher | ||
| default INPUT readInput() throws IOException { | ||
| return readAndDeserializeFile(OrchestratorConstants.INIT_FILE_INPUT, getInputClass()); | ||
| } | ||
|
|
||
| // reads the job run config from a file that was copied to the container launcher | ||
| default JobRunConfig readJobRunConfig() throws IOException { | ||
| return readAndDeserializeFile(OrchestratorConstants.INIT_FILE_JOB_RUN_CONFIG, JobRunConfig.class); | ||
| } | ||
|
|
||
| // the unique logic that belongs to each type of job belongs here | ||
| void runJob() throws Exception; | ||
|
|
||
| static <T> T readAndDeserializeFile(String path, Class<T> type) throws IOException { | ||
| return Jsons.deserialize(Files.readString(Path.of(path)), type); | ||
| } | ||
|
|
||
| } |
68 changes: 68 additions & 0 deletions
68
...strator/src/main/java/io/airbyte/container_orchestrator/NormalizationJobOrchestrator.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| /* | ||
| * Copyright (c) 2021 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.container_orchestrator; | ||
|
|
||
| import io.airbyte.config.Configs; | ||
| import io.airbyte.config.NormalizationInput; | ||
| import io.airbyte.scheduler.models.IntegrationLauncherConfig; | ||
| import io.airbyte.scheduler.models.JobRunConfig; | ||
| import io.airbyte.workers.DefaultNormalizationWorker; | ||
| import io.airbyte.workers.NormalizationWorker; | ||
| import io.airbyte.workers.WorkerConfigs; | ||
| import io.airbyte.workers.WorkerUtils; | ||
| import io.airbyte.workers.normalization.NormalizationRunnerFactory; | ||
| import io.airbyte.workers.process.ProcessFactory; | ||
| import io.airbyte.workers.temporal.sync.ReplicationLauncherWorker; | ||
| import java.nio.file.Path; | ||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| @Slf4j | ||
| public class NormalizationJobOrchestrator implements JobOrchestrator<NormalizationInput> { | ||
|
|
||
| private final Configs configs; | ||
| private final WorkerConfigs workerConfigs; | ||
| private final ProcessFactory processFactory; | ||
|
|
||
| public NormalizationJobOrchestrator(final Configs configs, final WorkerConfigs workerConfigs, final ProcessFactory processFactory) { | ||
| this.configs = configs; | ||
| this.workerConfigs = workerConfigs; | ||
| this.processFactory = processFactory; | ||
| } | ||
|
|
||
| @Override | ||
| public String getOrchestratorName() { | ||
| return "Normalization"; | ||
| } | ||
|
|
||
| @Override | ||
| public Class<NormalizationInput> getInputClass() { | ||
| return NormalizationInput.class; | ||
| } | ||
|
|
||
| @Override | ||
| public void runJob() throws Exception { | ||
| final JobRunConfig jobRunConfig = readJobRunConfig(); | ||
| final NormalizationInput normalizationInput = readInput(); | ||
|
|
||
| final IntegrationLauncherConfig destinationLauncherConfig = JobOrchestrator.readAndDeserializeFile( | ||
| ReplicationLauncherWorker.INIT_FILE_DESTINATION_LAUNCHER_CONFIG, IntegrationLauncherConfig.class); | ||
|
|
||
| log.info("Setting up normalization worker..."); | ||
| final NormalizationWorker normalizationWorker = new DefaultNormalizationWorker( | ||
| jobRunConfig.getJobId(), | ||
| Math.toIntExact(jobRunConfig.getAttemptId()), | ||
| NormalizationRunnerFactory.create( | ||
| workerConfigs, | ||
| destinationLauncherConfig.getDockerImage(), | ||
| processFactory), | ||
| configs.getWorkerEnvironment()); | ||
|
|
||
| log.info("Running normalization worker..."); | ||
| final Path jobRoot = WorkerUtils.getJobRoot(configs.getWorkspaceRoot(), jobRunConfig.getJobId(), jobRunConfig.getAttemptId()); | ||
| normalizationWorker.run(normalizationInput, jobRoot); | ||
|
|
||
| } | ||
|
|
||
| } |
Oops, something went wrong.