Add support for external engines

linkedin · Jan 16, 2025 · eba9397 · eba9397
1 parent aae0587
commit eba9397
Show file tree

Hide file tree

Showing 20 changed files with 1,178 additions and 41 deletions.
diff --git a/README.md b/README.md
@@ -83,7 +83,7 @@ Once the Flink deployment pod has STATUS 'Running', you can forward port 8081 an
 to access the Flink dashboard.
 
 ```
-  $ kubectl port-forward basic-session-deployment-7b94b98b6b-d6jt5 8081 &
+  $ kubectl port-forward svc/basic-session-deployment-rest 8081 &
 ```
 
 See the [Flink SQL Gateway Documentation](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/sql-gateway/overview/)

diff --git a/deploy/docker/flink/docker-compose-sql-gateway.yaml b/deploy/docker/flink/docker-compose-sql-gateway.yaml
@@ -3,10 +3,12 @@ services:
     image: flink:1.18.1
     restart: unless-stopped
     entrypoint: >
-      /bin/sh -c "./bin/sql-gateway.sh start-foreground -Dsql-gateway.endpoint.rest.address=localhost"
+      /bin/sh -c "./bin/sql-gateway.sh start-foreground -Dsql-gateway.endpoint.rest.address=localhost -Drest.address=host.docker.internal"
     ports:
       - 8083:8083
     deploy:
       resources:
         limits:
           memory: 1024M
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
diff --git a/deploy/samples/demodb.yaml b/deploy/samples/demodb.yaml
@@ -29,7 +29,6 @@ spec:
   - profile-database
   - ads-database
   connector: |
-    connector = demo
-    database = {{database}}
-    table = {{table}}
+    connector = datagen
+    number-of-rows = 10
 
diff --git a/deploy/samples/flinkDeployment.yaml b/deploy/samples/flinkDeployment.yaml
@@ -7,7 +7,7 @@ spec:
   imagePullPolicy: Never
   flinkVersion: v1_18
   flinkConfiguration:
-    taskmanager.numberOfTaskSlots: "3"
+    taskmanager.numberOfTaskSlots: "6"
   serviceAccount: flink
   jobManager:
     resource:

diff --git a/deploy/samples/flinkengine.yaml b/deploy/samples/flinkengine.yaml
@@ -0,0 +1,8 @@
+apiVersion: hoptimator.linkedin.com/v1alpha1
+kind: Engine
+metadata:
+  name: flink-engine
+spec:
+  url: jdbc:flink://localhost:8083
+  dialect: Flink
+
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
@@ -8,6 +8,7 @@ flink-clients = "org.apache.flink:flink-clients:1.18.1"
 flink-connector-base = "org.apache.flink:flink-connector-base:1.18.1"
 flink-core = "org.apache.flink:flink-core:1.18.1"
 flink-csv = "org.apache.flink:flink-csv:1.18.1"
+flink-jdbc = "org.apache.flink:flink-sql-jdbc-driver-bundle:1.18.1"
 flink-streaming-java = "org.apache.flink:flink-streaming-java:1.18.1"
 flink-table-api-java = "org.apache.flink:flink-table-api-java:1.18.1"
 flink-table-api-java-bridge = "org.apache.flink:flink-table-api-java-bridge:1.18.1"

diff --git a/hoptimator-api/src/main/java/com/linkedin/hoptimator/Engine.java b/hoptimator-api/src/main/java/com/linkedin/hoptimator/Engine.java
@@ -10,4 +10,6 @@ public interface Engine {
   DataSource dataSource();
 
   SqlDialect dialect();
+
+  String url();
 }
diff --git a/hoptimator-cli/build.gradle b/hoptimator-cli/build.gradle
@@ -17,6 +17,7 @@ dependencies {
   implementation libs.calcite.core
   implementation libs.sqlline
   implementation libs.slf4j.simple
+  implementation libs.flink.jdbc
 }
 
 publishing {

diff --git a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sEngine.java b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sEngine.java
@@ -5,6 +5,8 @@
 import com.linkedin.hoptimator.Engine;
 import com.linkedin.hoptimator.SqlDialect;
 
+import java.util.Objects;
+
 import org.apache.calcite.adapter.jdbc.JdbcSchema;
 
 
@@ -17,7 +19,7 @@ public class K8sEngine implements Engine {
 
   public K8sEngine(String name, String url, SqlDialect dialect, String driver) {
     this.name = name;
-    this.url = url;
+    this.url = Objects.requireNonNull(url, "url");
     this.dialect = dialect;
     this.driver = driver;
   }
@@ -33,8 +35,13 @@ public DataSource dataSource() {
     return JdbcSchema.dataSource(url, driver, null, null);
   }
 
+  @Override
+  public String url() {
+    return url;
+  }
+
   @Override
   public SqlDialect dialect() {
-    return SqlDialect.FLINK;  // TODO fix hardcoded dialect
+    return dialect;
   }
 }
diff --git a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sEngineTable.java b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sEngineTable.java
@@ -5,7 +5,6 @@
 import java.util.Locale;
 import java.util.Optional;
 import java.util.stream.Collectors;
-import javax.sql.DataSource;
 
 import org.apache.calcite.adapter.jdbc.JdbcSchema;
 import org.apache.calcite.schema.Schema;