This is an automated email from the ASF dual-hosted git repository. jin pushed a commit to branch client-rename in repository https://gitbox.apache.org/repos/asf/incubator-hugegraph-toolchain.git
commit 83030454670579cb43743b422623f684f2783681 Author: alanzhao <[email protected]> AuthorDate: Tue Nov 1 00:30:40 2022 +0800 bulkload代码提交社区第一版 --- .../scripts/dependency/known-dependencies.txt | 27 ++ hugegraph-loader/assembly/static/bin/get-params.sh | 2 +- .../static/example/spark/edge_created.json | 4 + .../assembly/static/example/spark/edge_knows.json | 2 + .../assembly/static/example/spark/schema.groovy | 36 +++ .../assembly/static/example/spark/struct.json | 58 +++++ .../static/example/spark/vertex_person.json | 7 + .../static/example/spark/vertex_software.json | 3 + hugegraph-loader/pom.xml | 127 ++++++++++ .../hugegraph/loader/builder/EdgeBuilder.java | 50 +++- .../hugegraph/loader/builder/ElementBuilder.java | 3 + .../hugegraph/loader/builder/VertexBuilder.java | 14 ++ .../baidu/hugegraph/loader/constant/Constants.java | 4 + .../loader/direct/loader/DirectLoader.java | 75 ++++++ .../loader/direct/loader/HBaseDirectLoader.java | 271 +++++++++++++++++++++ .../hugegraph/loader/direct/util/SinkToHBase.java | 166 +++++++++++++ .../hugegraph/loader/executor/LoadOptions.java | 199 +++++++++------ .../hugegraph/loader/mapping/BackendStoreInfo.java | 82 +++++++ .../hugegraph/loader/mapping/LoadMapping.java | 22 +- .../loader/metrics/LoadDistributeMetrics.java | 160 ++++++++++++ .../loader/spark/HugeGraphSparkLoader.java | 94 ++++++- .../hugegraph/loader/struct/GraphStructV1.java | 8 + .../baidu/hugegraph/loader/util/MappingUtil.java | 2 +- pom.xml | 1 + 24 files changed, 1321 insertions(+), 96 deletions(-) diff --git a/hugegraph-dist/scripts/dependency/known-dependencies.txt b/hugegraph-dist/scripts/dependency/known-dependencies.txt index e6a79893..492a9d3d 100644 --- a/hugegraph-dist/scripts/dependency/known-dependencies.txt +++ b/hugegraph-dist/scripts/dependency/known-dependencies.txt @@ -179,6 +179,24 @@ hadoop-yarn-client-3.1.1.jar hadoop-yarn-common-3.1.1.jar hamcrest-2.1.jar hamcrest-core-1.3.jar +hbase-client-2.2.3.jar +hbase-common-2.2.3.jar +hbase-hadoop-compat-2.2.3.jar +hbase-hadoop2-compat-2.2.3.jar +hbase-http-2.2.3.jar +hbase-mapreduce-2.2.3.jar +hbase-metrics-2.2.3.jar +hbase-metrics-api-2.2.3.jar +hbase-procedure-2.2.3.jar +hbase-protocol-2.2.3.jar +hbase-protocol-shaded-2.2.3.jar +hbase-replication-2.2.3.jar +hbase-server-2.2.3.jar +hbase-shaded-client-byo-hadoop-2.2.3.jar +hbase-shaded-miscellaneous-2.2.1.jar +hbase-shaded-netty-2.2.1.jar +hbase-shaded-protobuf-2.2.1.jar +hbase-zookeeper-2.2.3.jar hibernate-validator-6.0.17.Final.jar hive-classification-3.1.2.jar hive-classification-3.1.3.jar @@ -225,6 +243,7 @@ hk2-utils-3.0.1.jar hppc-0.7.2.jar htrace-core4-4.0.1-incubating.jar htrace-core4-4.1.0-incubating.jar +htrace-core4-4.2.0-incubating.jar httpclient-4.5.13.jar httpclient-4.5.2.jar httpclient-4.5.9.jar @@ -579,3 +598,11 @@ zookeeper-3.4.9.jar zookeeper-3.6.2.jar zookeeper-jute-3.6.2.jar zstd-jni-1.5.0-4.jar +disruptor-3.3.6.jar +findbugs-annotations-1.3.9-1.jar +jamon-runtime-2.4.1.jar +javax.el-3.0.1-b12.jar +javax.servlet.jsp-2.3.2.jar +javax.servlet.jsp-api-2.3.1.jar +jcodings-1.0.18.jar +joni-2.1.11.jar diff --git a/hugegraph-loader/assembly/static/bin/get-params.sh b/hugegraph-loader/assembly/static/bin/get-params.sh index c73ebbbd..e0302399 100644 --- a/hugegraph-loader/assembly/static/bin/get-params.sh +++ b/hugegraph-loader/assembly/static/bin/get-params.sh @@ -27,7 +27,7 @@ function get_params() { --incremental-mode | --failure-mode | --batch-insert-threads | --single-insert-threads | \ --max-conn | --max-conn-per-route | --batch-size | --max-parse-errors | --max-insert-errors | \ --timeout | --shutdown-timeout | --retry-times | --retry-interval | --check-vertex | \ - --print-progress | --dry-run | --help) + --print-progress | --dry-run | --sink-type | --vertex-partitions | --edge-partitions | --help ) HUGEGRAPH_PARAMS="$HUGEGRAPH_PARAMS $1 $2" shift 2 ;; diff --git a/hugegraph-loader/assembly/static/example/spark/edge_created.json b/hugegraph-loader/assembly/static/example/spark/edge_created.json new file mode 100644 index 00000000..ba093eab --- /dev/null +++ b/hugegraph-loader/assembly/static/example/spark/edge_created.json @@ -0,0 +1,4 @@ +{"source_name": "marko", "target_id": 1, "date": "2017-12-10", "weight": 0.4} +{"source_name": "josh", "target_id": 1, "date": "2009-11-11", "weight": 0.4} +{"source_name": "josh", "target_id": 2, "date": "2017-12-10", "weight": 1.0} +{"source_name": "peter", "target_id": 1, "date": "2017-03-24", "weight": 0.2} diff --git a/hugegraph-loader/assembly/static/example/spark/edge_knows.json b/hugegraph-loader/assembly/static/example/spark/edge_knows.json new file mode 100644 index 00000000..0c2af9ad --- /dev/null +++ b/hugegraph-loader/assembly/static/example/spark/edge_knows.json @@ -0,0 +1,2 @@ +{"source_name": "marko", "target_name": "vadas", "date": "20160110", "weight": 0.5} +{"source_name": "marko", "target_name": "josh", "date": "20130220", "weight": 1.0} diff --git a/hugegraph-loader/assembly/static/example/spark/schema.groovy b/hugegraph-loader/assembly/static/example/spark/schema.groovy new file mode 100644 index 00000000..ea69bb53 --- /dev/null +++ b/hugegraph-loader/assembly/static/example/spark/schema.groovy @@ -0,0 +1,36 @@ +// Define schema +schema.propertyKey("name").asText().ifNotExist().create(); +schema.propertyKey("age").asInt().ifNotExist().create(); +schema.propertyKey("city").asText().ifNotExist().create(); +schema.propertyKey("weight").asDouble().ifNotExist().create(); +schema.propertyKey("lang").asText().ifNotExist().create(); +schema.propertyKey("date").asText().ifNotExist().create(); +schema.propertyKey("price").asDouble().ifNotExist().create(); + +schema.vertexLabel("person") + .properties("name", "age", "city") + .primaryKeys("name") + .nullableKeys("age", "city") + .ifNotExist() + .create(); +schema.vertexLabel("software") + .properties("name", "lang", "price") + .primaryKeys("name") + .ifNotExist() + .create(); + + +schema.edgeLabel("knows") + .sourceLabel("person") + .targetLabel("person") + .properties("date", "weight") + .ifNotExist() + .create(); +schema.edgeLabel("created") + .sourceLabel("person") + .targetLabel("software") + .properties("date", "weight") + .ifNotExist() + .create(); + + diff --git a/hugegraph-loader/assembly/static/example/spark/struct.json b/hugegraph-loader/assembly/static/example/spark/struct.json new file mode 100644 index 00000000..8782a301 --- /dev/null +++ b/hugegraph-loader/assembly/static/example/spark/struct.json @@ -0,0 +1,58 @@ +{ + "vertices": [ + { + "label": "person", + "input": { + "type": "file", + "path": "example/spark/vertex_person.json", + "format": "JSON", + "header": ["name", "age", "city"], + "charset": "UTF-8", + "skipped_line": { + "regex": "(^#|^//).*" + } + }, + "id": "name", + "null_values": ["NULL", "null", ""] + }, + { + "label": "software", + "input": { + "type": "file", + "path": "example/spark/vertex_software.json", + "format": "JSON", + "header": ["id","name", "lang", "price","ISBN"], + "charset": "GBK" + }, + "id": "name", + "ignored": ["ISBN"] + } + ], + "edges": [ + { + "label": "knows", + "source": ["source_name"], + "target": ["target_name"], + "input": { + "type": "file", + "path": "example/spark/edge_knows.json", + "format": "JSON", + "date_format": "yyyyMMdd", + "header": ["source_name","target_name", "date", "weight"] + }, + "field_mapping": { + "source_name": "name", + "target_name": "name" + } + } + ], + "backendStoreInfo": + { + "edge_tablename": "hugegraph:g_oe", + "vertex_tablename": "hugegraph:g_v", + "hbase_zookeeper_quorum":"127.0.0.1", + "hbase_zookeeper_property_clientPort":"2181", + "zookeeper_znode_parent":"/hbase" + + } +} diff --git a/hugegraph-loader/assembly/static/example/spark/vertex_person.json b/hugegraph-loader/assembly/static/example/spark/vertex_person.json new file mode 100644 index 00000000..76fa1dd8 --- /dev/null +++ b/hugegraph-loader/assembly/static/example/spark/vertex_person.json @@ -0,0 +1,7 @@ +{"name": "marko", "age": "29", "city": "Beijing"} +{"name": "vadas", "age": "27", "city": "Hongkong"} +{"name": "josh", "age": "32", "city": "Beijing"} +{"name": "peter", "age": "35", "city": "Shanghai"} +{"name": "li,nary", "age": "26", "city": "Wu,han"} +{"name": "tom", "age": "null", "city": "NULL"} + diff --git a/hugegraph-loader/assembly/static/example/spark/vertex_software.json b/hugegraph-loader/assembly/static/example/spark/vertex_software.json new file mode 100644 index 00000000..ba68ffe8 --- /dev/null +++ b/hugegraph-loader/assembly/static/example/spark/vertex_software.json @@ -0,0 +1,3 @@ +{ "name": "lop", "lang": "java","price": "328","ISBN": "ISBN978-7-107-18618-5"} +{ "name": "ripple", "lang": "java","price": "199","ISBN": "ISBN978-7-100-13678-5"} + diff --git a/hugegraph-loader/pom.xml b/hugegraph-loader/pom.xml index 70132b08..cd2b38ef 100644 --- a/hugegraph-loader/pom.xml +++ b/hugegraph-loader/pom.xml @@ -142,6 +142,117 @@ <version>30.0-jre</version> </dependency> + <!--hbase --> + <dependency> + <groupId>org.apache.hbase</groupId> + <artifactId>hbase-common</artifactId> + <version>${hbase.version}</version> + <exclusions> + <exclusion> + <groupId>com.github.stephenc.findbugs</groupId> + <artifactId>findbugs-annotations</artifactId> + </exclusion> + <exclusion> + <groupId>com.sun.jersey</groupId> + <artifactId>jersey-server</artifactId> + </exclusion> + <exclusion> + <groupId>com.sun.jersey</groupId> + <artifactId>jersey-core</artifactId> + </exclusion> + <exclusion> + <groupId>com.sun.jersey</groupId> + <artifactId>jersey-json</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>org.apache.hbase</groupId> + <artifactId>hbase-client</artifactId> + <version>${hbase.version}</version> + <exclusions> + <exclusion> + <artifactId>hadoop-auth</artifactId> + <groupId>org.apache.hadoop</groupId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>org.apache.hbase</groupId> + <artifactId>hbase-shaded-client-byo-hadoop</artifactId> + <!--<artifactId>hbase-client</artifactId>--> + <version>${hbase.version}</version> + <exclusions> + <exclusion> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-api</artifactId> + </exclusion> + <exclusion> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-log4j12</artifactId> + </exclusion> + <exclusion> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-common</artifactId> + </exclusion> + <exclusion> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-ipc</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>org.apache.hbase</groupId> + <artifactId>hbase-mapreduce</artifactId> + <version>${hbase.version}</version> + <exclusions> + <exclusion> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-api</artifactId> + </exclusion> + <exclusion> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-log4j12</artifactId> + </exclusion> + <exclusion> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-common</artifactId> + </exclusion> + <exclusion> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-mapreduce-client-core</artifactId> + </exclusion> + <exclusion> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-hdfs</artifactId> + </exclusion> + <exclusion> + <artifactId>hadoop-auth</artifactId> + <groupId>org.apache.hadoop</groupId> + </exclusion> + <exclusion> + <artifactId>javax.ws.rs-api</artifactId> + <groupId>javax.ws.rs</groupId> + </exclusion> + <exclusion> + <artifactId>jersey-client</artifactId> + <groupId>org.glassfish.jersey.core</groupId> + </exclusion> + <exclusion> + <artifactId>jersey-common</artifactId> + <groupId>org.glassfish.jersey.core</groupId> + </exclusion> + <exclusion> + <artifactId>jersey-server</artifactId> + <groupId>org.glassfish.jersey.core</groupId> + </exclusion> + <exclusion> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-distcp</artifactId> + </exclusion> + </exclusions> + </dependency> + <!-- hadoop dependency --> <dependency> <groupId>org.apache.hadoop</groupId> @@ -323,10 +434,18 @@ <groupId>org.apache.parquet</groupId> <artifactId>parquet-hadoop-bundle</artifactId> </exclusion> + <exclusion> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-common</artifactId> + </exclusion> <exclusion> <groupId>org.apache.parquet</groupId> <artifactId>parquet-column</artifactId> </exclusion> + <exclusion> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-hdfs</artifactId> + </exclusion> <exclusion> <groupId>org.apache.thrift</groupId> <artifactId>libfb303</artifactId> @@ -335,6 +454,14 @@ <groupId>com.fasterxml.jackson.core</groupId> <artifactId>*</artifactId> </exclusion> + <exclusion> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-distcp</artifactId> + </exclusion> + <exclusion> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-mapreduce-client-core</artifactId> + </exclusion> </exclusions> </dependency> </dependencies> diff --git a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/builder/EdgeBuilder.java b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/builder/EdgeBuilder.java index 864e4f74..8c8f79dd 100644 --- a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/builder/EdgeBuilder.java +++ b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/builder/EdgeBuilder.java @@ -37,6 +37,7 @@ import com.baidu.hugegraph.structure.schema.SchemaLabel; import com.baidu.hugegraph.structure.schema.VertexLabel; import com.baidu.hugegraph.util.E; import com.google.common.collect.ImmutableList; +import org.apache.spark.sql.Row; public class EdgeBuilder extends ElementBuilder<Edge> { @@ -68,7 +69,7 @@ public class EdgeBuilder extends ElementBuilder<Edge> { public EdgeMapping mapping() { return this.mapping; } - + @Override public List<Edge> build(String[] names, Object[] values) { if (this.vertexIdsIndex == null || @@ -109,6 +110,53 @@ public class EdgeBuilder extends ElementBuilder<Edge> { } return edges; } + + @Override + public List<Edge> build(Row row) { + String[] names = row.schema().fieldNames(); + Object[] values = new Object[row.size()]; + for (int i = 0; i < row.size(); i++) { + values[i] = row.get(i); + } + if (this.vertexIdsIndex == null || + !Arrays.equals(this.lastNames, names)) { + this.vertexIdsIndex = this.extractVertexIdsIndex(names); + } + + this.lastNames = names; + EdgeKVPairs kvPairs = this.newEdgeKVPairs(); + kvPairs.source.extractFromEdge(names, values, + this.vertexIdsIndex.sourceIndexes); + kvPairs.target.extractFromEdge(names, values, + this.vertexIdsIndex.targetIndexes); + kvPairs.extractProperties(names, values); + + List<Vertex> sources = kvPairs.source.buildVertices(false); + List<Vertex> targets = kvPairs.target.buildVertices(false); + if (sources.isEmpty() || targets.isEmpty()) { + return ImmutableList.of(); + } + E.checkArgument(sources.size() == 1 || targets.size() == 1 || + sources.size() == targets.size(), + "The elements number of source and target must be: " + + "1 to n, n to 1, n to n"); + int size = Math.max(sources.size(), targets.size()); + List<Edge> edges = new ArrayList<>(size); + for (int i = 0; i < size; i++) { + Vertex source = i < sources.size() ? + sources.get(i) : sources.get(0); + Vertex target = i < targets.size() ? + targets.get(i) : targets.get(0); + Edge edge = new Edge(this.mapping.label()); + edge.source(source); + edge.target(target); + // Add properties + this.addProperties(edge, kvPairs.properties); + this.checkNonNullableKeys(edge); + edges.add(edge); + } + return edges; + } private EdgeKVPairs newEdgeKVPairs() { EdgeKVPairs kvPairs = new EdgeKVPairs(); diff --git a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/builder/ElementBuilder.java b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/builder/ElementBuilder.java index fa280c6e..57c7b3a9 100644 --- a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/builder/ElementBuilder.java +++ b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/builder/ElementBuilder.java @@ -54,6 +54,7 @@ import com.baidu.hugegraph.structure.schema.VertexLabel; import com.baidu.hugegraph.util.E; import com.baidu.hugegraph.util.LongEncoding; import com.google.common.collect.ImmutableList; +import org.apache.spark.sql.Row; public abstract class ElementBuilder<GE extends GraphElement> { @@ -75,6 +76,8 @@ public abstract class ElementBuilder<GE extends GraphElement> { public abstract List<GE> build(String[] names, Object[] values); + public abstract List<GE> build(Row row); + public abstract SchemaLabel schemaLabel(); protected abstract Collection<String> nonNullableKeys(); diff --git a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/builder/VertexBuilder.java b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/builder/VertexBuilder.java index f6bbb2f5..89ff06fc 100644 --- a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/builder/VertexBuilder.java +++ b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/builder/VertexBuilder.java @@ -29,6 +29,7 @@ import com.baidu.hugegraph.structure.graph.Vertex; import com.baidu.hugegraph.structure.schema.SchemaLabel; import com.baidu.hugegraph.structure.schema.VertexLabel; import com.baidu.hugegraph.util.E; +import org.apache.spark.sql.Row; public class VertexBuilder extends ElementBuilder<Vertex> { @@ -59,6 +60,19 @@ public class VertexBuilder extends ElementBuilder<Vertex> { return kvPairs.buildVertices(true); } + @Override + public List<Vertex> build(Row row) { + VertexKVPairs kvPairs = this.newKVPairs(this.vertexLabel, + this.mapping.unfold()); + String[] names = row.schema().fieldNames(); + Object[] values = new Object[row.size()]; + for (int i = 0; i < row.size(); i++) { + values[i] = row.get(i); + } + kvPairs.extractFromVertex(names, values); + return kvPairs.buildVertices(true); + } + @Override public SchemaLabel schemaLabel() { return this.vertexLabel; diff --git a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/constant/Constants.java b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/constant/Constants.java index 6f577f00..fb42312a 100644 --- a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/constant/Constants.java +++ b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/constant/Constants.java @@ -79,4 +79,8 @@ public final class Constants { public static final String CDC_DATA = "data"; public static final String CDC_OP = "op"; + public static final String HBASE_COL_FAMILY = "f"; + public static final String LOAD_DATA_PARSE_SUFFIX = "parse"; + public static final String LOAD_DATA_SER_SUFFIX = "ser"; + public static final String LOAD_DATA_INSERT_SUFFIX = "insert"; } diff --git a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/direct/loader/DirectLoader.java b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/direct/loader/DirectLoader.java new file mode 100644 index 00000000..bbbc4359 --- /dev/null +++ b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/direct/loader/DirectLoader.java @@ -0,0 +1,75 @@ +/* + * Copyright 2017 HugeGraph Authors + * + * 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 com.baidu.hugegraph.loader.direct.loader; + +import com.baidu.hugegraph.loader.builder.EdgeBuilder; +import com.baidu.hugegraph.loader.builder.ElementBuilder; +import com.baidu.hugegraph.loader.builder.VertexBuilder; +import com.baidu.hugegraph.loader.executor.LoadContext; +import com.baidu.hugegraph.loader.executor.LoadOptions; +import com.baidu.hugegraph.loader.mapping.EdgeMapping; +import com.baidu.hugegraph.loader.mapping.InputStruct; +import com.baidu.hugegraph.loader.mapping.VertexMapping; +import org.apache.spark.api.java.JavaPairRDD; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; + +import java.io.Serializable; +import java.util.LinkedList; +import java.util.List; + +public abstract class DirectLoader<T,R> implements Serializable { + LoadOptions loadOptions; + InputStruct struct; + + public DirectLoader(LoadOptions loadOptions, + InputStruct struct) { + this.loadOptions = loadOptions; + this.struct = struct; + } + + public final void bulkload(Dataset<Row> ds) { + JavaPairRDD<T, R> javaPairRDD = buildVertexAndEdge(ds); + String path = generateFiles(javaPairRDD); + loadFiles(path); + } + + protected List<ElementBuilder> getElementBuilders() { + LoadContext context = new LoadContext(loadOptions); + context.schemaCache().updateAll(); + List<ElementBuilder> buildersForGraphElement = new LinkedList<>(); + for (VertexMapping vertexMapping : struct.vertices()) { + buildersForGraphElement.add( + new VertexBuilder(context, struct, vertexMapping) + ); + } + for (EdgeMapping edgeMapping : struct.edges()) { + buildersForGraphElement.add(new EdgeBuilder(context, struct, edgeMapping)); + } + context.close(); + return buildersForGraphElement; + } + + abstract JavaPairRDD<T, R> buildVertexAndEdge(Dataset<Row> ds); + + abstract String generateFiles(JavaPairRDD<T, R> buildAndSerRdd); + + abstract void loadFiles(String path); +} diff --git a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/direct/loader/HBaseDirectLoader.java b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/direct/loader/HBaseDirectLoader.java new file mode 100644 index 00000000..89045ed5 --- /dev/null +++ b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/direct/loader/HBaseDirectLoader.java @@ -0,0 +1,271 @@ +/* + * Copyright 2017 HugeGraph Authors + * + * 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 com.baidu.hugegraph.loader.direct.loader; + +import com.baidu.hugegraph.loader.builder.ElementBuilder; +import com.baidu.hugegraph.loader.constant.Constants; +import com.baidu.hugegraph.loader.direct.util.SinkToHBase; +import com.baidu.hugegraph.loader.executor.LoadOptions; +import com.baidu.hugegraph.loader.mapping.ElementMapping; +import com.baidu.hugegraph.loader.mapping.InputStruct; +import com.baidu.hugegraph.loader.metrics.LoadDistributeMetrics; +import com.baidu.hugegraph.loader.util.HugeClientHolder; +import com.baidu.hugegraph.serializer.direct.HBaseSerializer; +import com.baidu.hugegraph.structure.GraphElement; +import com.baidu.hugegraph.structure.graph.Edge; +import com.baidu.hugegraph.structure.graph.Vertex; +import com.baidu.hugegraph.util.Log; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.FsShell; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.client.TableDescriptor; +import org.apache.hadoop.hbase.io.ImmutableBytesWritable; +import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.mapreduce.Job; +import org.apache.spark.Partitioner; +import org.apache.spark.api.java.JavaPairRDD; +import org.apache.spark.api.java.function.PairFlatMapFunction; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.slf4j.Logger; +import scala.Tuple2; + +import java.io.IOException; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.List; + +public class HBaseDirectLoader extends DirectLoader<ImmutableBytesWritable, KeyValue> { + + private SinkToHBase sinkToHBase; + private LoadDistributeMetrics loadDistributeMetrics; + + public static final Logger LOG = Log.logger(HBaseDirectLoader.class); + + public HBaseDirectLoader(LoadOptions loadOptions, + InputStruct struct, + LoadDistributeMetrics loadDistributeMetrics) { + super(loadOptions,struct); + this.loadDistributeMetrics = loadDistributeMetrics; + this.sinkToHBase = new SinkToHBase(loadOptions); + + } + + public String getTableName() { + + String tableName = null; + if (struct.edges().size() > 0) { + tableName = this.loadOptions.edgeTablename; + + } else if (struct.vertices().size() > 0) { + tableName = this.loadOptions.vertexTablename; + + } + return tableName; + } + + public Integer getTablePartitions () { + return struct.edges().size() > 0 ? + loadOptions.edgePartitions : + loadOptions.vertexPartitions; + } + + public JavaPairRDD<ImmutableBytesWritable, KeyValue> buildVertexAndEdge(Dataset<Row> ds) { + LOG.info("buildAndSer start execute >>>>"); + JavaPairRDD<ImmutableBytesWritable, KeyValue> tuple2KeyValueJavaPairRDD = + ds.toJavaRDD().mapPartitionsToPair( + new PairFlatMapFunction<Iterator<Row>, ImmutableBytesWritable, KeyValue>() { + @Override + public Iterator<Tuple2<ImmutableBytesWritable, KeyValue>> call( + Iterator<Row> rowIterator) throws Exception { + + HBaseSerializer serializer = new HBaseSerializer( + HugeClientHolder.create(loadOptions), + loadOptions.vertexPartitions,loadOptions.edgePartitions); + List<ElementBuilder> buildersForGraphElement = getElementBuilders(); + List<Tuple2<ImmutableBytesWritable, KeyValue>> result = + new LinkedList<>(); + while (rowIterator.hasNext()) { + Row row = rowIterator.next(); + List<Tuple2<ImmutableBytesWritable, KeyValue>> serList = + buildAndSer(serializer, row,buildersForGraphElement); + result.addAll(serList); + } + serializer.close(); + return result.iterator(); + } + } + ); + return tuple2KeyValueJavaPairRDD; + } + + @Override + String generateFiles(JavaPairRDD<ImmutableBytesWritable, KeyValue> buildAndSerRdd) { + LOG.info("bulkload start execute>>>"); + try { + Tuple2<SinkToHBase.IntPartitioner, TableDescriptor> tuple = + sinkToHBase.getPartitionerByTableName(getTablePartitions(), getTableName()); + Partitioner partitioner = (Partitioner) tuple._1; + TableDescriptor tableDescriptor = (TableDescriptor) tuple._2; + + JavaPairRDD<ImmutableBytesWritable, KeyValue> repartitionedRdd = + buildAndSerRdd.repartitionAndSortWithinPartitions(partitioner); + Configuration conf = sinkToHBase.getHBaseConfiguration().get(); + Job job = Job.getInstance(conf); + HFileOutputFormat2.configureIncrementalLoadMap(job, tableDescriptor); + conf.set("hbase.mapreduce.hfileoutputformat.table.name", + tableDescriptor.getTableName().getNameAsString()); + String path = getHFilePath(job.getConfiguration()); + repartitionedRdd.saveAsNewAPIHadoopFile( + path, + ImmutableBytesWritable.class, + KeyValue.class, + HFileOutputFormat2.class, + conf + ); + LOG.info("Saved to HFiles to: " + path); + flushPermission(conf,path); + return path; + } catch (IOException e) { + e.printStackTrace(); + } + + return Constants.EMPTY_STR; + + } + + public String getHFilePath (Configuration conf) throws IOException { + FileSystem fs = FileSystem.get(conf); + long timeStr = System.currentTimeMillis(); + String pathStr = fs.getWorkingDirectory().toString() + "/hfile-gen" + "/" + timeStr + "/"; + Path hfileGenPath = new Path(pathStr); + if (fs.exists(hfileGenPath)) { + LOG.info("\n delete hfile path \n"); + fs.delete(hfileGenPath,true); + } + return pathStr; + } + + @Override + public void loadFiles(String path) { + try { + sinkToHBase.loadHfiles(path, getTableName());// BulkLoad HFile to HBase + } catch (Exception e) { + e.printStackTrace(); + } + } + + private void flushPermission (Configuration conf, String path) { + FsShell shell = new FsShell(conf); + try { + LOG.info("shell start execute"); + shell.run(new String[]{"-chmod", "-R", "777", path}); + shell.close(); + } catch (Exception e) { + LOG.error("Couldnt change the file permissions " + e + + " Please run command:" + + "hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles " + path + + " '" + "test" + "'\n" + " to load generated HFiles into HBase table"); + } + } + + List<Tuple2<ImmutableBytesWritable, KeyValue>> buildAndSer (HBaseSerializer serializer, + Row row, + List<ElementBuilder> builders) { + List<GraphElement> elementsElement; + + List<Tuple2<ImmutableBytesWritable, KeyValue>> result = new LinkedList<>(); + + for (ElementBuilder builder : builders) { + ElementMapping elementMapping = builder.mapping(); + if (elementMapping.skip()) { + continue; + } + if ("".equals(row.mkString())) { + break; + } + switch (struct.input().type()) { + case FILE: + case HDFS: + + elementsElement = builder.build(row); + + break; + default: + throw new AssertionError(String.format( + "Unsupported input source '%s'", + struct.input().type())); + } + + boolean isVertex = builder.mapping().type().isVertex(); + if (isVertex) { + for (Vertex vertex : (List<Vertex>) (Object) elementsElement) { + LOG.debug("vertex already build done " + vertex.toString()); + Tuple2<ImmutableBytesWritable, KeyValue> tuple2 = + vertexSerialize(serializer,vertex); + loadDistributeMetrics.increaseDisVertexInsertSuccess(builder.mapping()); + result.add(tuple2); + } + } else { + for (Edge edge : (List<Edge>) (Object) elementsElement) { + LOG.debug("edge already build done " + edge.toString()); + Tuple2<ImmutableBytesWritable, KeyValue> tuple2 = + edgeSerialize(serializer,edge); + loadDistributeMetrics.increaseDisEdgeInsertSuccess(builder.mapping()); + result.add(tuple2); + + } + } + } + return result; + } + + private Tuple2<ImmutableBytesWritable, KeyValue> edgeSerialize (HBaseSerializer serializer, + Edge edge) { + LOG.debug("edge start serialize " + edge.toString()); + byte[] rowkey = serializer.getKeyBytes(edge); + byte[] values = serializer.getValueBytes(edge); + ImmutableBytesWritable rowKey = new ImmutableBytesWritable(); + rowKey.set(rowkey); + KeyValue keyValue = new KeyValue(rowkey, + Bytes.toBytes(Constants.HBASE_COL_FAMILY), + Bytes.toBytes(Constants.EMPTY_STR), + values); + return new Tuple2<>(rowKey,keyValue); + } + + private Tuple2<ImmutableBytesWritable, KeyValue> vertexSerialize (HBaseSerializer serializer, + Vertex vertex) { + LOG.debug("vertex start serialize " + vertex.toString()); + byte[] rowkey = serializer.getKeyBytes(vertex); + byte[] values = serializer.getValueBytes(vertex); + ImmutableBytesWritable rowKey = new ImmutableBytesWritable(); + rowKey.set(rowkey); + KeyValue keyValue = new KeyValue(rowkey, + Bytes.toBytes(Constants.HBASE_COL_FAMILY), + Bytes.toBytes(Constants.EMPTY_STR), + values); + return new Tuple2<>(rowKey,keyValue); + } + +} diff --git a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/direct/util/SinkToHBase.java b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/direct/util/SinkToHBase.java new file mode 100644 index 00000000..28062ab8 --- /dev/null +++ b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/direct/util/SinkToHBase.java @@ -0,0 +1,166 @@ +/* + * Copyright 2017 HugeGraph Authors + * + * 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 com.baidu.hugegraph.loader.direct.util; + +import com.baidu.hugegraph.loader.executor.LoadOptions; +import com.baidu.hugegraph.util.Log; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.Connection; +import org.apache.hadoop.hbase.client.ConnectionFactory; +import org.apache.hadoop.hbase.client.Table; +import org.apache.hadoop.hbase.client.TableDescriptor; +import org.apache.hadoop.hbase.io.ImmutableBytesWritable; +import org.apache.hadoop.hbase.tool.BulkLoadHFilesTool; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.util.Pair; +import org.apache.spark.Partitioner; +import org.slf4j.Logger; +import scala.Tuple2; +import org.apache.hadoop.hbase.client.HRegionLocator; +import java.util.ArrayList; +import java.util.Arrays; +import java.io.IOException; +import java.util.HashMap; +import java.io.Serializable; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +public class SinkToHBase implements Serializable { + + private LoadOptions loadOptions; + public static final Logger LOG = Log.logger(SinkToHBase.class); + + public SinkToHBase(LoadOptions loadOptions) { + this.loadOptions = loadOptions; + } + + public Optional<Configuration> getHBaseConfiguration() { + Configuration baseConf = HBaseConfiguration.create(); + baseConf.set("hbase.zookeeper.quorum", this.loadOptions.hbaseZKQuorum); + baseConf.set("hbase.zookeeper.property.clientPort", this.loadOptions.hbaseZKPort); + baseConf.set("zookeeper.znode.parent", this.loadOptions.hbaseZKParent); + return Optional.ofNullable(baseConf); + } + + private Optional<Connection> getConnection() { + Optional<Configuration> baseConf = getHBaseConfiguration(); + Connection conn = null; + try { + conn = ConnectionFactory.createConnection(baseConf.get()); + } catch (IOException e) { + e.printStackTrace(); + } + return Optional.ofNullable(conn); + } + + public Tuple2<IntPartitioner, TableDescriptor> + getPartitionerByTableName (int numPartitions, String tableName) throws IOException { + Optional<Connection> optionalConnection = getConnection(); + TableDescriptor descriptor = optionalConnection + .get() + .getTable(TableName.valueOf(tableName)) + .getDescriptor(); + LOG.debug("getPartitionerByTableName get TableDescriptor " + + descriptor.getTableName()); + optionalConnection.get().close(); + return new Tuple2<IntPartitioner,TableDescriptor>( + new IntPartitioner(numPartitions, tableName),descriptor); + } + + public void loadHfiles (String path, String tableName) throws Exception { + Connection conn = getConnection().get(); + Table table = conn.getTable(TableName.valueOf(tableName)); + Configuration conf = conn.getConfiguration(); + BulkLoadHFilesTool bulkLoadHFilesTool = new BulkLoadHFilesTool(conf); + bulkLoadHFilesTool.bulkLoad(table.getName(), new Path(path)); + table.close(); + conn.close(); + + } + + public class IntPartitioner extends Partitioner { + private final int numPartitions; + public Map<List<String>, Integer> rangeMap = new HashMap<>(); + private String tableName; + + public IntPartitioner(int numPartitions, String tableName) throws IOException { + this.numPartitions = numPartitions; + this.rangeMap = getRangeMap(tableName); + this.tableName = tableName; + } + + private Map<List<String>, Integer> getRangeMap(String tableName) throws IOException { + Connection conn = getConnection().get(); + HRegionLocator locator = + (HRegionLocator) conn.getRegionLocator(TableName.valueOf(tableName)); + Pair<byte[][], byte[][]> startEndKeys = locator.getStartEndKeys(); + Map<List<String>, Integer> rangeMap = new HashMap<>(); + for (int i = 0; i < startEndKeys.getFirst().length; i++) { + String startKey = Bytes.toString(startEndKeys.getFirst()[i]); + String endKey = Bytes.toString(startEndKeys.getSecond()[i]); + rangeMap.put(new ArrayList<>(Arrays.asList(startKey, endKey)), i); + } + conn.close(); + return rangeMap; + } + + @Override + public int numPartitions() { + return numPartitions; + } + + @Override + public int getPartition(Object key) { + if (key instanceof ImmutableBytesWritable) { + + try { + ImmutableBytesWritable immutableBytesWritableKey = (ImmutableBytesWritable) key; + + if (rangeMap == null || rangeMap.isEmpty()) { + rangeMap = getRangeMap(this.tableName); + } + + String keyString = Bytes.toString(immutableBytesWritableKey.get()); + for (List<String> range : rangeMap.keySet()) { + if (keyString.compareToIgnoreCase(range.get(0)) >= 0 && + ((keyString.compareToIgnoreCase(range.get(1)) < 0) || + range.get(1).equals(""))) { + return rangeMap.get(range); + } + } + LOG.error("Didn't find proper key in rangeMap, so returning 0 ..."); + return 0; + } catch (Exception e) { + LOG.error("When trying to get partitionID, encountered exception: " + e + + "\t key = " + key); + return 0; + } + } else { + LOG.error("key is NOT ImmutableBytesWritable type ..."); + return 0; + } + } + } + +} \ No newline at end of file diff --git a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/executor/LoadOptions.java b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/executor/LoadOptions.java index ac12d53b..aa719622 100644 --- a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/executor/LoadOptions.java +++ b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/executor/LoadOptions.java @@ -23,6 +23,7 @@ import java.io.File; import java.io.Serializable; import java.util.Set; +import com.baidu.hugegraph.loader.mapping.BackendStoreInfo; import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; @@ -44,168 +45,198 @@ public class LoadOptions implements Serializable { public static final String HTTP_SCHEMA = "http"; private static final int CPUS = Runtime.getRuntime().availableProcessors(); + @Parameter(names = {"-f", "--file"}, required = true, arity = 1, - validateWith = {FileValidator.class}, - description = "The path of the data mapping description file") + validateWith = {FileValidator.class}, + description = "The path of the data mapping description file") public String file; @Parameter(names = {"-s", "--schema"}, arity = 1, - validateWith = {FileValidator.class}, - description = "The schema file path which to create manually") + validateWith = {FileValidator.class}, + description = "The schema file path which to create manually") public String schema; @Parameter(names = {"-g", "--graph"}, required = true, arity = 1, - description = "The namespace of the graph to load into") + description = "The namespace of the graph to load into") public String graph; @Parameter(names = {"-h", "--host"}, arity = 1, - validateWith = {UrlValidator.class}, - description = "The host/IP of HugeGraphServer") + validateWith = {UrlValidator.class}, + description = "The host/IP of HugeGraphServer") public String host = "localhost"; @Parameter(names = {"-p", "--port"}, arity = 1, - validateWith = {PositiveValidator.class}, - description = "The port of HugeGraphServer") + validateWith = {PositiveValidator.class}, + description = "The port of HugeGraphServer") public int port = 8080; @Parameter(names = {"--username"}, arity = 1, - description = "The username of graph for authentication") + description = "The username of graph for authentication") public String username = null; @Parameter(names = {"--protocol"}, arity = 1, - validateWith = {ProtocolValidator.class}, - description = "The protocol of HugeGraphServer, " + - "allowed values are: http or https") + validateWith = {ProtocolValidator.class}, + description = "The protocol of HugeGraphServer, " + + "allowed values are: http or https") public String protocol = "http"; @Parameter(names = {"--trust-store-file"}, arity = 1, - description = "The path of client truststore file used " + - "when https protocol is enabled") + description = "The path of client truststore file used " + + "when https protocol is enabled") public String trustStoreFile = null; @Parameter(names = {"--trust-store-password"}, arity = 1, - description = "The password of client truststore file used " + - "when https protocol is enabled") + description = "The password of client truststore file used " + + "when https protocol is enabled") public String trustStoreToken = null; @Parameter(names = {"--token"}, arity = 1, - description = "The token of graph for authentication") + description = "The token of graph for authentication") public String token = null; @Parameter(names = {"--clear-all-data"}, arity = 1, - description = "Whether to clear all old data before loading") + description = "Whether to clear all old data before loading") public boolean clearAllData = false; @Parameter(names = {"--clear-timeout"}, arity = 1, - validateWith = {PositiveValidator.class}, - description = "The timeout waiting for clearing all data ") + validateWith = {PositiveValidator.class}, + description = "The timeout waiting for clearing all data ") public int clearTimeout = 240; @Parameter(names = {"--incremental-mode"}, arity = 1, - description = "Load data from the breakpoint of last time") + description = "Load data from the breakpoint of last time") public boolean incrementalMode = false; @Parameter(names = {"--failure-mode"}, arity = 1, - description = "Load data from the failure records, in this " + - "mode, only full load is supported, any read " + - "or parsing errors will cause load task stop") + description = "Load data from the failure records, in this " + + "mode, only full load is supported, any read " + + "or parsing errors will cause load task stop") public boolean failureMode = false; @Parameter(names = {"--batch-insert-threads"}, arity = 1, - validateWith = {PositiveValidator.class}, - description = "The number of threads to execute batch insert") + validateWith = {PositiveValidator.class}, + description = "The number of threads to execute batch insert") public int batchInsertThreads = CPUS; @Parameter(names = {"--single-insert-threads"}, arity = 1, - validateWith = {PositiveValidator.class}, - description = "The number of threads to execute single insert") + validateWith = {PositiveValidator.class}, + description = "The number of threads to execute single insert") public int singleInsertThreads = 8; @Parameter(names = {"--max-conn"}, arity = 1, - description = "Max number of HTTP connections to server") + description = "Max number of HTTP connections to server") public int maxConnections = CPUS * 4; @Parameter(names = {"--max-conn-per-route"}, arity = 1, - description = "Max number of HTTP connections to each route") + description = "Max number of HTTP connections to each route") public int maxConnectionsPerRoute = CPUS * 2; @Parameter(names = {"--batch-size"}, arity = 1, - validateWith = {PositiveValidator.class}, - description = "The number of lines in each submit") + validateWith = {PositiveValidator.class}, + description = "The number of lines in each submit") public int batchSize = 500; @Parameter(names = {"--cdc-flush-interval"}, arity = 1, - description = "The flush interval for flink cdc") + description = "The flush interval for flink cdc") public int flushIntervalMs = 30000; @Parameter(names = {"--cdc-sink-parallelism"}, arity = 1, - description = "The sink parallelism for flink cdc") + description = "The sink parallelism for flink cdc") public int sinkParallelism = 1; @Parameter(names = {"--shutdown-timeout"}, arity = 1, - validateWith = {PositiveValidator.class}, - description = "The timeout of awaitTermination in seconds") + validateWith = {PositiveValidator.class}, + description = "The timeout of awaitTermination in seconds") public int shutdownTimeout = 10; @Parameter(names = {"--check-vertex"}, arity = 1, - description = "Check vertices exists while inserting edges") + description = "Check vertices exists while inserting edges") public boolean checkVertex = false; @Parameter(names = {"--max-read-errors"}, arity = 1, - validateWith = {PositiveValidator.class}, - description = "The maximum number of lines that read error " + - "before exiting") + validateWith = {PositiveValidator.class}, + description = "The maximum number of lines that read error " + + "before exiting") public int maxReadErrors = 1; @Parameter(names = {"--max-parse-errors"}, arity = 1, - validateWith = {PositiveValidator.class}, - description = "The maximum number of lines that parse error " + - "before exiting") + validateWith = {PositiveValidator.class}, + description = "The maximum number of lines that parse error " + + "before exiting") public int maxParseErrors = 1; @Parameter(names = {"--max-insert-errors"}, arity = 1, - validateWith = {PositiveValidator.class}, - description = "The maximum number of lines that insert error " + - "before exiting") + validateWith = {PositiveValidator.class}, + description = "The maximum number of lines that insert error " + + "before exiting") public int maxInsertErrors = 500; @Parameter(names = {"--timeout"}, arity = 1, - validateWith = {PositiveValidator.class}, - description = "The timeout of HugeClient request") + validateWith = {PositiveValidator.class}, + description = "The timeout of HugeClient request") public int timeout = 60; @Parameter(names = {"--retry-times"}, arity = 1, - validateWith = {PositiveValidator.class}, - description = "Setting the max retry times when loading timeout") + validateWith = {PositiveValidator.class}, + description = "Setting the max retry times when loading timeout") public int retryTimes = 3; @Parameter(names = {"--retry-interval"}, arity = 1, - validateWith = {PositiveValidator.class}, - description = "Setting the interval time before retrying") + validateWith = {PositiveValidator.class}, + description = "Setting the interval time before retrying") public int retryInterval = 10; @Parameter(names = {"--max-read-lines"}, arity = 1, - validateWith = {PositiveValidator.class}, - description = "The maximum number of read lines, when reached " + - "this number, the load task will stop") + validateWith = {PositiveValidator.class}, + description = "The maximum number of read lines, when reached " + + "this number, the load task will stop") public long maxReadLines = -1L; @Parameter(names = {"--dry-run"}, arity = 1, - description = "Dry run means that only parse but doesn't load") + description = "Dry run means that only parse but doesn't load") public boolean dryRun = false; @Parameter(names = {"--print-progress"}, arity = 1, - description = "Whether to print real-time load progress") + description = "Whether to print real-time load progress") public boolean printProgress = true; @Parameter(names = {"--test-mode"}, arity = 1, - description = "Whether the hugegraph-loader work in test mode") + description = "Whether the hugegraph-loader work in test mode") public boolean testMode = false; @Parameter(names = {"--help"}, help = true, - description = "Print usage of HugeGraphLoader") + description = "Print usage of HugeGraphLoader") public boolean help; + @Parameter(names = {"--sink-type"}, arity = 1, + description = "Sink to different storage") + public boolean sinkType = true; + + + @Parameter(names = {"--edge-partitions"}, arity = 1, + description = "The number of partitions of the HBase edge table") + public int edgePartitions = 64; + + @Parameter(names = {"--vertex-partitions"}, arity = 1, + description = "The number of partitions of the HBase vertex table") + public int vertexPartitions = 64; + + @Parameter(names = {"edgeTablename"}, arity = 1, + description = "edgeTablename") + public String edgeTablename; + @Parameter(names = {"vertexTablename"}, arity = 1, + description = "vertexTablename") + public String vertexTablename; + @Parameter(names = {"hbaseZKQuorum"}, arity = 1, + description = "hbaseZKQuorum") + public String hbaseZKQuorum; + @Parameter(names = {"hbaseZKPort"}, arity = 1, + description = "hbaseZKPort") + public String hbaseZKPort; + @Parameter(names = {"hbaseZKParent"}, arity = 1, + description = "hbaseZKParent") + public String hbaseZKParent; + public String workModeString() { if (this.incrementalMode) { return "INCREMENTAL MODE"; @@ -219,8 +250,8 @@ public class LoadOptions implements Serializable { public static LoadOptions parseOptions(String[] args) { LoadOptions options = new LoadOptions(); JCommander commander = JCommander.newBuilder() - .addObject(options) - .build(); + .addObject(options) + .build(); commander.parse(args); // Print usage and exit if (options.help) { @@ -229,10 +260,10 @@ public class LoadOptions implements Serializable { // Check options // Check option "-f" E.checkArgument(!StringUtils.isEmpty(options.file), - "The mapping file must be specified"); + "The mapping file must be specified"); E.checkArgument(options.file.endsWith(Constants.JSON_SUFFIX), - "The mapping file name must be end with %s", - Constants.JSON_SUFFIX); + "The mapping file name must be end with %s", + Constants.JSON_SUFFIX); File mappingFile = new File(options.file); if (!mappingFile.canRead()) { LOG.error("The mapping file must be readable: '{}'", mappingFile); @@ -241,7 +272,7 @@ public class LoadOptions implements Serializable { // Check option "-g" E.checkArgument(!StringUtils.isEmpty(options.graph), - "The graph must be specified"); + "The graph must be specified"); // Check option "-h" if (!options.host.startsWith(Constants.HTTP_PREFIX)) { if (options.protocol.equals(HTTP_SCHEMA)) { @@ -252,7 +283,7 @@ public class LoadOptions implements Serializable { } // Check option --incremental-mode and --failure-mode E.checkArgument(!(options.incrementalMode && options.failureMode), - "The option --incremental-mode and --failure-mode " + + "The option --incremental-mode and --failure-mode " + "can't be true at same time"); if (options.failureMode) { LOG.info("The failure-mode will scan the entire error file"); @@ -263,17 +294,27 @@ public class LoadOptions implements Serializable { return options; } + public void copyBackendStoreInfo (BackendStoreInfo backendStoreInfo) { + E.checkArgument(null != backendStoreInfo,"the backendStoreInfo is null "); + this.edgeTablename = backendStoreInfo.getEdgeTablename(); + this.vertexTablename = backendStoreInfo.getVertexTablename(); + this.hbaseZKParent = backendStoreInfo.getHbaseZKParent(); + this.hbaseZKPort = backendStoreInfo.getHbaseZKPort(); + this.hbaseZKQuorum = backendStoreInfo.getHbaseZKQuorum(); + + } + public static class UrlValidator implements IParameterValidator { @Override public void validate(String name, String value) { String regex = "^((http)(s?)://)?" + - "(([0-9]{1,3}\\.){3}[0-9]{1,3}" + // IP URL - "|" + // Or domain name - "([0-9a-z_!~*'()-]+\\.)*[0-9a-z_!~*'()-]+)$"; + "(([0-9]{1,3}\\.){3}[0-9]{1,3}" + // IP URL + "|" + // Or domain name + "([0-9a-z_!~*'()-]+\\.)*[0-9a-z_!~*'()-]+)$"; if (!value.matches(regex)) { throw new ParameterException(String.format( - "Invalid url value of args '%s': '%s'", name, value)); + "Invalid url value of args '%s': '%s'", name, value)); } } } @@ -288,8 +329,8 @@ public class LoadOptions implements Serializable { public void validate(String name, String value) { if (!SSL_PROTOCOL.contains(value.toLowerCase())) { throw new ParameterException(String.format( - "Invalid --protocol '%s', valid value is %s", - value, SSL_PROTOCOL)); + "Invalid --protocol '%s', valid value is %s", + value, SSL_PROTOCOL)); } } } @@ -301,8 +342,8 @@ public class LoadOptions implements Serializable { File file = new File(value); if (!file.exists() || !file.isDirectory()) { throw new ParameterException(String.format( - "Ensure the directory exists and is indeed a " + - "directory instead of a file: '%s'", value)); + "Ensure the directory exists and is indeed a " + + "directory instead of a file: '%s'", value)); } } } @@ -314,8 +355,8 @@ public class LoadOptions implements Serializable { File file = new File(value); if (!file.exists() || !file.isFile()) { throw new ParameterException(String.format( - "Ensure the file exists and is indeed a file " + - "instead of a directory: '%s'", value)); + "Ensure the file exists and is indeed a file " + + "instead of a directory: '%s'", value)); } } } @@ -327,8 +368,8 @@ public class LoadOptions implements Serializable { int retry = Integer.parseInt(value); if (retry <= 0) { throw new ParameterException(String.format( - "Parameter '%s' should be positive, but got '%s'", - name, value)); + "Parameter '%s' should be positive, but got '%s'", + name, value)); } } } diff --git a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/mapping/BackendStoreInfo.java b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/mapping/BackendStoreInfo.java new file mode 100644 index 00000000..638ca011 --- /dev/null +++ b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/mapping/BackendStoreInfo.java @@ -0,0 +1,82 @@ +/* + * Copyright 2017 HugeGraph Authors + * + * 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 com.baidu.hugegraph.loader.mapping; + +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.annotation.JsonPropertyOrder; + +@JsonPropertyOrder({"edge_tablename", "vertex_tablename", "hbase_zookeeper_quorum", + "hbase_zookeeper_property_clientPort", "zookeeper_znode_parent"}) +public class BackendStoreInfo { + @JsonProperty("edge_tablename") + private String edgeTablename; + @JsonProperty("vertex_tablename") + private String vertexTablename; + @JsonProperty("hbase_zookeeper_quorum") + private String hbaseZKQuorum; + @JsonProperty("hbase_zookeeper_property_clientPort") + private String hbaseZKPort; + @JsonProperty("zookeeper_znode_parent") + private String hbaseZKParent; + + public String getEdgeTablename() { + return edgeTablename; + } + + public void setEdgeTablename(String edgeTablename) { + this.edgeTablename = edgeTablename; + } + + public String isVertexTablename() { + return vertexTablename; + } + + public void setVertexTablename(String vertexTablename) { + this.vertexTablename = vertexTablename; + } + + public String getVertexTablename() { + return vertexTablename; + } + + public String getHbaseZKQuorum() { + return hbaseZKQuorum; + } + + public void setHbaseZKQuorum(String hbaseZKQuorum) { + this.hbaseZKQuorum = hbaseZKQuorum; + } + + public String getHbaseZKPort() { + return hbaseZKPort; + } + + public void setHbaseZKPort(String hbaseZKPort) { + this.hbaseZKPort = hbaseZKPort; + } + + public String getHbaseZKParent() { + return hbaseZKParent; + } + + public void setHbaseZKParent(String hbaseZKParent) { + this.hbaseZKParent = hbaseZKParent; + } +} diff --git a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/mapping/LoadMapping.java b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/mapping/LoadMapping.java index ff342dd5..51280a3e 100644 --- a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/mapping/LoadMapping.java +++ b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/mapping/LoadMapping.java @@ -53,6 +53,12 @@ public class LoadMapping implements Checkable { private String version; @JsonProperty("structs") private List<InputStruct> structs; + @JsonProperty("backendStoreInfo") + private BackendStoreInfo backendStoreInfo; + + public BackendStoreInfo getBackendStoreInfo() { + return backendStoreInfo; + } public static LoadMapping of(String filePath) { File file = FileUtils.getFile(filePath); @@ -81,6 +87,14 @@ public class LoadMapping implements Checkable { this.structs = structs; } + @JsonCreator + public LoadMapping(@JsonProperty("structs") List<InputStruct> structs, + @JsonProperty("backendStoreInfo") BackendStoreInfo backendStoreInfo) { + this.version = Constants.V2_STRUCT_VERSION; + this.structs = structs; + this.backendStoreInfo = backendStoreInfo; + } + @Override public void check() throws IllegalArgumentException { E.checkArgument(!StringUtils.isEmpty(this.version), @@ -94,7 +108,7 @@ public class LoadMapping implements Checkable { Set<String> uniqueIds = this.structs.stream().map(InputStruct::id) .collect(Collectors.toSet()); E.checkArgument(this.structs.size() == uniqueIds.size(), - "The structs cannot contain the same id mapping"); + "The structs cannot contain the same id mapping"); } public List<InputStruct> structs() { @@ -122,7 +136,7 @@ public class LoadMapping implements Checkable { String json; try { json = FileUtils.readFileToString(failureFile.headerFile, - charset); + charset); } catch (IOException e) { throw new LoadException("Failed to read header file %s", failureFile.headerFile); @@ -143,7 +157,7 @@ public class LoadMapping implements Checkable { private Map<String, FailureFile> groupFailureFiles(File pathDir) { File[] subFiles = pathDir.listFiles(); E.checkArgument(subFiles != null && subFiles.length >= 1, - "Every input struct should have a failure data file, " + + "Every input struct should have a failure data file, " + "and a header file if need it"); Map<String, FailureFile> failureFiles = new LinkedHashMap<>(); for (File subFile : subFiles) { @@ -157,7 +171,7 @@ public class LoadMapping implements Checkable { failureFile.dataFile = subFile; } else { E.checkArgument(Constants.HEADER_SUFFIX.equals(suffix), - "The failure data file must end with %s or %s", + "The failure data file must end with %s or %s", Constants.FAILURE_SUFFIX, Constants.HEADER_SUFFIX); failureFile.headerFile = subFile; diff --git a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/metrics/LoadDistributeMetrics.java b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/metrics/LoadDistributeMetrics.java new file mode 100644 index 00000000..1f869051 --- /dev/null +++ b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/metrics/LoadDistributeMetrics.java @@ -0,0 +1,160 @@ +/* + * Copyright 2017 HugeGraph Authors + * + * 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 com.baidu.hugegraph.loader.metrics; + +import com.baidu.hugegraph.loader.constant.Constants; +import com.baidu.hugegraph.loader.mapping.EdgeMapping; +import com.baidu.hugegraph.loader.mapping.ElementMapping; +import com.baidu.hugegraph.loader.mapping.InputStruct; +import com.baidu.hugegraph.loader.mapping.VertexMapping; +import org.apache.spark.SparkContext; +import org.apache.spark.util.LongAccumulator; + +import java.io.Serializable; +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; + +public final class LoadDistributeMetrics implements Serializable { + + private final InputStruct struct; + private Map<String, Metrics> vertexDisMetrics; + private Map<String, Metrics> edgeDisMetrics; + + public LoadDistributeMetrics(InputStruct struct) { + this.struct = struct; + this.vertexDisMetrics = new HashMap<>(); + this.edgeDisMetrics = new HashMap<>(); + for (VertexMapping mapping : struct.vertices()) { + this.vertexDisMetrics.put(mapping.label(), new Metrics()); + } + for (EdgeMapping mapping : struct.edges()) { + this.edgeDisMetrics.put(mapping.label(), new Metrics()); + } + } + + public void init(SparkContext sc) { + for (VertexMapping mapping : this.struct.vertices()) { + Metrics metrics = this.vertexDisMetrics.get(mapping.label()); + metrics.insertSuccess = sc.longAccumulator(mapping.label() + + Constants.UNDERLINE_STR + Constants.LOAD_DATA_INSERT_SUFFIX); + metrics.parseSuccess = sc.longAccumulator(mapping.label() + + Constants.UNDERLINE_STR + Constants.LOAD_DATA_PARSE_SUFFIX); + + } + for (EdgeMapping mapping : this.struct.edges()) { + Metrics metrics = this.edgeDisMetrics.get(mapping.label()); + metrics.insertSuccess = sc.longAccumulator(mapping.label() + + Constants.UNDERLINE_STR + Constants.LOAD_DATA_INSERT_SUFFIX); + metrics.parseSuccess = sc.longAccumulator(mapping.label() + + Constants.UNDERLINE_STR + Constants.LOAD_DATA_PARSE_SUFFIX); + + } + } + + public void increaseDisVertexParseSuccess(ElementMapping mapping) { + + this.disMetrics(mapping).parseSuccess.add(1); + } + + public void pluseDisVertexParseSuccess(ElementMapping mapping, Long count) { + this.disMetrics(mapping).parseSuccess.add(count); + } + + public void increaseDisVertexInsertSuccess(ElementMapping mapping) { + this.disMetrics(mapping).insertSuccess.add(1); + } + + public void plusDisVertexInsertSuccess(ElementMapping mapping, Long count) { + this.disMetrics(mapping).insertSuccess.add(count); + } + + public void increaseDisEdgeParseSuccess(ElementMapping mapping) { + + this.disMetrics(mapping).parseSuccess.add(1); + } + + public void pluseDisEdgeParseSuccess(ElementMapping mapping, Long count) { + this.disMetrics(mapping).parseSuccess.add(count); + } + + public void increaseDisEdgeInsertSuccess(ElementMapping mapping) { + + this.disMetrics(mapping).insertSuccess.add(1); + } + + public void plusDisEdgeInsertSuccess(ElementMapping mapping, Long count) { + this.disMetrics(mapping).insertSuccess.add(count); + } + + public Long readVertexInsertSuccess() { + Long totalCnt = 0L; + Collection<Metrics> values = vertexDisMetrics.values(); + for (Metrics metrics : values) { + totalCnt += metrics.insertSuccess(); + } + return totalCnt; + } + + public Long readEdgeInsertSuccess() { + Long totalCnt = 0L; + Collection<Metrics> values = edgeDisMetrics.values(); + for (Metrics metrics : values) { + totalCnt += metrics.insertSuccess(); + } + return totalCnt; + } + + private Metrics disMetrics(ElementMapping mapping) { + if (mapping.type().isVertex()) { + return this.vertexDisMetrics.get(mapping.label()); + } else { + return this.edgeDisMetrics.get(mapping.label()); + } + } + + public static class Metrics implements Serializable { + + private LongAccumulator parseSuccess; + private LongAccumulator parseFailure; + private LongAccumulator insertSuccess; + private LongAccumulator insertFailure; + + public Metrics() { + + } + + public long parseSuccess() { + return this.parseSuccess.value(); + } + + public long parseFailure() { + return this.parseFailure.value(); + } + + public long insertSuccess() { + return this.insertSuccess.value(); + } + + public long insertFailure() { + return this.insertFailure.value(); + } + } +} diff --git a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java index 143dd2f0..984a9fb9 100644 --- a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java +++ b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java @@ -23,6 +23,7 @@ import com.baidu.hugegraph.driver.GraphManager; import com.baidu.hugegraph.loader.builder.EdgeBuilder; import com.baidu.hugegraph.loader.builder.ElementBuilder; import com.baidu.hugegraph.loader.builder.VertexBuilder; +import com.baidu.hugegraph.loader.direct.loader.HBaseDirectLoader; import com.baidu.hugegraph.loader.executor.LoadContext; import com.baidu.hugegraph.loader.executor.LoadOptions; import com.baidu.hugegraph.loader.mapping.EdgeMapping; @@ -30,6 +31,7 @@ import com.baidu.hugegraph.loader.mapping.ElementMapping; import com.baidu.hugegraph.loader.mapping.InputStruct; import com.baidu.hugegraph.loader.mapping.LoadMapping; import com.baidu.hugegraph.loader.mapping.VertexMapping; +import com.baidu.hugegraph.loader.metrics.LoadDistributeMetrics; import com.baidu.hugegraph.loader.source.InputSource; import com.baidu.hugegraph.loader.source.file.Compression; import com.baidu.hugegraph.loader.source.file.FileFilter; @@ -46,11 +48,14 @@ import com.baidu.hugegraph.structure.graph.UpdateStrategy; import com.baidu.hugegraph.structure.graph.Vertex; import com.baidu.hugegraph.util.Log; +import org.apache.spark.SparkConf; +import org.apache.spark.SparkContext; import org.apache.spark.sql.DataFrameReader; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; import org.apache.spark.sql.SparkSession; import org.apache.spark.sql.types.StructField; +import org.apache.spark.util.LongAccumulator; import org.slf4j.Logger; import java.io.Serializable; @@ -89,29 +94,98 @@ public class HugeGraphSparkLoader implements Serializable { public void load() { LoadMapping mapping = LoadMapping.of(this.loadOptions.file); List<InputStruct> structs = mapping.structs(); + boolean sinkType = this.loadOptions.sinkType; + if(!sinkType){ + this.loadOptions.copyBackendStoreInfo(mapping.getBackendStoreInfo()); + } + SparkConf conf = new SparkConf() + .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")// kryo序列化 + .set("spark.kryo.registrationRequired", "true"); + try { + conf.registerKryoClasses ( + new Class[] + { + org.apache.hadoop.hbase.io.ImmutableBytesWritable.class, + org.apache.hadoop.hbase.KeyValue.class, + org.apache.spark.sql.types.StructType.class, + StructField[].class, + StructField.class, + org.apache.spark.sql.types.LongType$.class, + org.apache.spark.sql.types.Metadata.class, + org.apache.spark.sql.types.StringType$.class, + Class.forName( + "org.apache.spark.internal.io.FileCommitProtocol$TaskCommitMessage"), + Class.forName("scala.reflect.ClassTag$$anon$1"), + Class.forName("scala.collection.immutable.Set$EmptySet$"), + Class.forName("org.apache.spark.sql.types.DoubleType$") + }); + } catch (ClassNotFoundException e) { + LOG.error("spark kryo serialized registration failed"); + } + SparkSession session = SparkSession.builder() + .config(conf) + .getOrCreate(); + SparkContext sc = session.sparkContext(); - SparkSession session = SparkSession.builder().getOrCreate(); + LongAccumulator totalInsertSuccess = sc.longAccumulator("totalInsertSuccess"); for (InputStruct struct : structs) { + LOG.info("\n init" + struct.input().asFileSource().path() + + " distribute metrics---- \n"); + LoadDistributeMetrics loadDistributeMetrics = new LoadDistributeMetrics(struct); + loadDistributeMetrics.init(sc); + LOG.info("\n load dat info: \t" + struct.input().asFileSource().path() + + "\n start load data ; \n"); Dataset<Row> ds = read(session, struct); - ds.foreachPartition((Iterator<Row> p) -> { - LoadContext context = initPartition(this.loadOptions, struct); - p.forEachRemaining((Row row) -> { - loadRow(struct, row, p, context); + if (sinkType) { + LOG.info("\n ------ spark api start load data ------ \n"); + ds.foreachPartition((Iterator<Row> p) -> { + LoadContext context = initPartition(this.loadOptions, struct); + p.forEachRemaining((Row row) -> { + loadRow(struct, row, p, context); + }); + context.close(); }); - context.close(); - }); + + } else { + LOG.info("\n spark bulkload gen hfile start \n"); + // gen-hfile + HBaseDirectLoader directLoader = new HBaseDirectLoader(loadOptions, + struct,loadDistributeMetrics); + directLoader.bulkload(ds); + + } + collectLoadMetrics(loadDistributeMetrics,totalInsertSuccess); + LOG.info(" \n load data info : \t" + struct.input().asFileSource().path() + + "\n load data finish!!!; \n"); } + Long totalInsertSuccessCnt = totalInsertSuccess.value(); + LOG.info("\n ------------The data import task is complete-------------------\n" + + "\n insertSuccess cnt:\t" + totalInsertSuccess + " \n" + + "\n ---------------------------------------------\n" + ); + + sc.stop(); session.close(); session.stop(); + + } + + private void collectLoadMetrics (LoadDistributeMetrics loadMetrics, + LongAccumulator totalInsertSuccess) { + + Long edgeInsertSuccess = loadMetrics.readEdgeInsertSuccess(); + Long vertexInsertSuccess = loadMetrics.readVertexInsertSuccess(); + totalInsertSuccess.add(edgeInsertSuccess); + totalInsertSuccess.add(vertexInsertSuccess); + } private LoadContext initPartition( LoadOptions loadOptions, InputStruct struct) { LoadContext context = new LoadContext(loadOptions); for (VertexMapping vertexMapping : struct.vertices()) { - this.builders.put( - new VertexBuilder(context, struct, vertexMapping), - new ArrayList<>()); + this.builders.put(new VertexBuilder(context, struct, vertexMapping), + new ArrayList<>()); } for (EdgeMapping edgeMapping : struct.edges()) { this.builders.put(new EdgeBuilder(context, struct, edgeMapping), diff --git a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/struct/GraphStructV1.java b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/struct/GraphStructV1.java index ec66497a..4a8081d8 100644 --- a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/struct/GraphStructV1.java +++ b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/struct/GraphStructV1.java @@ -26,6 +26,7 @@ import java.util.List; import java.util.Set; import java.util.stream.Collectors; +import com.baidu.hugegraph.loader.mapping.BackendStoreInfo; import org.apache.commons.collections.ListUtils; import org.apache.commons.io.FileUtils; import org.slf4j.Logger; @@ -56,6 +57,9 @@ public class GraphStructV1 implements Checkable { @JsonProperty("edges") private final List<EdgeStructV1> edgeStructs; + @JsonProperty("backendStoreInfo") + private BackendStoreInfo backendStoreInfo; + public GraphStructV1() { this.vertexStructs = new ArrayList<>(); this.edgeStructs = new ArrayList<>(); @@ -76,6 +80,10 @@ public class GraphStructV1 implements Checkable { } } + public BackendStoreInfo getBackendStoreInfo() { + return backendStoreInfo; + } + @Override public void check() throws IllegalArgumentException { LOG.info("Checking vertex mapping descriptions"); diff --git a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/util/MappingUtil.java b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/util/MappingUtil.java index f6c9fe52..a749d842 100644 --- a/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/util/MappingUtil.java +++ b/hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/util/MappingUtil.java @@ -129,7 +129,7 @@ public final class MappingUtil { inputStruct.id(String.valueOf(++id)); inputStructs.add(inputStruct); } - return new LoadMapping(inputStructs); + return new LoadMapping(inputStructs,graphStruct.getBackendStoreInfo()); } private static ElementMapping convertV1ToV2(ElementStructV1 origin) { diff --git a/pom.xml b/pom.xml index e1f2eebe..09431d12 100644 --- a/pom.xml +++ b/pom.xml @@ -80,6 +80,7 @@ <docker.hub>hugegraph</docker.hub> <docker.repo>${project.name}</docker.repo> <docker.tag>${project.version}</docker.tag> + <hbase.version>2.2.3</hbase.version> </properties> <dependencies>
