-
Notifications
You must be signed in to change notification settings - Fork 11
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add SqlJob and Flink Streaming impl (#69)
- Loading branch information
1 parent
701ce75
commit a400db2
Showing
31 changed files
with
1,300 additions
and
122 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
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
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,10 @@ | ||
| apiVersion: hoptimator.linkedin.com/v1alpha1 | ||
| kind: SqlJob | ||
| metadata: | ||
| name: hello-world | ||
| spec: | ||
| dialect: Flink | ||
| executionMode: Streaming | ||
| sql: | ||
| - create table bh (text varchar) with ('connector' = 'blackhole'); | ||
| - insert into bh values ('hello world'); |
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,87 @@ | ||
| apiVersion: apiextensions.k8s.io/v1 | ||
| kind: CustomResourceDefinition | ||
| metadata: | ||
| name: sqljobs.hoptimator.linkedin.com | ||
| spec: | ||
| group: hoptimator.linkedin.com | ||
| names: | ||
| kind: SqlJob | ||
| listKind: SqlJobList | ||
| plural: sqljobs | ||
| singular: sqljob | ||
| shortNames: | ||
| - sql | ||
| - sj | ||
| preserveUnknownFields: false | ||
| scope: Namespaced | ||
| versions: | ||
| - name: v1alpha1 | ||
| served: true | ||
| storage: true | ||
| schema: | ||
| openAPIV3Schema: | ||
| description: Hoptimator generic SQL job | ||
| type: object | ||
| properties: | ||
| apiVersion: | ||
| type: string | ||
| kind: | ||
| type: string | ||
| metadata: | ||
| type: object | ||
| spec: | ||
| description: SQL job spec | ||
| type: object | ||
| properties: | ||
| sql: | ||
| description: SQL script the job should run. | ||
| type: array | ||
| items: | ||
| type: string | ||
| dialect: | ||
| description: Flink, etc. | ||
| type: string | ||
| enum: | ||
| - Flink | ||
| default: Flink | ||
| executionMode: | ||
| description: Streaming or Batch. | ||
| type: string | ||
| enum: | ||
| - Streaming | ||
| - Batch | ||
| default: Streaming | ||
| required: | ||
| - sql | ||
| status: | ||
| description: Filled in by the operator. | ||
| type: object | ||
| properties: | ||
| ready: | ||
| description: Whether the SqlJob is running or completed. | ||
| type: boolean | ||
| failed: | ||
| description: Whether the SqlJob has failed. | ||
| type: boolean | ||
| message: | ||
| description: Error or success message, for information only. | ||
| type: string | ||
| sql: | ||
| description: The SQL being implemented by this SqlJob. | ||
| type: string | ||
| subresources: | ||
| status: {} | ||
| additionalPrinterColumns: | ||
| - name: DIALECT | ||
| type: string | ||
| description: SQL dialect. | ||
| jsonPath: .spec.dialect | ||
| - name: MODE | ||
| type: string | ||
| description: Execution mode. | ||
| jsonPath: .spec.executionMode | ||
| - name: STATUS | ||
| type: string | ||
| description: Job status. | ||
| jsonPath: .status.message | ||
|
|
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
13 changes: 13 additions & 0 deletions
13
...ink-adapter/src/main/java/com/linkedin/hoptimator/catalog/flink/FlinkStreamingSqlJob.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,13 @@ | ||
| package com.linkedin.hoptimator.catalog.flink; | ||
|
|
||
| import com.linkedin.hoptimator.catalog.Resource; | ||
|
|
||
| public class FlinkStreamingSqlJob extends Resource { | ||
|
|
||
| public FlinkStreamingSqlJob(String namespace, String name, String sql) { | ||
| super("FlinkStreamingSqlJob"); | ||
| export("namespace", namespace); | ||
| export("name", name); | ||
| export("sql", sql); | ||
| } | ||
| } |
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
109 changes: 109 additions & 0 deletions
109
.../src/main/java/com/linkedin/hoptimator/operator/flink/FlinkStreamingSqlJobReconciler.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,109 @@ | ||
| package com.linkedin.hoptimator.operator.flink; | ||
|
|
||
| import com.linkedin.hoptimator.catalog.Resource; | ||
| import com.linkedin.hoptimator.catalog.flink.FlinkStreamingSqlJob; | ||
| import com.linkedin.hoptimator.operator.Operator; | ||
| import com.linkedin.hoptimator.models.V1alpha1SqlJob; | ||
| import com.linkedin.hoptimator.models.V1alpha1SqlJobSpec.DialectEnum; | ||
| import com.linkedin.hoptimator.models.V1alpha1SqlJobSpec.ExecutionModeEnum; | ||
| import com.linkedin.hoptimator.models.V1alpha1SqlJobStatus; | ||
|
|
||
| import io.kubernetes.client.extended.controller.reconciler.Reconciler; | ||
| import io.kubernetes.client.extended.controller.reconciler.Request; | ||
| import io.kubernetes.client.extended.controller.reconciler.Result; | ||
|
|
||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.util.List; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| /** | ||
| * Manifests streaming SqlJobs as Flink jobs. | ||
| * | ||
| */ | ||
| public class FlinkStreamingSqlJobReconciler implements Reconciler { | ||
| private final static Logger log = LoggerFactory.getLogger(FlinkStreamingSqlJobReconciler.class); | ||
| private final static String SQLJOB = "hoptimator.linkedin.com/v1alpha1/SqlJob"; | ||
|
|
||
| private final Operator operator; | ||
|
|
||
| public FlinkStreamingSqlJobReconciler(Operator operator) { | ||
| this.operator = operator; | ||
| } | ||
|
|
||
| @Override | ||
| public Result reconcile(Request request) { | ||
| log.info("Reconciling request {}", request); | ||
| String name = request.getName(); | ||
| String namespace = request.getNamespace(); | ||
| Result result = new Result(true, operator.pendingRetryDuration()); | ||
|
|
||
| try { | ||
| V1alpha1SqlJob object = operator.<V1alpha1SqlJob>fetch(SQLJOB, namespace, name); | ||
|
|
||
| if (object == null) { | ||
| log.info("Object {}/{} deleted. Skipping."); | ||
| return new Result(false); | ||
| } | ||
|
|
||
| V1alpha1SqlJobStatus status = object.getStatus(); | ||
| if (status == null) { | ||
| status = new V1alpha1SqlJobStatus(); | ||
| object.setStatus(status); | ||
| } | ||
|
|
||
| List<String> sql = object.getSpec().getSql(); | ||
| String script = sql.stream().collect(Collectors.joining(";\n")); | ||
|
|
||
| DialectEnum dialect = object.getSpec().getDialect(); | ||
| if (!DialectEnum.FLINK.equals(dialect)) { | ||
| log.info("Not Flink SQL. Skipping."); | ||
| return new Result(false); | ||
| } | ||
|
|
||
| ExecutionModeEnum mode = object.getSpec().getExecutionMode(); | ||
| if (!ExecutionModeEnum.STREAMING.equals(mode)) { | ||
| log.info("Not a streaming job. Skipping."); | ||
| return new Result(false); | ||
| } | ||
|
|
||
| Resource.TemplateFactory templateFactory = new Resource.SimpleTemplateFactory(Resource.Environment.EMPTY); | ||
| Resource sqlJob = new FlinkStreamingSqlJob(namespace, name, script); | ||
| boolean allReady = true; | ||
| boolean anyFailed = false; | ||
| for (String yaml : sqlJob.render(templateFactory)) { | ||
| operator.apply(yaml, object); | ||
| if (!operator.isReady(yaml)) { | ||
| allReady = false; | ||
| } | ||
| if (operator.isFailed(yaml)) { | ||
| anyFailed = true; | ||
| allReady = false; | ||
| } | ||
| } | ||
|
|
||
| object.getStatus().setReady(allReady); | ||
| object.getStatus().setFailed(anyFailed); | ||
|
|
||
| if (allReady) { | ||
| object.getStatus().setMessage("Ready."); | ||
| result = new Result(false); // done | ||
| } | ||
| if (anyFailed) { | ||
| object.getStatus().setMessage("Failed."); | ||
| result = new Result(false); // done | ||
| } | ||
|
|
||
| operator.apiFor(SQLJOB).updateStatus(object, x -> object.getStatus()) | ||
| .onFailure((x, y) -> log.error("Failed to update status of SqlJob {}: {}.", name, y.getMessage())) | ||
| .throwsApiException(); | ||
|
|
||
| } catch (Exception e) { | ||
| log.error("Encountered exception while reconciling Flink streaming SqlJob {}/{}", namespace, name, e); | ||
| return new Result(true, operator.failureRetryDuration()); | ||
| } | ||
| return result; | ||
| } | ||
| } | ||
|
|
29 changes: 29 additions & 0 deletions
29
hoptimator-flink-adapter/src/main/resources/FlinkStreamingSqlJob.yaml.template
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,29 @@ | ||
| apiVersion: flink.apache.org/v1beta1 | ||
| kind: FlinkDeployment | ||
| metadata: | ||
| namespace: {{namespace}} | ||
| name: {{name}}-flink-job | ||
| spec: | ||
| image: docker.io/library/hoptimator-flink-runner | ||
| imagePullPolicy: Never | ||
| flinkVersion: v1_16 | ||
| flinkConfiguration: | ||
| taskmanager.numberOfTaskSlots: "1" | ||
| serviceAccount: flink | ||
| jobManager: | ||
| resource: | ||
| memory: "2048m" | ||
| cpu: 0.1 | ||
| taskManager: | ||
| resource: | ||
| memory: "2048m" | ||
| cpu: 0.1 | ||
| job: | ||
| entryClass: com.linkedin.hoptimator.flink.runner.FlinkRunner | ||
| args: | ||
| - {{sql}} | ||
| jarURI: local:///opt/hoptimator-flink-runner.jar | ||
| parallelism: 1 | ||
| upgradeMode: stateless | ||
| state: running | ||
|
|
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
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
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
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
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
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
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
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
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
Oops, something went wrong.