Skip to content

Commit 11c53f8

Browse files
ryannedolanryannedolan
authored
Add JobTemplates and Flink demo (#74)
* Add JobTemplates and Flink demo * PR fixes h/t @jogrogan
1 parent 159a0e9 commit 11c53f8

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

47 files changed

+1023
-105
lines changed

‎deploy/samples/flink.yaml

+36
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
## This template adds Flink support.
2+
3+
apiVersion: hoptimator.linkedin.com/v1alpha1
4+
kind: JobTemplate
5+
metadata:
6+
name: flink-template
7+
spec:
8+
yaml: |
9+
apiVersion: flink.apache.org/v1beta1
10+
kind: FlinkDeployment
11+
metadata:
12+
name: {{name}}
13+
spec:
14+
image: docker.io/library/hoptimator-flink-runner
15+
imagePullPolicy: Never
16+
flinkVersion: v1_16
17+
flinkConfiguration:
18+
taskmanager.numberOfTaskSlots: "1"
19+
serviceAccount: flink
20+
jobManager:
21+
resource:
22+
memory: "2048m"
23+
cpu: 0.1
24+
taskManager:
25+
resource:
26+
memory: "2048m"
27+
cpu: 0.1
28+
job:
29+
entryClass: com.linkedin.hoptimator.flink.runner.FlinkRunner
30+
args:
31+
- {{sql}}
32+
jarURI: local:///opt/hoptimator-flink-runner.jar
33+
parallelism: 1
34+
upgradeMode: stateless
35+
state: running
36+

‎generate-models.sh

+1
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ docker run \
1111
ghcr.io/kubernetes-client/java/crd-model-gen:v1.0.6 \
1212
/generate.sh -o "$(pwd)/hoptimator-k8s" -n "" -p "com.linkedin.hoptimator.k8s" \
1313
-u "$(pwd)/hoptimator-k8s/src/main/resources/databases.crd.yaml" \
14+
-u "$(pwd)/hoptimator-k8s/src/main/resources/jobtemplates.crd.yaml" \
1415
-u "$(pwd)/hoptimator-k8s/src/main/resources/pipelines.crd.yaml" \
1516
-u "$(pwd)/hoptimator-k8s/src/main/resources/tabletemplates.crd.yaml" \
1617
-u "$(pwd)/hoptimator-k8s/src/main/resources/views.crd.yaml" \

‎hoptimator-cli/src/main/java/sqlline/HoptimatorAppConfig.java

+2-3
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77

88
import org.apache.calcite.rel.RelNode;
99
import org.apache.calcite.jdbc.CalciteConnection;
10+
import org.apache.calcite.sql.dialect.AnsiSqlDialect;
1011

1112
import org.jline.reader.Completer;
1213

@@ -90,9 +91,7 @@ public void execute(String line, DispatchCallback dispatchCallback) {
9091
try {
9192
RelNode rel = HoptimatorDriver.convert(conn.createPrepareContext(), sql).root.rel;
9293
PipelineRel.Implementor plan = DeploymentService.plan(rel);
93-
Sink sink = new Sink("PIPELINE", Arrays.asList(new String[]{"PIPELINE", "SINK"}), plan.rowType(),
94-
Collections.emptyMap());
95-
sqlline.output(plan.insertInto(sink).sql());
94+
sqlline.output(plan.sql().apply(AnsiSqlDialect.DEFAULT));
9695
} catch (SQLException e) {
9796
sqlline.error(e);
9897
dispatchCallback.setToFailure();

‎hoptimator-jdbc/build.gradle

+2-2
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,6 @@ test {
3232
}
3333

3434
java {
35-
sourceCompatibility = JavaVersion.VERSION_11
36-
targetCompatibility = JavaVersion.VERSION_11
35+
sourceCompatibility = JavaVersion.VERSION_11
36+
targetCompatibility = JavaVersion.VERSION_11
3737
}

‎hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorDdlExecutor.java

+17-2
Original file line numberDiff line numberDiff line change
@@ -22,9 +22,14 @@
2222
import com.linkedin.hoptimator.Database;
2323
import com.linkedin.hoptimator.util.DeploymentService;
2424
import com.linkedin.hoptimator.util.MaterializedView;
25+
import com.linkedin.hoptimator.util.Sink;
26+
import com.linkedin.hoptimator.util.planner.Pipeline;
27+
import com.linkedin.hoptimator.util.planner.PipelineRel;
28+
import com.linkedin.hoptimator.util.planner.ScriptImplementor;
2529

2630
import org.apache.calcite.jdbc.CalcitePrepare;
2731
import org.apache.calcite.jdbc.CalciteSchema;
32+
import org.apache.calcite.rel.RelNode;
2833
import org.apache.calcite.rel.type.RelDataType;
2934
import org.apache.calcite.rel.type.RelDataTypeSystem;
3035
import org.apache.calcite.server.DdlExecutor;
@@ -35,6 +40,7 @@
3540
import org.apache.calcite.schema.impl.ViewTable;
3641
import org.apache.calcite.schema.impl.ViewTableMacro;
3742
import org.apache.calcite.sql.SqlCall;
43+
import org.apache.calcite.sql.SqlDialect;
3844
import org.apache.calcite.sql.SqlIdentifier;
3945
import org.apache.calcite.sql.SqlNode;
4046
import org.apache.calcite.sql.SqlNodeList;
@@ -166,9 +172,18 @@ public void execute(SqlCreateMaterializedView create,
166172
MaterializedViewTable materializedViewTable = new MaterializedViewTable(viewTableMacro);
167173
RelDataType rowType = materializedViewTable.getRowType(new SqlTypeFactoryImpl(
168174
RelDataTypeSystem.DEFAULT));
169-
MaterializedView hook = new MaterializedView(context, database, viewPath, rowType, sql,
170-
Collections.emptyMap()); // TODO support CREATE ... WITH (options...)
175+
176+
// Plan a pipeline to materialize the view.
177+
RelNode rel = HoptimatorDriver.convert(context, sql).root.rel;
178+
PipelineRel.Implementor plan = DeploymentService.plan(rel);
179+
plan.setSink(database, viewPath, rowType, Collections.emptyMap());
180+
Pipeline pipeline = plan.pipeline();
181+
182+
MaterializedView hook = new MaterializedView(database, viewPath, sql, plan.sql(),
183+
plan.pipeline());
184+
// TODO support CREATE ... WITH (options...)
171185
ValidationService.validateOrThrow(hook, MaterializedView.class);
186+
pipeline.update();
172187
if (create.getReplace()) {
173188
DeploymentService.update(hook, MaterializedView.class);
174189
} else {

‎hoptimator-jdbc/src/testFixtures/java/com/linkedin/hoptimator/jdbc/QuidemTestBase.java

+3-2
Original file line numberDiff line numberDiff line change
@@ -83,8 +83,9 @@ public void execute(Context context, boolean execute) throws Exception {
8383
CalciteConnection conn = (CalciteConnection) context.connection();
8484
RelNode rel = HoptimatorDriver.convert(conn.createPrepareContext(), sql).root.rel;
8585
String specs = DeploymentService.plan(rel).pipeline().specify().stream()
86-
.collect(Collectors.joining("\n---\n"));
87-
context.echo(Collections.singletonList(specs));
86+
.collect(Collectors.joining("---\n"));
87+
String[] lines = specs.replaceAll(";\n","\n").split("\n");
88+
context.echo(Arrays.asList(lines));
8889
} else {
8990
context.echo(content);
9091
}

‎hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sApiEndpoints.java

+5
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@
22

33
import com.linkedin.hoptimator.k8s.models.V1alpha1Database;
44
import com.linkedin.hoptimator.k8s.models.V1alpha1DatabaseList;
5+
import com.linkedin.hoptimator.k8s.models.V1alpha1JobTemplate;
6+
import com.linkedin.hoptimator.k8s.models.V1alpha1JobTemplateList;
57
import com.linkedin.hoptimator.k8s.models.V1alpha1Pipeline;
68
import com.linkedin.hoptimator.k8s.models.V1alpha1PipelineList;
79
import com.linkedin.hoptimator.k8s.models.V1alpha1View;
@@ -35,6 +37,9 @@ public final class K8sApiEndpoints {
3537
public static final K8sApiEndpoint<V1alpha1TableTemplate, V1alpha1TableTemplateList> TABLE_TEMPLATES =
3638
new K8sApiEndpoint<>("TableTemplate", "hoptimator.linkedin.com", "v1alpha1", "tabletemplates", false,
3739
V1alpha1TableTemplate.class, V1alpha1TableTemplateList.class);
40+
public static final K8sApiEndpoint<V1alpha1JobTemplate, V1alpha1JobTemplateList> JOB_TEMPLATES =
41+
new K8sApiEndpoint<>("JobTemplate", "hoptimator.linkedin.com", "v1alpha1", "jobtemplates", false,
42+
V1alpha1JobTemplate.class, V1alpha1JobTemplateList.class);
3843

3944
private K8sApiEndpoints() {
4045
}

‎hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sDeployerProvider.java

+5
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
package com.linkedin.hoptimator.k8s;
22

3+
import com.linkedin.hoptimator.util.Sink;
34
import com.linkedin.hoptimator.util.Source;
5+
import com.linkedin.hoptimator.util.Job;
46

57
import com.linkedin.hoptimator.Deployer;
68
import com.linkedin.hoptimator.DeployerProvider;
@@ -28,6 +30,9 @@ public <T> Collection<Deployer<T>> deployers(Class<T> clazz) {
2830
if (Source.class.isAssignableFrom(clazz)) {
2931
list.add((Deployer<T>) new K8sSourceDeployer(K8sContext.currentContext()));
3032
}
33+
if (Job.class.isAssignableFrom(clazz)) {
34+
list.add((Deployer<T>) new K8sJobDeployer(K8sContext.currentContext()));
35+
}
3136
return list;
3237
}
3338
}

‎hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sJobDeployer.java

+51
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
package com.linkedin.hoptimator.k8s;
2+
3+
import com.linkedin.hoptimator.util.Job;
4+
import com.linkedin.hoptimator.util.Template;
5+
import com.linkedin.hoptimator.k8s.models.V1alpha1JobTemplate;
6+
import com.linkedin.hoptimator.k8s.models.V1alpha1JobTemplateList;
7+
8+
import org.apache.calcite.sql.SqlDialect;
9+
import org.apache.calcite.sql.dialect.AnsiSqlDialect;
10+
import org.apache.calcite.sql.dialect.MysqlSqlDialect;
11+
import org.apache.calcite.sql.dialect.CalciteSqlDialect;
12+
13+
import java.util.Collection;
14+
import java.util.Collections;
15+
import java.util.List;
16+
import java.util.Locale;
17+
import java.util.function.Function;
18+
import java.util.stream.Collectors;
19+
import java.sql.SQLException;
20+
21+
/** Specifies an abstract Job with concrete YAML by applying JobTemplates. */
22+
class K8sJobDeployer extends K8sYamlDeployer<Job> {
23+
24+
private final K8sApi<V1alpha1JobTemplate, V1alpha1JobTemplateList> jobTemplateApi;
25+
26+
K8sJobDeployer(K8sContext context) {
27+
super(context);
28+
this.jobTemplateApi = new K8sApi<>(context, K8sApiEndpoints.JOB_TEMPLATES);
29+
}
30+
31+
@Override
32+
public List<String> specify(Job job) throws SQLException {
33+
Function<SqlDialect, String> sql = job.sql();
34+
Template.Environment env = Template.Environment.EMPTY
35+
.with("name", job.sink().database() + "-" + job.sink().table().toLowerCase(Locale.ROOT))
36+
.with("database", job.sink().database())
37+
.with("schema", job.sink().schema())
38+
.with("table", job.sink().table())
39+
.with("sql", sql.apply(MysqlSqlDialect.DEFAULT))
40+
.with("ansisql", sql.apply(AnsiSqlDialect.DEFAULT))
41+
.with("calcitesql", sql.apply(CalciteSqlDialect.DEFAULT))
42+
.with(job.sink().options());
43+
return jobTemplateApi.list().stream()
44+
.map(x -> x.getSpec())
45+
.filter(x -> x.getDatabases() == null || x.getDatabases().contains(job.sink().database()))
46+
.filter(x -> x.getYaml() != null)
47+
.map(x -> x.getYaml())
48+
.map(x -> new Template.SimpleTemplate(x).render(env))
49+
.collect(Collectors.toList());
50+
}
51+
}

‎hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sMaterializedViewDeployer.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,6 @@ protected V1alpha1View toK8sObject(MaterializedView view) {
2525
.metadata(new V1ObjectMeta().name(name))
2626
.spec(new V1alpha1ViewSpec()
2727
.view(q.pollLast()).schema(q.pollLast())
28-
.sql(view.sql()).materialized(true));
28+
.sql(view.viewSql()).materialized(true));
2929
}
3030
}

‎hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sPipelineDeployer.java

+3-5
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
import com.linkedin.hoptimator.k8s.models.V1alpha1PipelineList;
1212

1313
import org.apache.calcite.rel.RelNode;
14+
import org.apache.calcite.sql.dialect.AnsiSqlDialect;
1415

1516
import io.kubernetes.client.openapi.models.V1ObjectMeta;
1617

@@ -27,12 +28,9 @@ class K8sPipelineDeployer extends K8sDeployer<MaterializedView, V1alpha1Pipeline
2728
@Override
2829
protected V1alpha1Pipeline toK8sObject(MaterializedView view) throws SQLException {
2930
String name = K8sUtils.canonicalizeName(view.path());
30-
RelNode rel = HoptimatorDriver.convert(view.context(), view.sql()).root.rel;
31-
PipelineRel.Implementor plan = DeploymentService.plan(rel);
32-
String sql = plan.insertInto(view).sql();
33-
plan.implement(view, Sink.class); // include the sink in the plan
34-
String yaml = plan.pipeline().specify().stream()
31+
String yaml = view.pipeline().specify().stream()
3532
.collect(Collectors.joining("\n---\n"));
33+
String sql = view.pipelineSql().apply(AnsiSqlDialect.DEFAULT);
3634
return new V1alpha1Pipeline().kind(K8sApiEndpoints.PIPELINES.kind())
3735
.apiVersion(K8sApiEndpoints.PIPELINES.apiVersion())
3836
.metadata(new V1ObjectMeta().name(name))

‎hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sSourceDeployer.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@
1212
import java.sql.SQLException;
1313
import java.util.Locale;
1414

15-
/** Specifies an abstract Source/Sink with concrete YAML by applying TableTemplates. */
15+
/** Specifies an abstract Source with concrete YAML by applying TableTemplates. */
1616
class K8sSourceDeployer extends K8sYamlDeployer<Source> {
1717

1818
private final K8sApi<V1alpha1TableTemplate, V1alpha1TableTemplateList> tableTemplateApi;

‎hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sYamlApi.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,7 @@ public void update(String yaml) throws SQLException {
6161
private DynamicKubernetesObject objFromYaml(String yaml) {
6262
DynamicKubernetesObject obj = Dynamics.newFromYaml(yaml);
6363
if (obj.getMetadata().getNamespace() == null) {
64-
obj.getMetadata().namespace(context.namespace());
64+
obj.setMetadata(obj.getMetadata().namespace(context.namespace()));
6565
}
6666
return obj;
6767
}

‎hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/models/V1alpha1Database.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@
3030
* Database metadata.
3131
*/
3232
@ApiModel(description = "Database metadata.")
33-
@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2024-11-26T23:21:26.031Z[Etc/UTC]")
33+
@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2024-12-04T03:00:43.571Z[Etc/UTC]")
3434
public class V1alpha1Database implements io.kubernetes.client.common.KubernetesObject {
3535
public static final String SERIALIZED_NAME_API_VERSION = "apiVersion";
3636
@SerializedName(SERIALIZED_NAME_API_VERSION)

‎hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/models/V1alpha1DatabaseList.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@
3232
* DatabaseList is a list of Database
3333
*/
3434
@ApiModel(description = "DatabaseList is a list of Database")
35-
@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2024-11-26T23:21:26.031Z[Etc/UTC]")
35+
@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2024-12-04T03:00:43.571Z[Etc/UTC]")
3636
public class V1alpha1DatabaseList implements io.kubernetes.client.common.KubernetesListObject {
3737
public static final String SERIALIZED_NAME_API_VERSION = "apiVersion";
3838
@SerializedName(SERIALIZED_NAME_API_VERSION)

‎hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/models/V1alpha1DatabaseSpec.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@
2828
* Database spec.
2929
*/
3030
@ApiModel(description = "Database spec.")
31-
@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2024-11-26T23:21:26.031Z[Etc/UTC]")
31+
@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2024-12-04T03:00:43.571Z[Etc/UTC]")
3232
public class V1alpha1DatabaseSpec {
3333
/**
3434
* SQL dialect the driver expects.

0 commit comments

Comments
 (0)