forked from apache/inlong
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[INLONG-11005][SDK] Add YAML formatted data source for Transform
- Loading branch information
Showing
6 changed files
with
352 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
18 changes: 18 additions & 0 deletions
18
inlong-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/YamlNode.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| package org.apache.inlong.sdk.transform.decode; | ||
|
|
||
| import lombok.Data; | ||
|
|
||
| @Data | ||
| public class YamlNode { | ||
|
|
||
| private String name; | ||
| private Object value; | ||
|
|
||
| public YamlNode() { | ||
| } | ||
|
|
||
| public YamlNode(String name, Object value) { | ||
| this.name = name; | ||
| this.value = value; | ||
| } | ||
| } |
82 changes: 82 additions & 0 deletions
82
...dk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/YamlSourceData.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| package org.apache.inlong.sdk.transform.decode; | ||
|
|
||
| import java.util.List; | ||
| import java.util.Map; | ||
|
|
||
| public class YamlSourceData implements SourceData{ | ||
|
|
||
| public static final String ROOT_KEY = "$root"; | ||
|
|
||
| public static final String CHILD_KEY = "$child"; | ||
|
|
||
| private YamlNode root; | ||
|
|
||
| private YamlNode childRoot; | ||
|
|
||
| public YamlSourceData(YamlNode root, YamlNode childRoot) { | ||
| this.root = root; | ||
| this.childRoot = childRoot; | ||
| } | ||
| @Override | ||
| public int getRowCount() { | ||
| if (this.childRoot == null) { | ||
| return 1; | ||
| } else { | ||
| Object value = this.childRoot.getValue(); | ||
| if (value instanceof List) { | ||
| return ((List<YamlNode>) value).size(); | ||
| } else { | ||
| return 1; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public String getField(int rowNum, String fieldName) { | ||
| try { | ||
| String[] nodeString = fieldName.split("\\."); | ||
| Object cur = null, last = null; | ||
| int start = -1; | ||
|
|
||
| if (nodeString[0].equals(ROOT_KEY)) { | ||
| cur = root; | ||
| } else if (nodeString[0].equals(CHILD_KEY)) { | ||
| cur = ((List<YamlNode>) childRoot.getValue()).get(rowNum); | ||
| } | ||
|
|
||
| for (int i = 1; i < nodeString.length; i++) { | ||
| if (cur == null) { | ||
| cur = last; | ||
| continue; | ||
| } | ||
| last = cur; | ||
| if (cur instanceof List) { | ||
| int idx = 0; | ||
| start = nodeString[i].indexOf('('); | ||
| if (start != -1) { | ||
| idx = Integer.parseInt(nodeString[1].substring(start + 1, nodeString[1].indexOf(')'))); | ||
| } | ||
| cur = ((List<YamlNode>) cur).get(idx).getValue(); | ||
| } else if (cur instanceof Map) { | ||
| start = nodeString[i].indexOf('('); | ||
| String key = nodeString[i]; | ||
| if (start != -1) { | ||
| key = key.substring(0, start); | ||
| } | ||
| cur = ((Map<String, YamlNode>) cur).get(key); | ||
| } else if (cur instanceof YamlNode) { | ||
| cur = ((YamlNode) cur).getValue(); | ||
| } else { | ||
| i++; | ||
| } | ||
| i--; | ||
| } | ||
| if (cur == null) { | ||
| return ""; | ||
| } | ||
| return cur.toString(); | ||
| } catch (Exception e) { | ||
| return ""; | ||
| } | ||
| } | ||
| } |
111 changes: 111 additions & 0 deletions
111
...transform-sdk/src/main/java/org/apache/inlong/sdk/transform/decode/YamlSourceDecoder.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| package org.apache.inlong.sdk.transform.decode; | ||
|
|
||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.apache.commons.lang3.StringUtils; | ||
| import org.apache.flink.shaded.jackson2.org.yaml.snakeyaml.Yaml; | ||
| import org.apache.inlong.sdk.transform.pojo.YamlSourceInfo; | ||
| import org.apache.inlong.sdk.transform.process.Context; | ||
|
|
||
| import java.nio.charset.Charset; | ||
| import java.util.ArrayList; | ||
| import java.util.Arrays; | ||
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
|
|
||
| @Slf4j | ||
| public class YamlSourceDecoder implements SourceDecoder<String> { | ||
|
|
||
| protected YamlSourceInfo sourceInfo; | ||
| private Charset srcCharset = Charset.defaultCharset(); | ||
| private String rowsNodePath; | ||
| private List<String> childNodes; | ||
|
|
||
| public YamlSourceDecoder(YamlSourceInfo sourceInfo) { | ||
| this.sourceInfo = sourceInfo; | ||
| if (!StringUtils.isBlank(sourceInfo.getCharset())) { | ||
| this.srcCharset = Charset.forName(sourceInfo.getCharset()); | ||
| } | ||
| this.rowsNodePath = sourceInfo.getRowsNodePath(); | ||
| if (!StringUtils.isBlank(rowsNodePath)) { | ||
| this.childNodes = new ArrayList<>(); | ||
| String[] nodeStrings = this.rowsNodePath.split("\\."); | ||
| childNodes.addAll(Arrays.asList(nodeStrings)); | ||
| } | ||
| } | ||
| @Override | ||
| public SourceData decode(byte[] srcBytes, Context context) { | ||
| String srcString = new String(srcBytes, srcCharset); | ||
| return this.decode(srcString, context); | ||
| } | ||
|
|
||
| @Override | ||
| public SourceData decode(String srcString, Context context) { | ||
| try { | ||
| Yaml yaml = new Yaml(); | ||
| Map<String, Object> yamlData = yaml.load(srcString); | ||
| if (yamlData == null || yamlData.isEmpty()) { | ||
| log.error("YAML data is empty or null."); | ||
| return null; | ||
| } | ||
| Map<String, YamlNode> rootMap = new HashMap<>(); | ||
| List<YamlNode> childList = new ArrayList<>(); | ||
|
|
||
| for (Map.Entry<String, Object> entry : yamlData.entrySet()) { | ||
| String key = entry.getKey(); | ||
| Object value = entry.getValue(); | ||
|
|
||
| if (value instanceof Map) { | ||
| // 递归处理子Map，并将其作为根节点的一部分 | ||
| Map<String, YamlNode> childNodes = parser((Map<String, Object>) value); | ||
| rootMap.put(key, new YamlNode(key, childNodes)); | ||
| } else if (value instanceof List) { | ||
| // 处理列表，并将其作为子节点的一部分 | ||
| for (Object item : (List<?>) value) { | ||
| if (item instanceof Map) { | ||
| Map<String, YamlNode> childNodes = parser((Map<String, Object>) item); | ||
| childList.add(new YamlNode(key, childNodes)); | ||
| } else { | ||
| childList.add(new YamlNode(key, item)); | ||
| } | ||
| } | ||
| rootMap.put(key, new YamlNode(key, value)); | ||
| } else { | ||
| // 处理简单类型，将其作为根节点的一部分 | ||
| rootMap.put(key, new YamlNode(key, value)); | ||
| } | ||
| } | ||
| YamlNode root = new YamlNode(YamlSourceData.ROOT_KEY, rootMap.isEmpty() ? null : rootMap); | ||
| YamlNode childRoot = new YamlNode(YamlSourceData.CHILD_KEY, childList.isEmpty() ? null : childList); | ||
| return new YamlSourceData(root,childRoot); | ||
| } catch (Exception e) { | ||
| log.error("Data parsing failed", e); | ||
| return null; | ||
| } | ||
| } | ||
|
|
||
| private static Map<String, YamlNode> parser(Map<String, Object> yamlData) { | ||
| Map<String, YamlNode> yamlNodes = new HashMap<>(); | ||
| for (Map.Entry<String, Object> entry : yamlData.entrySet()) { | ||
| String key = entry.getKey(); | ||
| Object value = entry.getValue(); | ||
|
|
||
| /*if (value instanceof Map) { | ||
| yamlNodes.put(key, new YamlNode(key, parser((Map<String, Object>) value))); | ||
| } else */if (value instanceof List) { | ||
| List<YamlNode> list = new ArrayList<>(); | ||
| for (Object item : (List<?>) value) { | ||
| if (item instanceof Map) { | ||
| list.add(new YamlNode(key, parser((Map<String, Object>) item))); | ||
| } else { | ||
| list.add(new YamlNode(key, item)); | ||
| } | ||
| } | ||
| yamlNodes.put(key, new YamlNode(key, list)); | ||
| } else { | ||
| yamlNodes.put(key, new YamlNode(key, value)); | ||
| } | ||
| } | ||
| return yamlNodes; | ||
| } | ||
| } |
35 changes: 35 additions & 0 deletions
35
...-sdk/transform-sdk/src/main/java/org/apache/inlong/sdk/transform/pojo/YamlSourceInfo.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| package org.apache.inlong.sdk.transform.pojo; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonCreator; | ||
| import com.fasterxml.jackson.annotation.JsonIgnoreProperties; | ||
| import com.fasterxml.jackson.annotation.JsonProperty; | ||
|
|
||
| @JsonIgnoreProperties(ignoreUnknown = true) | ||
| public class YamlSourceInfo extends SourceInfo { | ||
| private String rowsNodePath; | ||
|
|
||
| @JsonCreator | ||
| public YamlSourceInfo( | ||
| @JsonProperty("charset") String charset, | ||
| @JsonProperty("rowsNodePath") String rowsNodePath) { | ||
| super(charset); | ||
| this.rowsNodePath = rowsNodePath; | ||
| } | ||
|
|
||
| /** | ||
| * get rowsNodePath | ||
| * @return the rowsNodePath | ||
| */ | ||
| @JsonProperty("rowsNodePath") | ||
| public String getRowsNodePath() { | ||
| return rowsNodePath; | ||
| } | ||
|
|
||
| /** | ||
| * set rowsNodePath | ||
| * @param rowsNodePath the rowsNodePath to set | ||
| */ | ||
| public void setRowsNodePath(String rowsNodePath) { | ||
| this.rowsNodePath = rowsNodePath; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters