|
| 1 | +package com.webank.wedatasphere.exchangis.job.server.builder.transform.handlers; |
| 2 | + |
| 3 | +import com.webank.wedatasphere.exchangis.job.builder.ExchangisJobBuilderContext; |
| 4 | +import com.webank.wedatasphere.exchangis.job.domain.SubExchangisJob; |
| 5 | +import com.webank.wedatasphere.exchangis.job.domain.params.JobParamDefine; |
| 6 | +import com.webank.wedatasphere.exchangis.job.domain.params.JobParamSet; |
| 7 | +import com.webank.wedatasphere.exchangis.job.domain.params.JobParams; |
| 8 | +import com.webank.wedatasphere.exchangis.job.server.builder.JobParamConstraints; |
| 9 | +import org.apache.linkis.common.exception.ErrorException; |
| 10 | +import org.slf4j.Logger; |
| 11 | +import org.slf4j.LoggerFactory; |
| 12 | + |
| 13 | +import java.util.HashMap; |
| 14 | +import java.util.Map; |
| 15 | +import java.util.Objects; |
| 16 | + |
| 17 | +/** |
| 18 | + * File source handler for the datax engine (DataX 引擎的文件 source 处理器) |
| 19 | + * |
| 20 | + * <p>Registration: auto-discovered via reflection by {@code GenericExchangisTransformJobBuilder.initHandlers()} |
| 21 | + * (keyed by {@link #dataSourceType()} = "file"), no manual registration needed. |
| 22 | + * |
| 23 | + * <p>Architecture ② / M5: the file source does NOT go through datasource management. This handler |
| 24 | + * does NOT call {@code DataSourceService}; it only extracts the BML reference from the source |
| 25 | + * params and stashes it for the engine builder to materialize as an {@code EngineBmlResource}. |
| 26 | + * |
| 27 | + * <p>架构②/M5：文件 source 不走数据源管理流程。本 handler 不调用 DataSourceService， |
| 28 | + * 仅从 source 参数中提取 BML 引用并暂存，供引擎 builder 物化为 EngineBmlResource。 |
| 29 | + * |
| 30 | + * <h2>Param convention (参数约定)</h2> |
| 31 | + * <ul> |
| 32 | + * <li>BML reference params are namespaced with the {@code __file_bml_} prefix |
| 33 | + * (e.g. {@code __file_bml_resource_id}). They are read with |
| 34 | + * {@link JobParams#define(String)} and stay in the source param set; txtfilereader ignores |
| 35 | + * these unknown keys, so no manual get+remove is needed.</li> |
| 36 | + * <li>txtfilereader params ({@code path}/{@code encoding}/{@code delimiter}/{@code nullFormat}) |
| 37 | + * are real reader keys. {@code path} is set to the BML file name (the file the EC downloads |
| 38 | + * to its workdir); the others are sent by the frontend from the parse result.</li> |
| 39 | + * </ul> |
| 40 | + * |
| 41 | + * <p>BML 引用参数以 {@code __file_bml_} 前缀命名空间隔离，用 {@link JobParams#define(String)} 读取， |
| 42 | + * 保留在 source 参数集中（txtfilereader 忽略未知键，无需 get+remove）。txtfilereader 参数 |
| 43 | + * （path/encoding/delimiter/nullFormat）为真实 reader 键，path 设为 BML 文件名。 |
| 44 | + * |
| 45 | + * <h2>BML injection adaptation (BML 注入适配)</h2> |
| 46 | + * The design doc describes writing a plain-JSON {@code wds.linkis.engineconn.datax.bml.resources} job |
| 47 | + * property in the handler. This open-source codebase has no {@code getJobProps()} on |
| 48 | + * {@link SubExchangisJob}, but it ALREADY serializes {@code engineJob.getResources()} into |
| 49 | + * {@code wds.linkis.engineconn.datax.bml.resources} at launch time |
| 50 | + * (see {@code ExchangisLauncherConfiguration.LAUNCHER_LINKIS_RESOURCES} + |
| 51 | + * {@code DataxEngineConnLaunchBuilder.getBmlResources}). So the handler stashes the BML ref into |
| 52 | + * {@link SubExchangisJob#getJobParams()} and {@code DataxExchangisEngineJobBuilder} materializes it |
| 53 | + * as an {@code EngineBmlResource} — the same pattern as the existing {@code settingProcessorInfo}. |
| 54 | + * |
| 55 | + * <p>设计文档描述在 handler 中写入明文 JSON 作业属性。本仓库的 {@link SubExchangisJob} 无 |
| 56 | + * {@code getJobProps()}，但已有机制：launcher 把 {@code engineJob.getResources()} 序列化为 |
| 57 | + * {@code wds.linkis.engineconn.datax.bml.resources}。故 handler 将 BML 引用暂存到 jobParams， |
| 58 | + * 由 {@code DataxExchangisEngineJobBuilder} 物化为 {@code EngineBmlResource}。 |
| 59 | + */ |
| 60 | +public class FileDataxSubExchangisJobHandler extends AuthEnabledSubExchangisJobHandler { |
| 61 | + |
| 62 | + private static final Logger LOG = LoggerFactory.getLogger(FileDataxSubExchangisJobHandler.class); |
| 63 | + |
| 64 | + /** |
| 65 | + * Job param key used to stash the file BML reference for the engine builder. |
| 66 | + * 暂存文件 BML 引用供引擎 builder 使用的 job param key。 |
| 67 | + */ |
| 68 | + public static final String FILE_BML_RESOURCE_KEY = "__file_bml_resource"; |
| 69 | + |
| 70 | + /** |
| 71 | + * BML reference source params (namespaced with __file_bml_ prefix). |
| 72 | + * BML 引用 source 参数（以 __file_bml_ 前缀命名空间隔离）。 |
| 73 | + */ |
| 74 | + public static final String PARAM_BML_RESOURCE_ID = "__file_bml_resource_id"; |
| 75 | + public static final String PARAM_BML_VERSION = "__file_bml_version"; |
| 76 | + public static final String PARAM_BML_OWNER = "__file_bml_owner"; |
| 77 | + public static final String PARAM_BML_NAME = "__file_bml_name"; |
| 78 | + |
| 79 | + private static final JobParamDefine<String> BML_RESOURCE_ID = JobParams.define(PARAM_BML_RESOURCE_ID); |
| 80 | + private static final JobParamDefine<String> BML_VERSION = JobParams.define(PARAM_BML_VERSION); |
| 81 | + private static final JobParamDefine<String> BML_OWNER = JobParams.define(PARAM_BML_OWNER); |
| 82 | + private static final JobParamDefine<String> BML_NAME = JobParams.define(PARAM_BML_NAME); |
| 83 | + |
| 84 | + /** |
| 85 | + * txtfilereader params (real reader keys) / txtfilereader 参数（真实 reader 键） |
| 86 | + */ |
| 87 | + private static final String PARAM_PATH = "path"; |
| 88 | + private static final String PARAM_DELIMITER = "delimiter"; |
| 89 | + private static final JobParamDefine<String> ENCODING = JobParams.define(JobParamConstraints.ENCODING); |
| 90 | + private static final JobParamDefine<String> DELIMITER = JobParams.define(PARAM_DELIMITER); |
| 91 | + private static final JobParamDefine<String> NULL_FORMAT = JobParams.define(JobParamConstraints.NULL_FORMAT); |
| 92 | + |
| 93 | + @Override |
| 94 | + public void handleJobSource(SubExchangisJob subExchangisJob, ExchangisJobBuilderContext ctx) throws ErrorException { |
| 95 | + JobParamSet paramSet = subExchangisJob.getRealmParams(SubExchangisJob.REALM_JOB_CONTENT_SOURCE); |
| 96 | + if (Objects.isNull(paramSet)) { |
| 97 | + return; |
| 98 | + } |
| 99 | + |
| 100 | + // 1. Read the BML reference from the namespaced __file_bml_ source params. |
| 101 | + // They stay in the source param set (txtfilereader ignores the __file_bml_ keys). |
| 102 | + // 读取 BML 引用（__file_bml_ 前缀命名空间隔离，保留在 source 参数集中，txtfilereader 忽略） |
| 103 | + String resourceId = BML_RESOURCE_ID.getValue(paramSet); |
| 104 | + String version = BML_VERSION.getValue(paramSet); |
| 105 | + String owner = BML_OWNER.getValue(paramSet); |
| 106 | + String name = BML_NAME.getValue(paramSet); |
| 107 | + |
| 108 | + if (Objects.nonNull(resourceId) && Objects.nonNull(version)) { |
| 109 | + // 2. Stash into jobParams for the engine builder (暂存到 jobParams 供引擎 builder 使用) |
| 110 | + Map<String, String> bmlRef = new HashMap<>(); |
| 111 | + bmlRef.put(PARAM_BML_RESOURCE_ID, resourceId); |
| 112 | + bmlRef.put(PARAM_BML_VERSION, version); |
| 113 | + bmlRef.put(PARAM_BML_OWNER, owner); |
| 114 | + bmlRef.put(PARAM_BML_NAME, name); |
| 115 | + subExchangisJob.getJobParams().put(FILE_BML_RESOURCE_KEY, bmlRef); |
| 116 | + LOG.info("File source BML reference stashed (文件 source BML 引用已暂存): name={}, resourceId={}, version={}", |
| 117 | + name, resourceId, version); |
| 118 | + |
| 119 | + // 3. path = BML file name. The EC downloads the BML resource to its workdir (PWD) using |
| 120 | + // this name, and txtfilereader reads the file by this path. |
| 121 | + // path = BML 文件名。EC 将 BML 资源下载到工作目录（PWD）后，txtfilereader 按此 path 读取。 |
| 122 | + if (Objects.nonNull(name)) { |
| 123 | + paramSet.add(JobParams.newOne(PARAM_PATH, name)); |
| 124 | + } |
| 125 | + } else { |
| 126 | + LOG.warn("File source handler: missing resourceId/version in source params (source 参数缺失 resourceId/version)"); |
| 127 | + } |
| 128 | + |
| 129 | + // 4. Ensure the txtfilereader params are present in the output param set. |
| 130 | + // 确保 txtfilereader 参数（encoding/delimiter/nullFormat）存在于输出参数集。 |
| 131 | + paramSet.addNonNull(ENCODING.get(paramSet)); |
| 132 | + paramSet.addNonNull(DELIMITER.get(paramSet)); |
| 133 | + paramSet.addNonNull(NULL_FORMAT.get(paramSet)); |
| 134 | + } |
| 135 | + |
| 136 | + @Override |
| 137 | + public void handleJobSink(SubExchangisJob subExchangisJob, ExchangisJobBuilderContext ctx) throws ErrorException { |
| 138 | + // FILE is source-only; job validation rejects sink=file. (FILE 仅作 source，不实现 sink) |
| 139 | + } |
| 140 | + |
| 141 | + @Override |
| 142 | + public String dataSourceType() { |
| 143 | + return "file"; |
| 144 | + } |
| 145 | + |
| 146 | + @Override |
| 147 | + public boolean acceptEngine(String engineType) { |
| 148 | + return "datax".equalsIgnoreCase(engineType); |
| 149 | + } |
| 150 | +} |
0 commit comments