configurationList = config.getListConfiguration(Key.COLUMN);
+ for (Configuration column : configurationList) {
+ ColumnType columnType;
+ try {
+ columnType = ColumnType.valueOf(column.getString(Key.COLUMN_NODE_TYPE));
+ } catch (NullPointerException | IllegalArgumentException e) {
+ throw DataXException.asDataXException(GdbReaderErrorCode.BAD_CONFIG_VALUE, Key.COLUMN_NODE_TYPE);
+ }
+
+ if (exportType == ExportType.VERTEX) {
+ // only id/label/property column allow when vertex
+ ConfigHelper.assertConfig(Key.COLUMN_NODE_TYPE, () ->
+ columnType == ColumnType.primaryKey || columnType == ColumnType.primaryLabel
+ || columnType == ColumnType.vertexProperty || columnType == ColumnType.vertexJsonProperty);
+ } else if (exportType == ExportType.EDGE) {
+ // edge
+ ConfigHelper.assertConfig(Key.COLUMN_NODE_TYPE, () ->
+ columnType == ColumnType.primaryKey || columnType == ColumnType.primaryLabel
+ || columnType == ColumnType.srcPrimaryKey || columnType == ColumnType.srcPrimaryLabel
+ || columnType == ColumnType.dstPrimaryKey || columnType == ColumnType.dstPrimaryLabel
+ || columnType == ColumnType.edgeProperty || columnType == ColumnType.edgeJsonProperty);
+ }
+
+ if (columnType == ColumnType.edgeProperty || columnType == ColumnType.vertexProperty) {
+ String name = column.getString(Key.COLUMN_NAME);
+ ValueType propType = ValueType.fromShortName(column.getString(Key.COLUMN_TYPE));
+
+ ConfigHelper.assertConfig(Key.COLUMN_NAME, () -> name != null);
+ if (propType == null) {
+ throw DataXException.asDataXException(GdbReaderErrorCode.UNSUPPORTED_TYPE, Key.COLUMN_TYPE);
+ }
+ rule.addColumn(columnType, propType, name);
+ } else if (columnType == ColumnType.vertexJsonProperty || columnType == ColumnType.edgeJsonProperty) {
+ rule.addJsonColumn(columnType);
+ } else {
+ rule.addColumn(columnType, ValueType.STRING, null);
+ }
+ }
+ return rule;
+ }
+}
diff --git a/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/mapping/ValueType.java b/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/mapping/ValueType.java
new file mode 100644
index 00000000..826e0493
--- /dev/null
+++ b/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/mapping/ValueType.java
@@ -0,0 +1,128 @@
+/*
+ * (C) 2019-present Alibaba Group Holding Limited.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 as
+ * published by the Free Software Foundation.
+ */
+package com.alibaba.datax.plugin.reader.gdbreader.mapping;
+
+import com.alibaba.datax.common.element.BoolColumn;
+import com.alibaba.datax.common.element.Column;
+import com.alibaba.datax.common.element.DoubleColumn;
+import com.alibaba.datax.common.element.LongColumn;
+import com.alibaba.datax.common.element.StringColumn;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Function;
+
+/**
+ * @author : Liu Jianping
+ * @date : 2019/9/6
+ */
+
+public enum ValueType {
+ /**
+ * transfer gdb element object value to DataX Column data
+ *
+ * int, long -> LongColumn
+ * float, double -> DoubleColumn
+ * bool -> BooleanColumn
+ * string -> StringColumn
+ */
+ INT(Integer.class, "int", ValueTypeHolder::longColumnMapper),
+ INTEGER(Integer.class, "integer", ValueTypeHolder::longColumnMapper),
+ LONG(Long.class, "long", ValueTypeHolder::longColumnMapper),
+ DOUBLE(Double.class, "double", ValueTypeHolder::doubleColumnMapper),
+ FLOAT(Float.class, "float", ValueTypeHolder::doubleColumnMapper),
+ BOOLEAN(Boolean.class, "boolean", ValueTypeHolder::boolColumnMapper),
+ STRING(String.class, "string", ValueTypeHolder::stringColumnMapper),
+ ;
+
+ private Class type = null;
+ private String shortName = null;
+ private Function columnFunc = null;
+
+ ValueType(Class type, String name, Function columnFunc) {
+ this.type = type;
+ this.shortName = name;
+ this.columnFunc = columnFunc;
+
+ ValueTypeHolder.shortName2type.put(shortName, this);
+ }
+
+ public static ValueType fromShortName(String name) {
+ return ValueTypeHolder.shortName2type.get(name);
+ }
+
+ public Column applyObject(Object value) {
+ if (value == null) {
+ return null;
+ }
+ return columnFunc.apply(value);
+ }
+
+ private static class ValueTypeHolder {
+ private static Map shortName2type = new HashMap<>();
+
+ private static LongColumn longColumnMapper(Object o) {
+ long v;
+ if (o instanceof Integer) {
+ v = (int) o;
+ } else if (o instanceof Long) {
+ v = (long) o;
+ } else if (o instanceof String) {
+ v = Long.valueOf((String) o);
+ } else {
+ throw new RuntimeException("Failed to cast " + o.getClass() + " to Long");
+ }
+
+ return new LongColumn(v);
+ }
+
+ private static DoubleColumn doubleColumnMapper(Object o) {
+ double v;
+ if (o instanceof Integer) {
+ v = (double) (int) o;
+ } else if (o instanceof Long) {
+ v = (double) (long) o;
+ } else if (o instanceof Float) {
+ v = (double) (float) o;
+ } else if (o instanceof Double) {
+ v = (double) o;
+ } else if (o instanceof String) {
+ v = Double.valueOf((String) o);
+ } else {
+ throw new RuntimeException("Failed to cast " + o.getClass() + " to Double");
+ }
+
+ return new DoubleColumn(v);
+ }
+
+ private static BoolColumn boolColumnMapper(Object o) {
+ boolean v;
+ if (o instanceof Integer) {
+ v = ((int) o != 0);
+ } else if (o instanceof Long) {
+ v = ((long) o != 0);
+ } else if (o instanceof Boolean) {
+ v = (boolean) o;
+ } else if (o instanceof String) {
+ v = Boolean.valueOf((String) o);
+ } else {
+ throw new RuntimeException("Failed to cast " + o.getClass() + " to Boolean");
+ }
+
+ return new BoolColumn(v);
+ }
+
+ private static StringColumn stringColumnMapper(Object o) {
+ if (o instanceof String) {
+ return new StringColumn((String) o);
+ } else {
+ return new StringColumn(String.valueOf(o));
+ }
+ }
+ }
+}
diff --git a/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/model/AbstractGdbGraph.java b/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/model/AbstractGdbGraph.java
new file mode 100644
index 00000000..4eda2eed
--- /dev/null
+++ b/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/model/AbstractGdbGraph.java
@@ -0,0 +1,89 @@
+/*
+ * (C) 2019-present Alibaba Group Holding Limited.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 as
+ * published by the Free Software Foundation.
+ */
+package com.alibaba.datax.plugin.reader.gdbreader.model;
+
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.reader.gdbreader.Key;
+import org.apache.tinkerpop.gremlin.driver.Client;
+import org.apache.tinkerpop.gremlin.driver.Cluster;
+import org.apache.tinkerpop.gremlin.driver.RequestOptions;
+import org.apache.tinkerpop.gremlin.driver.Result;
+import org.apache.tinkerpop.gremlin.driver.ResultSet;
+import org.apache.tinkerpop.gremlin.driver.ser.Serializers;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author : Liu Jianping
+ * @date : 2019/9/6
+ */
+
+public abstract class AbstractGdbGraph implements GdbGraph {
+ final static int DEFAULT_TIMEOUT = 30000;
+ private static final Logger log = LoggerFactory.getLogger(AbstractGdbGraph.class);
+ private Client client;
+
+ AbstractGdbGraph() {
+ }
+
+ AbstractGdbGraph(Configuration config) {
+ log.info("init graphdb client");
+ String host = config.getString(Key.HOST);
+ int port = config.getInt(Key.PORT);
+ String username = config.getString(Key.USERNAME);
+ String password = config.getString(Key.PASSWORD);
+
+ try {
+ Cluster cluster = Cluster.build(host).port(port).credentials(username, password)
+ .serializer(Serializers.GRAPHBINARY_V1D0)
+ .maxContentLength(1024 * 1024)
+ .resultIterationBatchSize(64)
+ .create();
+ client = cluster.connect().init();
+
+ warmClient();
+ } catch (RuntimeException e) {
+ log.error("Failed to connect to GDB {}:{}, due to {}", host, port, e);
+ throw e;
+ }
+ }
+
+ protected List runInternal(String dsl, Map params) throws Exception {
+ return runInternalAsync(dsl, params).all().get(DEFAULT_TIMEOUT + 1000, TimeUnit.MILLISECONDS);
+ }
+
+ protected ResultSet runInternalAsync(String dsl, Map params) throws Exception {
+ RequestOptions.Builder options = RequestOptions.build().timeout(DEFAULT_TIMEOUT);
+ if (params != null && !params.isEmpty()) {
+ params.forEach(options::addParameter);
+ }
+ return client.submitAsync(dsl, options.create()).get(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
+ }
+
+ private void warmClient() {
+ try {
+ runInternal("g.V('test')", null);
+ log.info("warm graphdb client over");
+ } catch (Exception e) {
+ log.error("warmClient error");
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (client != null) {
+ log.info("close graphdb client");
+ client.close();
+ }
+ }
+}
diff --git a/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/model/GdbElement.java b/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/model/GdbElement.java
new file mode 100644
index 00000000..79619ad0
--- /dev/null
+++ b/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/model/GdbElement.java
@@ -0,0 +1,39 @@
+/*
+ * (C) 2019-present Alibaba Group Holding Limited.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 as
+ * published by the Free Software Foundation.
+ */
+package com.alibaba.datax.plugin.reader.gdbreader.model;
+
+import lombok.Data;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author : Liu Jianping
+ * @date : 2019/9/6
+ */
+
+@Data
+public class GdbElement {
+ String id = null;
+ String label = null;
+ String to = null;
+ String from = null;
+ String toLabel = null;
+ String fromLabel = null;
+
+ Map properties = new HashMap<>();
+
+ public GdbElement() {
+ }
+
+ public GdbElement(String id, String label) {
+ this.id = id;
+ this.label = label;
+ }
+
+}
diff --git a/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/model/GdbGraph.java b/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/model/GdbGraph.java
new file mode 100644
index 00000000..e6651293
--- /dev/null
+++ b/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/model/GdbGraph.java
@@ -0,0 +1,65 @@
+/*
+ * (C) 2019-present Alibaba Group Holding Limited.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 as
+ * published by the Free Software Foundation.
+ */
+package com.alibaba.datax.plugin.reader.gdbreader.model;
+
+import org.apache.tinkerpop.gremlin.driver.ResultSet;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author : Liu Jianping
+ * @date : 2019/9/6
+ */
+
+public interface GdbGraph extends AutoCloseable {
+
+ /**
+ * Get All labels of GraphDB
+ *
+ * @return labels map included numbers
+ */
+ Map getLabels();
+
+ /**
+ * Get the Ids list of special 'label', size up to 'limit'
+ *
+ * @param label is Label of Vertex or Edge
+ * @param start of Ids range to get
+ * @param limit size of Ids list
+ * @return Ids list
+ */
+ List fetchIds(String label, String start, long limit);
+
+ /**
+ * Fetch element in async mode, just send query dsl to server
+ *
+ * @param label node label to filter
+ * @param start range begin(included)
+ * @param end range end(included)
+ * @param propNames propKey list to fetch
+ * @return future to get result later
+ */
+ ResultSet fetchElementsAsync(String label, String start, String end, List propNames);
+
+ /**
+ * Get get element from Response @{ResultSet}
+ *
+ * @param results Response of Server
+ * @return element sets
+ */
+ List getElement(ResultSet results);
+
+ /**
+ * close graph client
+ *
+ * @throws Exception if fails
+ */
+ @Override
+ void close() throws Exception;
+}
diff --git a/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/model/ScriptGdbGraph.java b/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/model/ScriptGdbGraph.java
new file mode 100644
index 00000000..8c08b819
--- /dev/null
+++ b/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/model/ScriptGdbGraph.java
@@ -0,0 +1,192 @@
+/*
+ * (C) 2019-present Alibaba Group Holding Limited.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 as
+ * published by the Free Software Foundation.
+ */
+package com.alibaba.datax.plugin.reader.gdbreader.model;
+
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.reader.gdbreader.Key.ExportType;
+import org.apache.tinkerpop.gremlin.driver.Result;
+import org.apache.tinkerpop.gremlin.driver.ResultSet;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceEdge;
+import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceVertex;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author : Liu Jianping
+ * @date : 2019/9/6
+ */
+
+public class ScriptGdbGraph extends AbstractGdbGraph {
+ private static final Logger log = LoggerFactory.getLogger(ScriptGdbGraph.class);
+
+ private final static String LABEL = "GDB___LABEL";
+ private final static String START_ID = "GDB___ID";
+ private final static String END_ID = "GDB___ID_END";
+ private final static String LIMIT = "GDB___LIMIT";
+
+ private final static String FETCH_VERTEX_IDS_DSL = "g.V().hasLabel(" + LABEL + ").has(id, gt(" + START_ID + ")).limit(" + LIMIT + ").id()";
+ private final static String FETCH_EDGE_IDS_DSL = "g.E().hasLabel(" + LABEL + ").has(id, gt(" + START_ID + ")).limit(" + LIMIT + ").id()";
+
+ private final static String FETCH_VERTEX_LABELS_DSL = "g.V().groupCount().by(label)";
+ private final static String FETCH_EDGE_LABELS_DSL = "g.E().groupCount().by(label)";
+
+ /**
+ * fetch node range [START_ID, END_ID]
+ */
+ private final static String FETCH_RANGE_VERTEX_DSL = "g.V().hasLabel(" + LABEL + ").has(id, gte(" + START_ID + ")).has(id, lte(" + END_ID + "))";
+ private final static String FETCH_RANGE_EDGE_DSL = "g.E().hasLabel(" + LABEL + ").has(id, gte(" + START_ID + ")).has(id, lte(" + END_ID + "))";
+ private final static String PART_WITH_PROP_DSL = ".as('a').project('node', 'props').by(select('a')).by(select('a').propertyMap(";
+
+ private final ExportType exportType;
+
+ public ScriptGdbGraph(ExportType exportType) {
+ super();
+ this.exportType = exportType;
+ }
+
+ public ScriptGdbGraph(Configuration config, ExportType exportType) {
+ super(config);
+ this.exportType = exportType;
+ }
+
+ @Override
+ public List fetchIds(final String label, final String start, long limit) {
+ Map params = new HashMap(3) {{
+ put(LABEL, label);
+ put(START_ID, start);
+ put(LIMIT, limit);
+ }};
+ String fetchDsl = exportType == ExportType.VERTEX ? FETCH_VERTEX_IDS_DSL : FETCH_EDGE_IDS_DSL;
+
+ List ids = new ArrayList<>();
+ try {
+ List results = runInternal(fetchDsl, params);
+
+ // transfer result to id string
+ results.forEach(id -> ids.add(id.getString()));
+ } catch (Exception e) {
+ log.error("fetch range node failed, label {}, start {}", label, start);
+ throw new RuntimeException(e);
+ }
+ return ids;
+ }
+
+ @Override
+ public ResultSet fetchElementsAsync(final String label, final String start, final String end, final List propNames) {
+ Map params = new HashMap<>(3);
+ params.put(LABEL, label);
+ params.put(START_ID, start);
+ params.put(END_ID, end);
+
+ String prefixDsl = exportType == ExportType.VERTEX ? FETCH_RANGE_VERTEX_DSL : FETCH_RANGE_EDGE_DSL;
+ StringBuilder fetchDsl = new StringBuilder(prefixDsl);
+ if (propNames != null) {
+ fetchDsl.append(PART_WITH_PROP_DSL);
+ for (int i = 0; i < propNames.size(); i++) {
+ String propName = "GDB___PK" + String.valueOf(i);
+ params.put(propName, propNames.get(i));
+
+ fetchDsl.append(propName);
+ if (i != propNames.size() - 1) {
+ fetchDsl.append(", ");
+ }
+ }
+ fetchDsl.append("))");
+ }
+
+ try {
+ return runInternalAsync(fetchDsl.toString(), params);
+ } catch (Exception e) {
+ log.error("Failed to fetch range node startId {}, end {} , e {}", start, end, e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public List getElement(ResultSet results) {
+ List elements = new LinkedList<>();
+ try {
+ List resultList = results.all().get(DEFAULT_TIMEOUT + 1000, TimeUnit.MILLISECONDS);
+
+ resultList.forEach(n -> {
+ Object o = n.getObject();
+ GdbElement element = new GdbElement();
+ if (o instanceof Map) {
+ // project response
+ Object node = ((Map) o).get("node");
+ Object props = ((Map) o).get("props");
+
+ mapNodeToElement(node, element);
+ mapPropToElement((Map) props, element);
+ } else {
+ // range node response
+ mapNodeToElement(n.getObject(), element);
+ }
+ if (element.getId() != null) {
+ elements.add(element);
+ }
+ });
+ } catch (Exception e) {
+ log.error("Failed to get node: {}", e);
+ throw new RuntimeException(e);
+ }
+ return elements;
+ }
+
+ private void mapNodeToElement(Object node, GdbElement element) {
+ if (node instanceof ReferenceVertex) {
+ ReferenceVertex v = (ReferenceVertex) node;
+
+ element.setId((String) v.id());
+ element.setLabel(v.label());
+ } else if (node instanceof ReferenceEdge) {
+ ReferenceEdge e = (ReferenceEdge) node;
+
+ element.setId((String) e.id());
+ element.setLabel(e.label());
+ element.setTo((String) e.inVertex().id());
+ element.setToLabel(e.inVertex().label());
+ element.setFrom((String) e.outVertex().id());
+ element.setFromLabel(e.outVertex().label());
+ }
+ }
+
+ private void mapPropToElement(Map props, GdbElement element) {
+ element.setProperties(props);
+ }
+
+ @Override
+ public Map getLabels() {
+ String dsl = exportType == ExportType.VERTEX ? FETCH_VERTEX_LABELS_DSL : FETCH_EDGE_LABELS_DSL;
+
+ try {
+ List results = runInternal(dsl, null);
+ Map labelMap = new HashMap<>(2);
+
+ Map labels = results.get(0).get(Map.class);
+ labels.forEach((k, v) -> {
+ String label = (String) k;
+ Long count = (Long) v;
+ labelMap.put(label, count);
+ });
+
+ return labelMap;
+ } catch (Exception e) {
+ log.error("Failed to fetch label list, please give special labels and run again, e {}", e);
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git a/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/util/ConfigHelper.java b/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/util/ConfigHelper.java
new file mode 100644
index 00000000..2ec9d153
--- /dev/null
+++ b/gdbreader/src/main/java/com/alibaba/datax/plugin/reader/gdbreader/util/ConfigHelper.java
@@ -0,0 +1,77 @@
+/*
+ * (C) 2019-present Alibaba Group Holding Limited.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 as
+ * published by the Free Software Foundation.
+ */
+package com.alibaba.datax.plugin.reader.gdbreader.util;
+
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.reader.gdbreader.GdbReaderErrorCode;
+import com.alibaba.datax.plugin.reader.gdbreader.Key;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Supplier;
+
+/**
+ * @author : Liu Jianping
+ * @date : 2019/9/6
+ */
+
+public interface ConfigHelper {
+ static void assertConfig(String key, Supplier f) {
+ if (!f.get()) {
+ throw DataXException.asDataXException(GdbReaderErrorCode.BAD_CONFIG_VALUE, key);
+ }
+ }
+
+ static void assertHasContent(Configuration config, String key) {
+ assertConfig(key, () -> StringUtils.isNotBlank(config.getString(key)));
+ }
+
+ static void assertGdbClient(Configuration config) {
+ assertHasContent(config, Key.HOST);
+ assertConfig(Key.PORT, () -> config.getInt(Key.PORT) > 0);
+
+ assertHasContent(config, Key.USERNAME);
+ assertHasContent(config, Key.PASSWORD);
+ }
+
+ static List assertLabels(Configuration config) {
+ Object labels = config.get(Key.LABEL);
+ if (!(labels instanceof List)) {
+ throw DataXException.asDataXException(GdbReaderErrorCode.BAD_CONFIG_VALUE, "labels should be List");
+ }
+
+ List list = (List) labels;
+ List configLabels = new ArrayList<>(0);
+ list.forEach(n -> configLabels.add(String.valueOf(n)));
+
+ return configLabels;
+ }
+
+ static List splitConfig(Configuration config, List labels) {
+ List configs = new ArrayList<>();
+ for (String label : labels) {
+ Configuration conf = config.clone();
+ conf.set(Key.LABEL, label);
+
+ configs.add(conf);
+ }
+ return configs;
+ }
+
+ static Configuration fromClasspath(String name) {
+ try (InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream(name)) {
+ return Configuration.from(is);
+ } catch (IOException e) {
+ throw new IllegalArgumentException("File not found: " + name);
+ }
+ }
+}
diff --git a/gdbreader/src/main/resources/plugin.json b/gdbreader/src/main/resources/plugin.json
new file mode 100644
index 00000000..4fedfa24
--- /dev/null
+++ b/gdbreader/src/main/resources/plugin.json
@@ -0,0 +1,6 @@
+{
+ "name": "gdbreader",
+ "class": "com.alibaba.datax.plugin.reader.gdbreader.GdbReader",
+ "description": "useScene: prod. mechanism: connect GDB with gremlin-client, execute 'g.V().propertyMap() or g.E().propertyMap()' to get record",
+ "developer": "alibaba"
+}
\ No newline at end of file
diff --git a/gdbreader/src/main/resources/plugin_job_template.json b/gdbreader/src/main/resources/plugin_job_template.json
new file mode 100644
index 00000000..bfca7780
--- /dev/null
+++ b/gdbreader/src/main/resources/plugin_job_template.json
@@ -0,0 +1,77 @@
+{
+ "job": {
+ "setting": {
+ "speed": {
+ "channel": 1
+ },
+ "errorLimit": {
+ "record": 1
+ }
+ },
+
+ "content": [
+ {
+ "reader": {
+ "name": "gdbreader",
+ "parameter": {
+ "host": "10.218.145.24",
+ "port": 8182,
+ "username": "***",
+ "password": "***",
+ "labelType": "EDGE",
+ "labels": ["label1", "label2"],
+ "column": [
+ {
+ "name": "id",
+ "type": "string",
+ "columnType": "primaryKey"
+ },
+ {
+ "name": "label",
+ "type": "string",
+ "columnType": "primaryLabel"
+ },
+ {
+ "name": "srcId",
+ "type": "string",
+ "columnType": "srcPrimaryKey"
+ },
+ {
+ "name": "srcLabel",
+ "type": "string",
+ "columnType": "srcPrimaryLabel"
+ },
+ {
+ "name": "dstId",
+ "type": "string",
+ "columnType": "srcPrimaryKey"
+ },
+ {
+ "name": "dstLabel",
+ "type": "string",
+ "columnType": "srcPrimaryLabel"
+ },
+ {
+ "name": "name",
+ "type": "string",
+ "columnType": "edgeProperty"
+ },
+ {
+ "name": "weight",
+ "type": "double",
+ "columnType": "edgeProperty"
+ }
+ ]
+ }
+ },
+
+ "writer": {
+ "name": "streamwriter",
+ "parameter": {
+ "print": true
+ }
+ }
+ }
+ ]
+ }
+}
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index dc006b5a..121c47d8 100755
--- a/pom.xml
+++ b/pom.xml
@@ -66,6 +66,7 @@
tsdbreader
opentsdbreader
cassandrareader
+ gdbreader
mysqlwriter
From d9f2f4aa0d09c99c6d000ecdca069b16a3338c7e Mon Sep 17 00:00:00 2001
From: "jiye.tjy"
Date: Tue, 14 Apr 2020 21:22:23 +0800
Subject: [PATCH 13/77] Add Clickhouse Writer
---
clickhousewriter/pom.xml | 94 +++++
.../src/main/assembly/package.xml | 35 ++
.../clickhousewriter/ClickhouseWriter.java | 330 ++++++++++++++++++
.../ClickhouseWriterErrorCode.java | 31 ++
.../src/main/resources/plugin.json | 6 +
.../main/resources/plugin_job_template.json | 21 ++
package.xml | 7 +
.../datax/plugin/rdbms/util/DataBaseType.java | 7 +-
pom.xml | 1 +
9 files changed, 531 insertions(+), 1 deletion(-)
create mode 100644 clickhousewriter/pom.xml
create mode 100755 clickhousewriter/src/main/assembly/package.xml
create mode 100644 clickhousewriter/src/main/java/com/alibaba/datax/plugin/writer/clickhousewriter/ClickhouseWriter.java
create mode 100644 clickhousewriter/src/main/java/com/alibaba/datax/plugin/writer/clickhousewriter/ClickhouseWriterErrorCode.java
create mode 100755 clickhousewriter/src/main/resources/plugin.json
create mode 100644 clickhousewriter/src/main/resources/plugin_job_template.json
diff --git a/clickhousewriter/pom.xml b/clickhousewriter/pom.xml
new file mode 100644
index 00000000..41b956f4
--- /dev/null
+++ b/clickhousewriter/pom.xml
@@ -0,0 +1,94 @@
+
+
+
+ datax-all
+ com.alibaba.datax
+ 0.0.1-SNAPSHOT
+
+
+ 4.0.0
+ clickhousewriter
+ clickhousewriter
+ jar
+
+
+
+ ru.yandex.clickhouse
+ clickhouse-jdbc
+ 0.2.4
+
+
+ com.alibaba.datax
+ datax-core
+ ${datax-project-version}
+
+
+ com.alibaba.datax
+ datax-common
+ ${datax-project-version}
+
+
+ com.alibaba.datax
+ simulator
+ ${datax-project-version}
+ test
+
+
+ org.slf4j
+ slf4j-api
+
+
+
+ ch.qos.logback
+ logback-classic
+
+
+
+ com.alibaba.datax
+ plugin-rdbms-util
+ ${datax-project-version}
+
+
+
+
+
+ src/main/java
+
+ **/*.properties
+
+
+
+
+
+
+ maven-compiler-plugin
+
+ ${jdk-version}
+ ${jdk-version}
+ ${project-sourceEncoding}
+
+
+
+
+ maven-assembly-plugin
+
+
+ src/main/assembly/package.xml
+
+ datax
+
+
+
+ dwzip
+ package
+
+ single
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/clickhousewriter/src/main/assembly/package.xml b/clickhousewriter/src/main/assembly/package.xml
new file mode 100755
index 00000000..d1128bd1
--- /dev/null
+++ b/clickhousewriter/src/main/assembly/package.xml
@@ -0,0 +1,35 @@
+
+
+
+ dir
+
+ false
+
+
+ src/main/resources
+
+ plugin.json
+ plugin_job_template.json
+
+ plugin/writer/clickhousewriter
+
+
+ target/
+
+ clickhousewriter-0.0.1-SNAPSHOT.jar
+
+ plugin/writer/clickhousewriter
+
+
+
+
+
+ false
+ plugin/writer/clickhousewriter/libs
+ runtime
+
+
+
diff --git a/clickhousewriter/src/main/java/com/alibaba/datax/plugin/writer/clickhousewriter/ClickhouseWriter.java b/clickhousewriter/src/main/java/com/alibaba/datax/plugin/writer/clickhousewriter/ClickhouseWriter.java
new file mode 100644
index 00000000..6b2de4a3
--- /dev/null
+++ b/clickhousewriter/src/main/java/com/alibaba/datax/plugin/writer/clickhousewriter/ClickhouseWriter.java
@@ -0,0 +1,330 @@
+package com.alibaba.datax.plugin.writer.clickhousewriter;
+
+import com.alibaba.datax.common.element.Column;
+import com.alibaba.datax.common.element.StringColumn;
+import com.alibaba.datax.common.exception.CommonErrorCode;
+import com.alibaba.datax.common.exception.DataXException;
+import com.alibaba.datax.common.plugin.RecordReceiver;
+import com.alibaba.datax.common.spi.Writer;
+import com.alibaba.datax.common.util.Configuration;
+import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
+import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
+import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import ru.yandex.clickhouse.ClickHouseTuple;
+
+import java.sql.Array;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.sql.Types;
+import java.util.List;
+import java.util.regex.Pattern;
+
+public class ClickhouseWriter extends Writer {
+ private static final DataBaseType DATABASE_TYPE = DataBaseType.ClickHouse;
+
+ public static class Job extends Writer.Job {
+ private Configuration originalConfig = null;
+ private CommonRdbmsWriter.Job commonRdbmsWriterMaster;
+
+ @Override
+ public void init() {
+ this.originalConfig = super.getPluginJobConf();
+ this.commonRdbmsWriterMaster = new CommonRdbmsWriter.Job(DATABASE_TYPE);
+ this.commonRdbmsWriterMaster.init(this.originalConfig);
+ }
+
+ @Override
+ public void prepare() {
+ this.commonRdbmsWriterMaster.prepare(this.originalConfig);
+ }
+
+ @Override
+ public List split(int mandatoryNumber) {
+ return this.commonRdbmsWriterMaster.split(this.originalConfig, mandatoryNumber);
+ }
+
+ @Override
+ public void post() {
+ this.commonRdbmsWriterMaster.post(this.originalConfig);
+ }
+
+ @Override
+ public void destroy() {
+ this.commonRdbmsWriterMaster.destroy(this.originalConfig);
+ }
+ }
+
+ public static class Task extends Writer.Task {
+ private Configuration writerSliceConfig;
+
+ private CommonRdbmsWriter.Task commonRdbmsWriterSlave;
+
+ @Override
+ public void init() {
+ this.writerSliceConfig = super.getPluginJobConf();
+
+ this.commonRdbmsWriterSlave = new CommonRdbmsWriter.Task(DATABASE_TYPE) {
+ @Override
+ protected PreparedStatement fillPreparedStatementColumnType(PreparedStatement preparedStatement, int columnIndex, int columnSqltype, Column column) throws SQLException {
+ try {
+ if (column.getRawData() == null) {
+ preparedStatement.setNull(columnIndex + 1, columnSqltype);
+ return preparedStatement;
+ }
+
+ java.util.Date utilDate;
+ switch (columnSqltype) {
+ case Types.CHAR:
+ case Types.NCHAR:
+ case Types.CLOB:
+ case Types.NCLOB:
+ case Types.VARCHAR:
+ case Types.LONGVARCHAR:
+ case Types.NVARCHAR:
+ case Types.LONGNVARCHAR:
+ preparedStatement.setString(columnIndex + 1, column
+ .asString());
+ break;
+
+ case Types.TINYINT:
+ case Types.SMALLINT:
+ case Types.INTEGER:
+ case Types.BIGINT:
+ case Types.DECIMAL:
+ case Types.FLOAT:
+ case Types.REAL:
+ case Types.DOUBLE:
+ String strValue = column.asString();
+ if (emptyAsNull && "".equals(strValue)) {
+ preparedStatement.setNull(columnIndex + 1, columnSqltype);
+ } else {
+ switch (columnSqltype) {
+ case Types.TINYINT:
+ case Types.SMALLINT:
+ case Types.INTEGER:
+ preparedStatement.setInt(columnIndex + 1, column.asBigInteger().intValue());
+ break;
+ case Types.BIGINT:
+ preparedStatement.setLong(columnIndex + 1, column.asLong());
+ break;
+ case Types.DECIMAL:
+ preparedStatement.setBigDecimal(columnIndex + 1, column.asBigDecimal());
+ break;
+ case Types.REAL:
+ case Types.FLOAT:
+ preparedStatement.setFloat(columnIndex + 1, column.asDouble().floatValue());
+ break;
+ case Types.DOUBLE:
+ preparedStatement.setDouble(columnIndex + 1, column.asDouble());
+ break;
+ }
+ }
+ break;
+
+ case Types.DATE:
+ if (this.resultSetMetaData.getRight().get(columnIndex)
+ .equalsIgnoreCase("year")) {
+ if (column.asBigInteger() == null) {
+ preparedStatement.setString(columnIndex + 1, null);
+ } else {
+ preparedStatement.setInt(columnIndex + 1, column.asBigInteger().intValue());
+ }
+ } else {
+ java.sql.Date sqlDate = null;
+ try {
+ utilDate = column.asDate();
+ } catch (DataXException e) {
+ throw new SQLException(String.format(
+ "Date 类型转换错误：[%s]", column));
+ }
+
+ if (null != utilDate) {
+ sqlDate = new java.sql.Date(utilDate.getTime());
+ }
+ preparedStatement.setDate(columnIndex + 1, sqlDate);
+ }
+ break;
+
+ case Types.TIME:
+ java.sql.Time sqlTime = null;
+ try {
+ utilDate = column.asDate();
+ } catch (DataXException e) {
+ throw new SQLException(String.format(
+ "Date 类型转换错误：[%s]", column));
+ }
+
+ if (null != utilDate) {
+ sqlTime = new java.sql.Time(utilDate.getTime());
+ }
+ preparedStatement.setTime(columnIndex + 1, sqlTime);
+ break;
+
+ case Types.TIMESTAMP:
+ Timestamp sqlTimestamp = null;
+ if (column instanceof StringColumn && column.asString() != null) {
+ String timeStampStr = column.asString();
+ // JAVA TIMESTAMP 类型入参必须是 "2017-07-12 14:39:00.123566" 格式
+ String pattern = "^\\d+-\\d+-\\d+ \\d+:\\d+:\\d+.\\d+";
+ boolean isMatch = Pattern.matches(pattern, timeStampStr);
+ if (isMatch) {
+ sqlTimestamp = Timestamp.valueOf(timeStampStr);
+ preparedStatement.setTimestamp(columnIndex + 1, sqlTimestamp);
+ break;
+ }
+ }
+ try {
+ utilDate = column.asDate();
+ } catch (DataXException e) {
+ throw new SQLException(String.format(
+ "Date 类型转换错误：[%s]", column));
+ }
+
+ if (null != utilDate) {
+ sqlTimestamp = new Timestamp(
+ utilDate.getTime());
+ }
+ preparedStatement.setTimestamp(columnIndex + 1, sqlTimestamp);
+ break;
+
+ case Types.BINARY:
+ case Types.VARBINARY:
+ case Types.BLOB:
+ case Types.LONGVARBINARY:
+ preparedStatement.setBytes(columnIndex + 1, column
+ .asBytes());
+ break;
+
+ case Types.BOOLEAN:
+ preparedStatement.setInt(columnIndex + 1, column.asBigInteger().intValue());
+ break;
+
+ // warn: bit(1) -> Types.BIT 可使用setBoolean
+ // warn: bit(>1) -> Types.VARBINARY 可使用setBytes
+ case Types.BIT:
+ if (this.dataBaseType == DataBaseType.MySql) {
+ Boolean asBoolean = column.asBoolean();
+ if (asBoolean != null) {
+ preparedStatement.setBoolean(columnIndex + 1, asBoolean);
+ } else {
+ preparedStatement.setNull(columnIndex + 1, Types.BIT);
+ }
+ } else {
+ preparedStatement.setString(columnIndex + 1, column.asString());
+ }
+ break;
+
+ default:
+ boolean isHandled = fillPreparedStatementColumnType4CustomType(preparedStatement,
+ columnIndex, columnSqltype, column);
+ if (isHandled) {
+ break;
+ }
+ throw DataXException
+ .asDataXException(
+ DBUtilErrorCode.UNSUPPORTED_TYPE,
+ String.format(
+ "您的配置文件中的列配置信息有误. 因为DataX 不支持数据库写入这种字段类型. 字段名:[%s], 字段类型:[%d], 字段Java类型:[%s]. 请修改表中该字段的类型或者不同步该字段.",
+ this.resultSetMetaData.getLeft()
+ .get(columnIndex),
+ this.resultSetMetaData.getMiddle()
+ .get(columnIndex),
+ this.resultSetMetaData.getRight()
+ .get(columnIndex)));
+ }
+ return preparedStatement;
+ } catch (DataXException e) {
+ // fix类型转换或者溢出失败时，将具体哪一列打印出来
+ if (e.getErrorCode() == CommonErrorCode.CONVERT_NOT_SUPPORT ||
+ e.getErrorCode() == CommonErrorCode.CONVERT_OVER_FLOW) {
+ throw DataXException
+ .asDataXException(
+ e.getErrorCode(),
+ String.format(
+ "类型转化错误. 字段名:[%s], 字段类型:[%d], 字段Java类型:[%s]. 请修改表中该字段的类型或者不同步该字段.",
+ this.resultSetMetaData.getLeft()
+ .get(columnIndex),
+ this.resultSetMetaData.getMiddle()
+ .get(columnIndex),
+ this.resultSetMetaData.getRight()
+ .get(columnIndex)));
+ } else {
+ throw e;
+ }
+ }
+ }
+
+ private Object toJavaArray(Object val) {
+ if (null == val) {
+ return null;
+ } else if (val instanceof JSONArray) {
+ Object[] valArray = ((JSONArray) val).toArray();
+ for (int i = 0; i < valArray.length; i++) {
+ valArray[i] = this.toJavaArray(valArray[i]);
+ }
+ return valArray;
+ } else {
+ return val;
+ }
+ }
+
+ boolean fillPreparedStatementColumnType4CustomType(PreparedStatement ps,
+ int columnIndex, int columnSqltype,
+ Column column) throws SQLException {
+ switch (columnSqltype) {
+ case Types.OTHER:
+ if (this.resultSetMetaData.getRight().get(columnIndex).startsWith("Tuple")) {
+ throw DataXException
+ .asDataXException(ClickhouseWriterErrorCode.TUPLE_NOT_SUPPORTED_ERROR, ClickhouseWriterErrorCode.TUPLE_NOT_SUPPORTED_ERROR.getDescription());
+ } else {
+ ps.setString(columnIndex + 1, column.asString());
+ }
+ return true;
+
+ case Types.ARRAY:
+ Connection conn = ps.getConnection();
+ List