Skip to content

Commit

Permalink
[Feature-16992][dolphinscheduler-task-seatunnel] Implement added seat…
Browse files Browse the repository at this point in the history
…unnel configuration generation in the form mode
  • Loading branch information
leibaoxin committed Feb 2, 2025
1 parent 2ae4402 commit 7cdb21d
Show file tree
Hide file tree
Showing 26 changed files with 2,011 additions and 21 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,13 @@
<version>${project.version}</version>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-datasource-api</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
</dependencies>

<build>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,4 +32,38 @@ private Constants() {
public static final String JSON_SUFFIX = "json";
public static final String CONF_SUFFIX = "conf";

public static final String EQUAL_SIGN = " = ";
public static final String LINE_BREAK = "\n";
public static final String INDENT_TWO_SPACE = " ";
public static final String INDENT_FOUR_SPACE = " ";
public static final String DOUBLE_QUOTE = "\"";
public static final String SINGLE_BRACKETS_RIGHT = "}";
public static final String SINGLE_BRACKETS_LEFT = "{";

/**
* doris config
*/
public static final String FENODES = "fenodes = ";
public static final String USERNAME = "username = ";
public static final String PASSWORD = "password = ";
public static final String DATABASE = "database = ";
public static final String TABLE = "table = ";
public static final String DORIS_CONFIG = "doris.config";
public static final String FORMAT = "format = ";
public static final String FORMAT_JSON = "json";
public static final String READ_JSON_BY_LINE = "read_json_by_line = ";
public static final String TRUE = "true";

/**
* hdfs config
*/
public static final String defaultFs = "fs.defaultFs = ";
public static final String HDFS_SITE_PATH = "hdfs_site_path = ";
public static final String KRB5_PATH = "krb5_path = ";
public static final String KERBEROS_PRINCIPAL = "kerberos_principal = ";
public static final String KERBEROS_KEYTAB_PATH = "kerberos_keytab_path = ";
public static final String TMP_PATH = "tmp_path = ";
public static final String TMP_PATH_VALUE = "/tmp/seatunnel";
public static final String REMOTE_USER = "remote_user = ";

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,60 @@
/*
* 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.dolphinscheduler.plugin.task.seatunnel;

import org.apache.dolphinscheduler.plugin.task.seatunnel.generator.ConfigTemplate;
import org.apache.dolphinscheduler.plugin.task.seatunnel.generator.DorisConfigTemplate;
import org.apache.dolphinscheduler.plugin.task.seatunnel.generator.HdfsFileConfigTemplate;
import org.apache.dolphinscheduler.plugin.task.seatunnel.generator.MysqlConfigTemplate;

public enum DbTypeEnum {

HDFS("hdfs"),
HIVE("hive"),
DORIS("doris"),
MYSQL("mysql"),
CLICKHOUSE("clickhouse");

private final String name;

DbTypeEnum(String name) {
this.name = name;
}

public String getName() {
return name;
}

public ConfigTemplate toConfigTemplate(SeatunnelParameters seatunnelConfigParameters,
SeatunnelTaskExecutionContext seatunnelTaskExecutionContext) {
switch (this) {
case HDFS:
return new HdfsFileConfigTemplate(seatunnelConfigParameters, seatunnelTaskExecutionContext);
case DORIS:
return new DorisConfigTemplate(seatunnelConfigParameters, seatunnelTaskExecutionContext);
case MYSQL:
return new MysqlConfigTemplate(seatunnelConfigParameters, seatunnelTaskExecutionContext);
default:
throw new IllegalArgumentException("not support db type => " + name);
}
}

public boolean isDatasourceType() {
return this != HDFS;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,14 @@

package org.apache.dolphinscheduler.plugin.task.seatunnel;

import org.apache.dolphinscheduler.plugin.task.api.enums.ResourceType;
import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo;
import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.DataSourceParameters;
import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
import org.apache.dolphinscheduler.plugin.task.seatunnel.parameter.BaseDataSourceParameters;
import org.apache.dolphinscheduler.plugin.task.seatunnel.parameter.SeatunnelConfigParameters;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.BooleanUtils;
import org.apache.commons.lang3.StringUtils;

Expand All @@ -36,8 +40,36 @@
@NoArgsConstructor
public class SeatunnelParameters extends AbstractParameters {

/**
* source config parameters
*/
SeatunnelConfigParameters sourceConfig;

/**
* sink config parameters
*/
SeatunnelConfigParameters sinkConfig;

/**
* task parallelism
*/
private int parallelism;

/**
* Enable custom data filtering
*/
private boolean customDataFilter;

/**
* Custom seatunnel transform config when customDataFilter set true
*/
private String customTransform;

private String startupScript;

/**
* Whether to use user-defined configuration
*/
private Boolean useCustom;

private String rawScript;
Expand All @@ -51,12 +83,60 @@ public class SeatunnelParameters extends AbstractParameters {
public boolean checkParameters() {
return Objects.nonNull(startupScript)
&& ((BooleanUtils.isTrue(useCustom) && StringUtils.isNotBlank(rawScript))
|| (BooleanUtils.isFalse(useCustom) && CollectionUtils.isNotEmpty(resourceList)
&& resourceList.size() == 1));
|| (BooleanUtils.isFalse(useCustom) && sourceConfig != null && sinkConfig != null));
}

@Override
public List<ResourceInfo> getResourceFilesList() {
return resourceList;
}

@Override
public ResourceParametersHelper getResources() {
ResourceParametersHelper resources = super.getResources();

if (Objects.nonNull(sourceConfig) && sourceConfig.getDbType().isDatasourceType()) {
BaseDataSourceParameters dataSourceParameters = (BaseDataSourceParameters) sourceConfig;
int sourceDatabaseId = dataSourceParameters.getDatabaseId();
if (sourceDatabaseId != 0) {
resources.put(ResourceType.DATASOURCE, sourceDatabaseId);
}
}

if (Objects.nonNull(sinkConfig) && sinkConfig.getDbType().isDatasourceType()) {
BaseDataSourceParameters dataSourceParameters = (BaseDataSourceParameters) sinkConfig;
int sinkDatabaseId = dataSourceParameters.getDatabaseId();
if (sinkDatabaseId != 0) {
resources.put(ResourceType.DATASOURCE, sinkDatabaseId);
}

}

return resources;
}

public SeatunnelTaskExecutionContext generateExtendedContext(ResourceParametersHelper resourceParametersHelper) {

SeatunnelTaskExecutionContext seatunnelTaskExecutionContext = new SeatunnelTaskExecutionContext();

if (!useCustom && sourceConfig.getDbType().isDatasourceType()) {
int dataSourceId = ((BaseDataSourceParameters) sourceConfig).getDatabaseId();
DataSourceParameters sourceParameters = (DataSourceParameters) resourceParametersHelper
.getResourceParameters(ResourceType.DATASOURCE, dataSourceId);
seatunnelTaskExecutionContext.setDataSourceId(dataSourceId);
seatunnelTaskExecutionContext.setDataSourceType(sourceConfig.getDbType());
seatunnelTaskExecutionContext.setSourceConnectionParams(sourceParameters.getConnectionParams());
}

if (!useCustom && sinkConfig.getDbType().isDatasourceType()) {
int dataTargetId = ((BaseDataSourceParameters) sinkConfig).getDatabaseId();
DataSourceParameters targetParameters = (DataSourceParameters) resourceParametersHelper
.getResourceParameters(ResourceType.DATASOURCE, dataTargetId);
seatunnelTaskExecutionContext.setDataTargetId(dataTargetId);
seatunnelTaskExecutionContext.setDataTargetType(sinkConfig.getDbType());
seatunnelTaskExecutionContext.setTargetConnectionParams(targetParameters.getConnectionParams());
}

return seatunnelTaskExecutionContext;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import org.apache.dolphinscheduler.plugin.task.api.shell.IShellInterceptorBuilder;
import org.apache.dolphinscheduler.plugin.task.api.shell.ShellInterceptorBuilderFactory;
import org.apache.dolphinscheduler.plugin.task.api.utils.ParameterUtils;
import org.apache.dolphinscheduler.plugin.task.seatunnel.generator.SeatunnelConfigGenerator;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.BooleanUtils;
Expand Down Expand Up @@ -72,6 +73,11 @@ public class SeatunnelTask extends AbstractRemoteTask {
*/
protected final TaskExecutionContext taskExecutionContext;

/**
* seatunnelTaskExecutionContext
*/
protected SeatunnelTaskExecutionContext seatunnelTaskExecutionContext;

/**
* constructor
*
Expand All @@ -95,6 +101,9 @@ public void init() {
if (seatunnelParameters == null || !seatunnelParameters.checkParameters()) {
throw new TaskException("SeaTunnel task params is not valid");
}

seatunnelTaskExecutionContext =
seatunnelParameters.generateExtendedContext(taskExecutionContext.getResourceParametersHelper());
}

// todo split handle to submit and track
Expand Down Expand Up @@ -160,13 +169,21 @@ protected List<String> buildOptions() throws Exception {
args.add(CONFIG_OPTIONS);
String scriptContent;
if (BooleanUtils.isTrue(seatunnelParameters.getUseCustom())) {
scriptContent = buildCustomConfigContent();
if (null != seatunnelParameters.getResourceList() && !seatunnelParameters.getResourceList().isEmpty()) {
// use resource file
String resourceFileName = seatunnelParameters.getResourceList().get(0).getResourceName();
ResourceContext resourceContext = taskExecutionContext.getResourceContext();
scriptContent = FileUtils.readFileToString(
new File(resourceContext.getResourceItem(resourceFileName).getResourceAbsolutePathInLocal()),
StandardCharsets.UTF_8);
} else {
// use custom config
scriptContent = buildCustomConfigContent();
}
} else {
String resourceFileName = seatunnelParameters.getResourceList().get(0).getResourceName();
ResourceContext resourceContext = taskExecutionContext.getResourceContext();
scriptContent = FileUtils.readFileToString(
new File(resourceContext.getResourceItem(resourceFileName).getResourceAbsolutePathInLocal()),
StandardCharsets.UTF_8);
// use generator config
scriptContent =
SeatunnelConfigGenerator.generate(seatunnelParameters, seatunnelTaskExecutionContext);
}
String filePath = buildConfigFilePath();
createConfigFileIfNotExists(scriptContent, filePath);
Expand All @@ -175,6 +192,26 @@ protected List<String> buildOptions() throws Exception {
return args;
}

// protected List<String> buildOptions() throws Exception {
// List<String> args = new ArrayList<>();s
// args.add(CONFIG_OPTIONS);
// String scriptContent;
// if (BooleanUtils.isTrue(seatunnelParameters.getUseCustom())) {
// scriptContent = buildCustomConfigContent();
// } else {
// String resourceFileName = seatunnelParameters.getResourceList().get(0).getResourceName();
// ResourceContext resourceContext = taskExecutionContext.getResourceContext();
// scriptContent = FileUtils.readFileToString(
// new File(resourceContext.getResourceItem(resourceFileName).getResourceAbsolutePathInLocal()),
// StandardCharsets.UTF_8);
// }
// String filePath = buildConfigFilePath();
// createConfigFileIfNotExists(scriptContent, filePath);
// args.add(filePath);
// args.addAll(generateTaskParameters());
// return args;
// }

private List<String> generateTaskParameters() {
Map<String, String> variables = new HashMap<>();
Map<String, Property> paramsMap = taskExecutionContext.getPrepareParamsMap();
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
/*
* 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.dolphinscheduler.plugin.task.seatunnel;

import java.io.Serializable;

import lombok.Getter;
import lombok.Setter;

@Getter
@Setter
public class SeatunnelTaskExecutionContext implements Serializable {

/**
* data source id
*/
private int dataSourceId;

/**
* data source type
*/
private DbTypeEnum dataSourceType;

/**
* source connection params
*/
private String sourceConnectionParams;

/**
* data target id
*/
private int dataTargetId;

/**
* data target type
*/
private DbTypeEnum dataTargetType;

/**
* target connection params
*/
private String targetConnectionParams;

}
Loading

0 comments on commit 7cdb21d

Please sign in to comment.