对datax进行了插件扩展-以支持对http接口数据的采集-jsonreader-v1.1.1.zip

package com.alibaba.datax.plugin.reader.jsonreader; import com.alibaba.datax.common.element.Record; import com.alibaba.datax.common.element.StringColumn; import com.alibaba.datax.common.exception.DataXException; import com.alibaba.datax.common.plugin.RecordSender; import com.alibaba.datax.common.spi.Reader; import com.alibaba.datax.common.util.Configuration; import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode; import org.apache.http.client.fluent.Request; import org.apache.http.client.fluent.Response; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import java.io.IOException; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.stream.Collectors; public class JsonReader extends Reader { public static class Job extends Reader.Job { private Configuration configuration; @Override public void init() { this.configuration = getPluginJobConf(); } @Override public void prepare() { } @Override public void post() { } @Override public void destroy() { } @Override public List<Configuration> split(int adviceNumber) { // 分割任务，返回多个配置片段 List<Configuration> splitConfigurations = new ArrayList<>(); for (int i = 0; i < adviceNumber; i++) { splitConfigurations.add(this.configuration.clone()); } return splitConfigurations; } } public static class Task extends Reader.Task { private Configuration configuration; private ObjectMapper objectMapper; @Override public void init() { this.configuration = getPluginJobConf(); this.objectMapper = new ObjectMapper(); } @Override public void prepare() { } @Override public void startRead(RecordSender recordSender) { String url = configuration.getString("url"); String method = configuration.getString("method", "GET"); Map<String, Object> headers = configuration.getMap("headers", new HashMap<>()); Map<String, Object> params = configuration.getMap("params", new HashMap<>()); int startLevel = configuration.getInt("startLevel", 1); String listKey = configuration.getString("listKey", ""); List<String> columns = configuration.getList("column", String.class); // // 获取字段映射配置 暂时不用 // List<Map> rawColumns = configuration.getList("column", Map.class); // // // 手动转换为 List<Map<String, String>> 暂时不用 // List<Map<String, String>> columns = new ArrayList<>(); // for (Map rawColumn : rawColumns) { // columns.add((Map<String, String>) rawColumn); // } try { String fullUrl = buildFullUrl(url, params); if (fullUrl == null) { throw new DataXException("url拼接params参数后为空，请检查url和params是否为空"); } String jsonResponse = executeRequest(fullUrl, method, headers); List<Map<String, Object>> parsedData = parseJson(jsonResponse, startLevel, listKey); if (parsedData == null) { throw new DataXException("经解析过的json数据为空，或者层级、层级名称填写错误"); } for (Map<String, Object> data : parsedData) { Record record = recordSender.createRecord(); for (String column : columns) { // String jsonKey = column.get("jsonKey"); //这里暂时改一下，配置文件改成数组 // String columnName = column.get("columnName"); //这是是本想匹配写入字段的，按照顺序就行了 Object value = data.get(column); if (value != null) { record.addColumn(new StringColumn(value.toString())); } else { record.addColumn(new StringColumn(null)); // 插入null值 } } recordSender.sendToWriter(record); } } catch (Exception e) { throw DataXException.asDataXException(DBUtilErrorCode.CONF_ERROR, String.format("JsonReader失败原因：", e)); } } @Override public void post() { } @Override public void destroy() { } private String buildFullUrl(String baseUrl, Map<String, Object> params) { StringBuilder fullUrl = new StringBuilder(baseUrl); if (!params.isEmpty()) { fullUrl.append("?"); params.forEach((key, value) -> fullUrl.append(key).append("=").append(value).append("&")); fullUrl.deleteCharAt(fullUrl.length() - 1); } return fullUrl.toString(); } private String executeRequest(String url, String method, Map<String, Object> headers) throws Exception { Request request = method.equalsIgnoreCase("POST") ? Request.Post(url) : Request.Get(url); headers.forEach((key, value) -> request.addHeader(key, value.toString())); Response response = request.execute(); // 强制使用UTF-8解析响应内容 String content = response.returnContent().asString(java.nio.charset.StandardCharsets.UTF_8); return content; } private List<Map<String, Object>> parseJson(String jsonStr, int startLevel, String listKey) { Object root = parseJsonStr(jsonStr); List<Map<String, Object>> resultList = new ArrayList<>(); processObject(root, resultList, startLevel, 1, listKey); return resultList; } private Object parseJsonStr(String jsonString) { try { return objectMapper.readValue(jsonString.replaceAll("\\r\\n|\\n", ""), new TypeReference<Map<String, Object>>() {}); } catch (IOException e) { try { return objectMapper.readValue(jsonString.replaceAll("\\r\\n|\\n", ""), new TypeReference<List<Object>>() {}); } catch (IOException ex) { return null; } } } private void processObject(Object obj, List<Map<String, Object>> resultList, int startLevel, int currentLevel, String listKey) { if (obj instanceof List) { processList((List<?>) obj, resultList, startLevel, currentLevel, listKey); } else if (obj instanceof Map) { processMap((Map<?, ?>) obj, resultList, startLevel, currentLevel, listKey); } } private void processList(List<?> list, List<Map<String, Object>> resultList, int startLevel, int currentLevel, String listKey) { for (Object element : list) { if (element instanceof Map) { processMap((Map<?, ?>) element, resultList, startLevel, currentLevel, listKey); } } } private void processMap(Map<?, ?> map, List<Map<String, Object>> resultList, int startLevel, int currentLevel, String listKey) { if (currentLevel >= startLevel) { Map<String, Object> tempMap = new HashMap<>(); for (Map.Entry<?, ?> entry : map.entrySet()) { Object key = ent