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 44d2bbf4e [Feature] Pyflink remote mode support maven dependency
(#3311)
44d2bbf4e is described below
commit 44d2bbf4e618bb9681bd5118bdcfba5f01503616
Author: ChengJie1053 <[email protected]>
AuthorDate: Sat Nov 4 22:59:02 2023 +0800
[Feature] Pyflink remote mode support maven dependency (#3311)
* [Feature] Pyflink remote mode support maven dependency
---
.../streampark/flink/client/trait/FlinkClientTrait.scala | 13 +++++++++----
1 file changed, 9 insertions(+), 4 deletions(-)
diff --git
a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
index 5025a92b2..435d882c7 100644
---
a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
+++
b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
@@ -22,7 +22,7 @@ import org.apache.streampark.common.conf.ConfigKeys._
import org.apache.streampark.common.conf.Workspace
import org.apache.streampark.common.enums.{ApplicationType,
FlinkDevelopmentMode, FlinkExecutionMode, FlinkRestoreMode}
import org.apache.streampark.common.fs.FsOperator
-import org.apache.streampark.common.util.{DeflaterUtils, ExceptionUtils,
FileUtils, Logger, PropertiesUtils, SystemPropertyUtils, Utils}
+import org.apache.streampark.common.util._
import org.apache.streampark.flink.client.bean._
import org.apache.streampark.flink.core.FlinkClusterClient
import org.apache.streampark.flink.core.conf.FlinkRunOption
@@ -42,11 +42,10 @@ import org.apache.flink.runtime.jobgraph.{JobGraph,
SavepointConfigOptions}
import org.apache.flink.util.FlinkException
import org.apache.flink.util.Preconditions.checkNotNull
+import java.io.File
import java.util
import java.util.{Collections, List => JavaList, Map => JavaMap}
-import scala.annotation.tailrec
-import scala.collection.JavaConverters._
import scala.collection.convert.ImplicitConversions._
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
@@ -277,7 +276,13 @@ trait FlinkClientTrait extends Logger {
throw new RuntimeException(s"$pythonVenv File does not exist")
}
// including $app/lib
- includingPipelineJars(submitRequest, flinkConfig)
+ val localLib: String =
s"${Workspace.local.APP_WORKSPACE}/${submitRequest.id}/lib"
+ if (FileUtils.exists(localLib) &&
FileUtils.directoryNotBlank(localLib)) {
+ val localLibUrl = new
File(localLib).listFiles().map(_.toURI.toURL).toList
+ pkgBuilder.setUserClassPaths(
+ Lists.newArrayList(localLibUrl: _*)
+ )
+ }
flinkConfig
// python.archives
.safeSet(PythonOptions.PYTHON_ARCHIVES, pythonVenv)
