[Feature][starter] support user define parameter on spark/flink engine #6387

liunaijie · 2024-02-26T08:57:08Z

Purpose of this pull request

close #6376

Does this PR introduce any user-facing change?

No

How was this patch tested?

add example to test.

Check list

If any new Jar binary package adding in your PR, please add License Notice according
New License Guide
If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
If you are contributing the connector code, please check that the following files are updated:
1. Update change log that in connector document. For more details you can refer to connector-v2
2. Update plugin-mapping.properties and add new connector information in it
3. Update the pom file of seatunnel-dist
Update the release-note.

JueLance

Thank you for your contribution! There are some problems in your commit. Unit test for project connector-cdc-sqlserver-e2e was failed, please check your commit.

Hisoka-X · 2024-03-05T10:32:16Z

...tunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigBuilder.java

+            if (variables != null) {
+                Map<String, String> map =
+                        variables.stream()
+                                .filter(Objects::nonNull)
+                                .map(variable -> variable.split("=", 2))
+                                .filter(pair -> pair.length == 2F)
+                                .collect(Collectors.toMap(pair -> pair[0], pair -> pair[1]));
+                config =
+                        config.resolveWith(
+                                ConfigFactory.parseMap(map),
+                                ConfigResolveOptions.defaults().setAllowUnresolved(true));


Seem like this part show twice in this file. Could you refactor it with new method?

Hisoka-X · 2024-03-05T10:33:22Z

...mple/src/main/java/org/apache/seatunnel/example/spark/v2/SeaTunnelSparkVariablesExample.java

        String configFile = getTestConfigFile(configurePath);
        SparkCommandArgs sparkCommandArgs = new SparkCommandArgs();
        sparkCommandArgs.setConfigFile(configFile);
        sparkCommandArgs.setCheckConfig(false);
-        sparkCommandArgs.setVariables(null);
+        sparkCommandArgs.setVariables(Collections.singletonList("intVal=1"));


Please revert this.

Hisoka-X · 2024-03-05T10:33:58Z

...ctor-v2-example/src/main/java/org/apache/seatunnel/example/spark/v2/SeaTunnelApiExample.java

@@ -17,16 +17,36 @@

 package org.apache.seatunnel.example.spark.v2;


Please do not touch example module.

Hisoka-X · 2024-03-05T10:34:04Z

.../seatunnel-spark-connector-v2-example/src/main/resources/examples/spark.batch.variables.conf

@@ -0,0 +1,99 @@
+#


...nnector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/UserVariableIT.java

...n/src/main/java/org/apache/seatunnel/core/starter/flink/command/FlinkTaskExecuteCommand.java

liunaijie · 2024-03-21T07:29:58Z

@Hisoka-X PTAL

Hisoka-X

Could the processing parameter logic of different engines be unified together? It seems that zeta's processing code is a bit scattered now.

seatunnel/seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/parse/MultipleTableJobConfigParser.java

Line 130 in e5df7f0

this.seaTunnelJobConfig = ConfigBuilder.of(Paths.get(jobDefineFilePath));

seatunnel/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/args/ClientCommandArgs.java

Line 119 in e5df7f0

private void userParamsToSysEnv() {

...nel-flink-13-starter/src/main/java/org/apache/seatunnel/core/starter/flink/FlinkStarter.java

...flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/FlinkStarter.java

Hisoka-X · 2024-04-08T09:32:46Z

How about add some intro in https://github.com/apache/seatunnel/blob/e5df7f00f8e21edcc5168c4f3dedfc7f85a4a150/docs/en/concept/config.md? Then user can easiler to found this feature.

liunaijie · 2024-04-08T10:14:09Z

How about add some intro in https://github.com/apache/seatunnel/blob/e5df7f00f8e21edcc5168c4f3dedfc7f85a4a150/docs/en/concept/config.md? Then user can easiler to found this feature.

Sure, will do it.

Hisoka-X · 2024-04-09T02:18:19Z

Could the processing parameter logic of different engines be unified together? It seems that zeta's processing code is a bit scattered now.

seatunnel/seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/parse/MultipleTableJobConfigParser.java

Line 130 in e5df7f0

this.seaTunnelJobConfig = ConfigBuilder.of(Paths.get(jobDefineFilePath));

seatunnel/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/args/ClientCommandArgs.java

Line 119 in e5df7f0

private void userParamsToSysEnv() {

How about this one?

liunaijie · 2024-04-09T09:57:30Z

...er/src/test/java/org/apache/seatunnel/core/starter/seatunnel/args/ClientCommandArgsTest.java

delete this ut, because we has e2e test
And in this ut, it's call clientCommandArgs.buildCommand() in this method, it will set the variables to system properties. then the config parse read from system properties.
as we unified the process, we won't set the variables to system properties, then this test will be fail.

liunaijie · 2024-04-10T08:55:39Z

Could the processing parameter logic of different engines be unified together? It seems that zeta's processing code is a bit scattered now.

seatunnel/seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/parse/MultipleTableJobConfigParser.java

Line 130 in e5df7f0

this.seaTunnelJobConfig = ConfigBuilder.of(Paths.get(jobDefineFilePath));

seatunnel/seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/args/ClientCommandArgs.java

Line 119 in e5df7f0

private void userParamsToSysEnv() {

How about this one?

Done, PTAL

Hisoka-X · 2024-04-10T12:00:17Z

docs/en/concept/config.md

+}
+```
+
+In the above config, i define some variables, like `${rowNum}`, `${resName}`.


Suggested change

In the above config, i define some variables, like `${rowNum}`, `${resName}`.

In the above config, we define some variables, like `${rowNum}`, `${resName}`.

Hisoka-X · 2024-04-10T12:00:50Z

docs/en/concept/config.md

+./bin/seatunnel.sh -c <this_config_file> -i rowNum=10 -i resName=fake -i strTemplate=abc -i nameVal=abc
+```
+
+Then the finial submitted config is:


Suggested change

Then the finial submitted config is:

Then the final submitted config is:

Hisoka-X · 2024-04-10T12:09:35Z

...er/src/test/java/org/apache/seatunnel/core/starter/seatunnel/args/ClientCommandArgsTest.java

-        int fakeParallelism = 16;
-        String username = "seatunnel=2.3.1";
-        String password = "dsjr42=4wfskahdsd=w1chh";
-        String fakeSourceTable = "fake";
-        String fakeSinkTable = "sink";
-        String list = "[par1=20230829,par2=20230829]";
-        String blankSpace = "2023-12-26 11:30:00";
-        String[] args = {
-            "-c",
-            "/args/user_defined_params.conf",
-            "-e",
-            "local",
-            "-i",
-            "fake_source_table=" + fakeSourceTable,
-            "-i",
-            "fake_parallelism=" + fakeParallelism,
-            "-i",
-            "fake_sink_table=" + fakeSinkTable,
-            "-i",
-            "password=" + password,
-            "-i",
-            "username=" + username,
-            "-i",
-            "blankSpace=" + blankSpace,
-            "-i",
-            "list=" + list,
-            "-i",
-            "sql=" + "\"select a , b from fake_source_table\""


We need add test case like this:

String username = "seatunnel=2.3.1"; String password = "dsjr42=4wfskahdsd=w1chh"; String list = "[par1=20230829,par2=20230829]"; String blankSpace = "2023-12-26 11:30:00"; String sql = "\"select a , b from fake_source_table\""

It contains =, , \" so on

Hisoka-X · 2024-04-11T13:18:57Z

...core-starter/src/main/java/org/apache/seatunnel/core/starter/command/ConfEncryptCommand.java

+                            .collect(Collectors.toMap(pair -> pair[0], pair -> pair[1]));
+            config =
+                    config.resolveWith(
+                            ConfigFactory.parseMap(map),


Why use parameter map as arg in here but https://github.com/apache/seatunnel/pull/6387/files#diff-f4fa1a9b065718cc135d232712063b7b3a9135933396dd2674253da754d95c86R122 use system properties?

oh, i need change all of them to system properties.
when i test the array type variabel replacement, i find it only can be replace when use system properties.

Hisoka-X

LGTM, please retrigger failed CI.

Hisoka-X · 2024-04-12T04:50:25Z

Something not right.

liunaijie · 2024-04-12T05:09:37Z

Something not right.

yes i see this error. i have some test code in other branch.
in my local desktop (windows) this ut can run successful, so run some test by github action.

liunaijie · 2024-04-15T07:10:29Z

Something not right.

I has some test, find in github action windows test, it can't read the user variables system properties. But in my local window desktop, it can read the properties, so i am disable this test in window to pass to ci.

the screenshot in my local windows desktop

Hisoka-X · 2024-04-16T13:26:16Z

Thanks @liunaijie for updated. But I think we should find the reason instead of disable it.

hailin0 · 2024-04-17T02:02:26Z

Thanks @liunaijie for updated. But I think we should find the reason instead of disable it.

+1

@liunaijie

liunaijie · 2024-04-17T06:45:15Z

@Hisoka-X @hailin0 Done, PTAL.
When call ConfigFactory.systemProperties() method, it has an static variable to hold the system properties.
so it will only read system properties once. if the properties changed, it can't read the new properties.

liunaijie · 2024-04-29T08:27:44Z

@Hisoka-X @hailin0 PTAL

Hisoka-X

Oh almost forgot this one. LGTM. Thanks @liunaijie

JueLance reviewed Feb 26, 2024

View reviewed changes

Hisoka-X reviewed Mar 5, 2024

View reviewed changes

Hisoka-X added feature New feature Spark Flink labels Mar 5, 2024

liunaijie force-pushed the user-define-parameter branch from d999cac to c964f1f Compare March 12, 2024 05:15

[Improve][starter] support user define parameter on spark/flink engine

9943122

liunaijie force-pushed the user-define-parameter branch 2 times, most recently from 89f00f6 to facd28f Compare March 12, 2024 05:30

liunaijie commented Mar 12, 2024

View reviewed changes

...nnector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/UserVariableIT.java Outdated Show resolved Hide resolved

liunaijie commented Mar 12, 2024

View reviewed changes

...n/src/main/java/org/apache/seatunnel/core/starter/flink/command/FlinkTaskExecuteCommand.java Show resolved Hide resolved

[Improve][e2e] e2e support user defined parameter

aa00bfd

liunaijie force-pushed the user-define-parameter branch from ff8bed4 to aa00bfd Compare March 14, 2024 01:28

hailin0 added this to the 2.3.5 milestone Mar 29, 2024

Hisoka-X reviewed Apr 8, 2024

View reviewed changes

...nel-flink-13-starter/src/main/java/org/apache/seatunnel/core/starter/flink/FlinkStarter.java Show resolved Hide resolved

...flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/FlinkStarter.java Show resolved Hide resolved

[Improve] update config about variable replacement

f09981c

liunaijie force-pushed the user-define-parameter branch 2 times, most recently from a266f91 to 25da29d Compare April 9, 2024 08:17

[Improve] update zeta/spark

9d50885

liunaijie force-pushed the user-define-parameter branch from 25da29d to 9d50885 Compare April 9, 2024 09:53

liunaijie commented Apr 9, 2024

View reviewed changes

Hisoka-X reviewed Apr 10, 2024

View reviewed changes

[Improve][starter] add test and update document

645b1d4

liunaijie force-pushed the user-define-parameter branch from e2a6f12 to 645b1d4 Compare April 11, 2024 10:04

Merge branch 'dev' into user-define-parameter

93fb72c

[Improve][starter] add test and update document

1c70f7a

Hisoka-X reviewed Apr 11, 2024

View reviewed changes

unified config replacement way

0259700

Hisoka-X previously approved these changes Apr 12, 2024

View reviewed changes

github-actions bot added approved reviewed labels Apr 12, 2024

disable variable replace ut on windows

2f90671

liunaijie dismissed Hisoka-X’s stale review via 2f90671 April 15, 2024 07:06

github-actions bot removed approved reviewed labels Apr 15, 2024

Hisoka-X removed this from the 2.3.5 milestone Apr 15, 2024

fix

3629c5d

liunaijie mentioned this pull request Apr 29, 2024

[Bug] [shell client] variable not correct when it has space, eg: -i dt="2024-04-28 00:00:00" #6769

Closed

3 tasks

Hisoka-X approved these changes Apr 29, 2024

View reviewed changes

github-actions bot added approved reviewed labels Apr 29, 2024

hailin0 approved these changes Apr 30, 2024

View reviewed changes

hailin0 merged commit e2a4b64 into apache:dev Apr 30, 2024
5 of 6 checks passed

liunaijie deleted the user-define-parameter branch May 11, 2024 09:55

hailin0 mentioned this pull request May 13, 2024

SeaTunnel community meeting Topic collect #6848

Closed

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Feature][starter] support user define parameter on spark/flink engine #6387

[Feature][starter] support user define parameter on spark/flink engine #6387

liunaijie commented Feb 26, 2024

JueLance left a comment

Hisoka-X Mar 5, 2024

liunaijie Mar 12, 2024

Hisoka-X Mar 5, 2024

liunaijie Mar 12, 2024

Hisoka-X Mar 5, 2024

Hisoka-X Mar 5, 2024

liunaijie commented Mar 21, 2024

Hisoka-X left a comment

Hisoka-X commented Apr 8, 2024

liunaijie commented Apr 8, 2024

Hisoka-X commented Apr 9, 2024

liunaijie Apr 9, 2024

liunaijie commented Apr 10, 2024

Hisoka-X Apr 10, 2024

Hisoka-X Apr 10, 2024

Hisoka-X Apr 10, 2024

Hisoka-X Apr 11, 2024

liunaijie Apr 12, 2024

Hisoka-X left a comment

Hisoka-X commented Apr 12, 2024

liunaijie commented Apr 12, 2024

liunaijie commented Apr 15, 2024

Hisoka-X commented Apr 16, 2024

hailin0 commented Apr 17, 2024

liunaijie commented Apr 17, 2024 •

edited

Loading

liunaijie commented Apr 29, 2024

Hisoka-X left a comment

		@@ -17,16 +17,36 @@

		package org.apache.seatunnel.example.spark.v2;

	In the above config, i define some variables, like `${rowNum}`, `${resName}`.
	In the above config, we define some variables, like `${rowNum}`, `${resName}`.

	Then the finial submitted config is:
	Then the final submitted config is:

[Feature][starter] support user define parameter on spark/flink engine #6387

[Feature][starter] support user define parameter on spark/flink engine #6387

Conversation

liunaijie commented Feb 26, 2024

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

JueLance left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

liunaijie commented Mar 21, 2024

Hisoka-X left a comment

Choose a reason for hiding this comment

Hisoka-X commented Apr 8, 2024

liunaijie commented Apr 8, 2024

Hisoka-X commented Apr 9, 2024

Choose a reason for hiding this comment

liunaijie commented Apr 10, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Hisoka-X left a comment

Choose a reason for hiding this comment

Hisoka-X commented Apr 12, 2024

liunaijie commented Apr 12, 2024

liunaijie commented Apr 15, 2024

Hisoka-X commented Apr 16, 2024

hailin0 commented Apr 17, 2024

liunaijie commented Apr 17, 2024 • edited Loading

liunaijie commented Apr 29, 2024

Hisoka-X left a comment

Choose a reason for hiding this comment

liunaijie commented Apr 17, 2024 •

edited

Loading