You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by be...@apache.org on 2023/03/31 05:55:52 UTC
[incubator-streampark] branch dev updated: [Feature] Support Flink 1.17 application submit (#2533)
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
The following commit(s) were added to refs/heads/dev by this push:
new 03435a1c7 [Feature] Support Flink 1.17 application submit (#2533)
03435a1c7 is described below
commit 03435a1c750d6c7b706591993d92dce19f630f40
Author: lvshaokang <lv...@hotmail.com>
AuthorDate: Fri Mar 31 13:55:45 2023 +0800
[Feature] Support Flink 1.17 application submit (#2533)
* [Feature] Support Flink 1.17 commit
---
.../streampark/common/conf/ConfigConst.scala | 2 +
.../streampark-console-service/pom.xml | 7 +
.../console/core/runner/EnvInitializer.java | 2 +-
.../flink/client/impl/YarnApplicationClient.scala | 1 +
.../flink/packer/pipeline/BuildRequest.scala | 1 +
.../streampark/flink/proxy/FlinkShimsProxy.scala | 2 +-
streampark-flink/streampark-flink-shims/pom.xml | 1 +
.../streampark-flink-shims_flink-1.17/pom.xml | 148 +++++++++++++++++++++
.../streampark/flink/core/FlinkClusterClient.scala | 39 ++++++
.../flink/core/FlinkKubernetesClient.scala | 30 +++++
.../streampark/flink/core/StreamTableContext.scala | 131 ++++++++++++++++++
.../streampark/flink/core/TableContext.scala | 88 ++++++++++++
.../apache/streampark/flink/core/TableExt.scala | 42 ++++++
13 files changed, 492 insertions(+), 2 deletions(-)
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigConst.scala b/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigConst.scala
index 029b258f6..288769bb6 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigConst.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigConst.scala
@@ -20,6 +20,8 @@ import java.time.LocalDateTime
object ConfigConst {
+ val STREAMPARK_FLINK_SHIMS_JAR_REGEX = "streampark-flink-shims_flink-(1.12|1.13|1.14|1.15|1.16|1.17)_(2.11|2.12)-(.*).jar"
+
/**
* common const
*/
diff --git a/streampark-console/streampark-console-service/pom.xml b/streampark-console/streampark-console-service/pom.xml
index 61f192ed6..7e7e28324 100644
--- a/streampark-console/streampark-console-service/pom.xml
+++ b/streampark-console/streampark-console-service/pom.xml
@@ -563,6 +563,13 @@
<version>${project.version}</version>
<outputDirectory>${project.build.directory}/shims</outputDirectory>
</dependency>
+ <!-- flink 1.17 support-->
+ <dependency>
+ <groupId>org.apache.streampark</groupId>
+ <artifactId>streampark-flink-shims_flink-1.17_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <outputDirectory>${project.build.directory}/shims</outputDirectory>
+ </dependency>
<!-- flink-submit-core -->
<dependency>
<groupId>org.apache.streampark</groupId>
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java
index 04155ae23..bd9929bdf 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java
@@ -69,7 +69,7 @@ public class EnvInitializer implements ApplicationRunner {
private static final Pattern PATTERN_FLINK_SHIMS_JAR =
Pattern.compile(
- "^streampark-flink-shims_flink-(1.12|1.13|1.14|1.15|1.16)_(2.11|2.12)-(.*).jar$",
+ String.format("^%s$", ConfigConst.STREAMPARK_FLINK_SHIMS_JAR_REGEX()),
Pattern.CASE_INSENSITIVE | Pattern.DOTALL);
@Override
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala
index d2729833d..d410ec83a 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala
@@ -74,6 +74,7 @@ object YarnApplicationClient extends YarnClientTrait {
case Array(1, 14, _) => array += s"${workspace.APP_SHIMS}/flink-1.14"
case Array(1, 15, _) => array += s"${workspace.APP_SHIMS}/flink-1.15"
case Array(1, 16, _) => array += s"${workspace.APP_SHIMS}/flink-1.16"
+ case Array(1, 17, _) => array += s"${workspace.APP_SHIMS}/flink-1.17"
case _ => throw new UnsupportedOperationException(s"Unsupported flink version: ${submitRequest.flinkVersion}")
}
val jobLib = s"${workspace.APP_WORKSPACE}/${submitRequest.id}/lib"
diff --git a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala
index 6a512e7f6..8c93acbd3 100644
--- a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala
+++ b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala
@@ -64,6 +64,7 @@ sealed trait FlinkBuildParam extends BuildParam {
case Array(1, 14, _) => s"${localWorkspace.APP_SHIMS}/flink-1.14"
case Array(1, 15, _) => s"${localWorkspace.APP_SHIMS}/flink-1.15"
case Array(1, 16, _) => s"${localWorkspace.APP_SHIMS}/flink-1.16"
+ case Array(1, 17, _) => s"${localWorkspace.APP_SHIMS}/flink-1.17"
case _ => throw new UnsupportedOperationException(s"Unsupported flink version: $flinkVersion")
}
}
diff --git a/streampark-flink/streampark-flink-proxy/src/main/scala/org/apache/streampark/flink/proxy/FlinkShimsProxy.scala b/streampark-flink/streampark-flink-proxy/src/main/scala/org/apache/streampark/flink/proxy/FlinkShimsProxy.scala
index 1f94fa498..a25d13c98 100644
--- a/streampark-flink/streampark-flink-proxy/src/main/scala/org/apache/streampark/flink/proxy/FlinkShimsProxy.scala
+++ b/streampark-flink/streampark-flink-proxy/src/main/scala/org/apache/streampark/flink/proxy/FlinkShimsProxy.scala
@@ -34,7 +34,7 @@ object FlinkShimsProxy extends Logger {
Pattern.CASE_INSENSITIVE | Pattern.DOTALL)
private[this] val SHIMS_PATTERN = Pattern.compile(
- "streampark-flink-shims_flink-(1.12|1.13|1.14|1.15|1.16)_(2.11|2.12)-(.*).jar",
+ ConfigConst.STREAMPARK_FLINK_SHIMS_JAR_REGEX,
Pattern.CASE_INSENSITIVE | Pattern.DOTALL)
private[this] val SHIMS_CLASS_LOADER_CACHE = MutableMap[String, ClassLoader]()
diff --git a/streampark-flink/streampark-flink-shims/pom.xml b/streampark-flink/streampark-flink-shims/pom.xml
index 27f719e8d..41ca70913 100644
--- a/streampark-flink/streampark-flink-shims/pom.xml
+++ b/streampark-flink/streampark-flink-shims/pom.xml
@@ -43,6 +43,7 @@
<!-- flink 1.15+ only support scala 2.12 -->
<module>streampark-flink-shims_flink-1.15</module>
<module>streampark-flink-shims_flink-1.16</module>
+ <module>streampark-flink-shims_flink-1.17</module>
</modules>
</profile>
</profiles>
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/pom.xml b/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/pom.xml
new file mode 100644
index 000000000..1334a304f
--- /dev/null
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/pom.xml
@@ -0,0 +1,148 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.streampark</groupId>
+ <artifactId>streampark-flink-shims</artifactId>
+ <version>2.1.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>streampark-flink-shims_flink-1.17_${scala.binary.version}</artifactId>
+ <name>StreamPark : Flink Shims 1.17</name>
+
+ <properties>
+ <flink.version>1.17.0</flink.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.streampark</groupId>
+ <artifactId>streampark-flink-shims-base_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <!--flink-->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-scala_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-scala_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-java-uber</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-scala-bridge_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-statebackend-rocksdb</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-yarn</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-yarn-api</artifactId>
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-kubernetes</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <createDependencyReducedPom>true</createDependencyReducedPom>
+ <dependencyReducedPomLocation>${project.basedir}/target/dependency-reduced-pom.xml</dependencyReducedPomLocation>
+ <artifactSet>
+ <includes>
+ <include>org.apache.flink:flink-table-api-scala-bridge_${scala.binary.version}</include>
+ </includes>
+ </artifactSet>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/src/main/scala/org/apache/streampark/flink/core/FlinkClusterClient.scala b/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/src/main/scala/org/apache/streampark/flink/core/FlinkClusterClient.scala
new file mode 100644
index 000000000..a68fdac1e
--- /dev/null
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/src/main/scala/org/apache/streampark/flink/core/FlinkClusterClient.scala
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.streampark.flink.core
+
+import org.apache.flink.api.common.JobID
+import org.apache.flink.client.program.ClusterClient
+import org.apache.flink.core.execution.SavepointFormatType
+
+import java.util.concurrent.CompletableFuture
+
+class FlinkClusterClient[T](clusterClient: ClusterClient[T]) extends FlinkClientTrait[T](clusterClient) {
+
+ override def triggerSavepoint(jobID: JobID, savepointDir: String): CompletableFuture[String] = {
+ clusterClient.triggerSavepoint(jobID, savepointDir, SavepointFormatType.DEFAULT)
+ }
+
+ override def cancelWithSavepoint(jobID: JobID, savepointDirectory: String): CompletableFuture[String] = {
+ clusterClient.cancelWithSavepoint(jobID, savepointDirectory, SavepointFormatType.DEFAULT)
+ }
+
+ override def stopWithSavepoint(jobID: JobID, advanceToEndOfEventTime: Boolean, savepointDirectory: String): CompletableFuture[String] = {
+ clusterClient.stopWithSavepoint(jobID, advanceToEndOfEventTime, savepointDirectory, SavepointFormatType.DEFAULT)
+ }
+
+}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/src/main/scala/org/apache/streampark/flink/core/FlinkKubernetesClient.scala b/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/src/main/scala/org/apache/streampark/flink/core/FlinkKubernetesClient.scala
new file mode 100644
index 000000000..299c04a27
--- /dev/null
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/src/main/scala/org/apache/streampark/flink/core/FlinkKubernetesClient.scala
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.streampark.flink.core
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesService
+
+import java.util.Optional
+
+class FlinkKubernetesClient(kubeClient: FlinkKubeClient) extends FlinkKubernetesClientTrait(kubeClient) {
+
+ override def getService(serviceName: String): Optional[KubernetesService] = {
+ kubeClient.getService(serviceName)
+ }
+
+}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/src/main/scala/org/apache/streampark/flink/core/StreamTableContext.scala b/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/src/main/scala/org/apache/streampark/flink/core/StreamTableContext.scala
new file mode 100644
index 000000000..ad51b1ec6
--- /dev/null
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/src/main/scala/org/apache/streampark/flink/core/StreamTableContext.scala
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.flink.core
+
+import org.apache.flink.api.java.utils.ParameterTool
+import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
+import org.apache.flink.table.api.{CompiledPlan, ExplainDetail, ExplainFormat, PlanReference, Schema, Table, TableDescriptor, TableResult}
+import org.apache.flink.table.api.bridge.scala.{StreamStatementSet, StreamTableEnvironment}
+import org.apache.flink.table.connector.ChangelogMode
+import org.apache.flink.table.module.ModuleEntry
+import org.apache.flink.table.resource.ResourceUri
+import org.apache.flink.table.types.AbstractDataType
+import org.apache.flink.types.Row
+
+import java.util.{List => JList}
+
+class StreamTableContext(override val parameter: ParameterTool, private val streamEnv: StreamExecutionEnvironment, private val tableEnv: StreamTableEnvironment)
+ extends FlinkStreamTableTrait(parameter, streamEnv, tableEnv) {
+
+ def this(args: (ParameterTool, StreamExecutionEnvironment, StreamTableEnvironment)) = this(args._1, args._2, args._3)
+
+ def this(args: StreamTableEnvConfig) = this(FlinkTableInitializer.initialize(args))
+
+ override def fromDataStream[T](dataStream: DataStream[T], schema: Schema): Table = tableEnv.fromDataStream[T](dataStream, schema)
+
+ override def fromChangelogStream(dataStream: DataStream[Row]): Table = tableEnv.fromChangelogStream(dataStream)
+
+ override def fromChangelogStream(dataStream: DataStream[Row], schema: Schema): Table = tableEnv.fromChangelogStream(dataStream, schema)
+
+ override def fromChangelogStream(dataStream: DataStream[Row], schema: Schema, changelogMode: ChangelogMode): Table =
+ tableEnv.fromChangelogStream(dataStream, schema, changelogMode)
+
+ override def createTemporaryView[T](path: String, dataStream: DataStream[T], schema: Schema): Unit = tableEnv.createTemporaryView[T](path, dataStream, schema)
+
+ override def toDataStream(table: Table): DataStream[Row] = {
+ isConvertedToDataStream = true
+ tableEnv.toDataStream(table)
+ }
+
+ override def toDataStream[T](table: Table, targetClass: Class[T]): DataStream[T] = {
+ isConvertedToDataStream = true
+ tableEnv.toDataStream[T](table, targetClass)
+ }
+
+ override def toDataStream[T](table: Table, targetDataType: AbstractDataType[_]): DataStream[T] = {
+ isConvertedToDataStream = true
+ tableEnv.toDataStream[T](table, targetDataType)
+ }
+
+ override def toChangelogStream(table: Table): DataStream[Row] = {
+ isConvertedToDataStream = true
+ tableEnv.toChangelogStream(table)
+ }
+
+ override def toChangelogStream(table: Table, targetSchema: Schema): DataStream[Row] = {
+ isConvertedToDataStream = true
+ tableEnv.toChangelogStream(table, targetSchema)
+ }
+
+ override def toChangelogStream(table: Table, targetSchema: Schema, changelogMode: ChangelogMode): DataStream[Row] = {
+ isConvertedToDataStream = true
+ tableEnv.toChangelogStream(table, targetSchema, changelogMode)
+ }
+
+ override def createStatementSet(): StreamStatementSet = tableEnv.createStatementSet()
+
+ override def useModules(strings: String*): Unit = tableEnv.useModules(strings: _*)
+
+ override def createTemporaryTable(path: String, descriptor: TableDescriptor): Unit = tableEnv.createTemporaryTable(path, descriptor)
+
+ override def createTable(path: String, descriptor: TableDescriptor): Unit = tableEnv.createTable(path, descriptor)
+
+ override def from(descriptor: TableDescriptor): Table = tableEnv.from(descriptor)
+
+ override def listFullModules(): Array[ModuleEntry] = tableEnv.listFullModules()
+
+ /**
+ * @since 1.15
+ */
+ override def listTables(s: String, s1: String): Array[String] = tableEnv.listTables(s, s1)
+
+ /**
+ * @since 1.15
+ */
+ override def loadPlan(planReference: PlanReference): CompiledPlan = tableEnv.loadPlan(planReference)
+
+ /**
+ * @since 1.15
+ */
+ override def compilePlanSql(s: String): CompiledPlan = tableEnv.compilePlanSql(s)
+
+ /**
+ * @since 1.17
+ */
+ override def createFunction(path: String, className: String, resourceUris: JList[ResourceUri]): Unit = tableEnv.createFunction(path, className, resourceUris)
+
+ /**
+ * @since 1.17
+ */
+ override def createFunction(path: String, className: String, resourceUris: JList[ResourceUri], ignoreIfExists: Boolean): Unit = tableEnv.createFunction(path, className, resourceUris, ignoreIfExists)
+
+ /**
+ * @since 1.17
+ */
+ override def createTemporaryFunction(path: String, className: String, resourceUris: JList[ResourceUri]): Unit = tableEnv.createTemporaryFunction(path, className, resourceUris)
+
+ /**
+ * @since 1.17
+ */
+ override def createTemporarySystemFunction(name: String, className: String, resourceUris: JList[ResourceUri]): Unit = tableEnv.createTemporarySystemFunction(name, className, resourceUris)
+
+ /**
+ * @since 1.17
+ */
+ override def explainSql(statement: String, format: ExplainFormat, extraDetails: ExplainDetail*): String = tableEnv.explainSql(statement, format, extraDetails: _*)
+}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/src/main/scala/org/apache/streampark/flink/core/TableContext.scala b/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/src/main/scala/org/apache/streampark/flink/core/TableContext.scala
new file mode 100644
index 000000000..f624da32e
--- /dev/null
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/src/main/scala/org/apache/streampark/flink/core/TableContext.scala
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.flink.core
+
+import org.apache.flink.api.java.utils.ParameterTool
+import org.apache.flink.table.api.{CompiledPlan, ExplainDetail, ExplainFormat, PlanReference, Table, TableDescriptor, TableEnvironment, TableResult}
+import org.apache.flink.table.module.ModuleEntry
+import org.apache.flink.table.resource.ResourceUri
+import java.util.{List => JList}
+
+class TableContext(override val parameter: ParameterTool, private val tableEnv: TableEnvironment)
+ extends FlinkTableTrait(parameter, tableEnv) {
+
+ def this(args: (ParameterTool, TableEnvironment)) = this(args._1, args._2)
+
+ def this(args: TableEnvConfig) = this(FlinkTableInitializer.initialize(args))
+
+ override def useModules(strings: String*): Unit = tableEnv.useModules(strings: _*)
+
+ override def createTemporaryTable(path: String, descriptor: TableDescriptor): Unit = {
+ tableEnv.createTemporaryTable(path, descriptor)
+ }
+
+ override def createTable(path: String, descriptor: TableDescriptor): Unit = {
+ tableEnv.createTable(path, descriptor)
+ }
+
+ override def from(tableDescriptor: TableDescriptor): Table = {
+ tableEnv.from(tableDescriptor)
+ }
+
+ override def listFullModules(): Array[ModuleEntry] = tableEnv.listFullModules()
+
+ /**
+ * @since 1.15
+ */
+ override def listTables(catalogName: String, databaseName: String): Array[String] = tableEnv.listTables(catalogName, databaseName)
+
+ /**
+ * @since 1.15
+ */
+ override def loadPlan(planReference: PlanReference): CompiledPlan = tableEnv.loadPlan(planReference)
+
+ /**
+ * @since 1.15
+ */
+ override def compilePlanSql(stmt: String): CompiledPlan = tableEnv.compilePlanSql(stmt)
+
+ /**
+ * @since 1.17
+ */
+ override def createFunction(path: String, className: String, resourceUris: JList[ResourceUri]): Unit = tableEnv.createFunction(path, className, resourceUris)
+
+ /**
+ * @since 1.17
+ */
+ override def createFunction(path: String, className: String, resourceUris: JList[ResourceUri], ignoreIfExists: Boolean): Unit = tableEnv.createFunction(path, className, resourceUris, ignoreIfExists)
+
+ /**
+ * @since 1.17
+ */
+ override def createTemporaryFunction(path: String, className: String, resourceUris: JList[ResourceUri]): Unit = tableEnv.createTemporaryFunction(path, className, resourceUris)
+
+ /**
+ * @since 1.17
+ */
+ override def createTemporarySystemFunction(name: String, className: String, resourceUris: JList[ResourceUri]): Unit = tableEnv.createTemporarySystemFunction(name, className, resourceUris)
+
+ /**
+ * @since 1.17
+ */
+ override def explainSql(statement: String, format: ExplainFormat, extraDetails: ExplainDetail*): String = tableEnv.explainSql(statement, format, extraDetails: _*)
+}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/src/main/scala/org/apache/streampark/flink/core/TableExt.scala b/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/src/main/scala/org/apache/streampark/flink/core/TableExt.scala
new file mode 100644
index 000000000..cab368e36
--- /dev/null
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims_flink-1.17/src/main/scala/org/apache/streampark/flink/core/TableExt.scala
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.flink.core
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.streaming.api.scala.DataStream
+import org.apache.flink.table.api.{Table => FlinkTable}
+import org.apache.flink.table.api.bridge.scala.{TableConversions => FlinkTableConversions}
+import org.apache.flink.types.Row
+
+object TableExt {
+
+ class Table(val table: FlinkTable) {
+ def ->(field: String, fields: String*): FlinkTable = table.as(field, fields: _*)
+ }
+
+ class TableConversions(table: FlinkTable) extends FlinkTableConversions(table) {
+
+ def \\ : DataStream[Row] = toDataStream
+
+ def >>[T: TypeInformation](implicit context: StreamTableContext): DataStream[T] = {
+ context.isConvertedToDataStream = true
+ super.toAppendStream
+ }
+ }
+
+}