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 d0b4ad994d5cb4c54f47aaf9c07084dbc5c5498b Merge: 964ac5b1 a622f986 Author: imbajin <[email protected]> AuthorDate: Thu Nov 10 16:51:43 2022 +0800 Merge branch 'master' into client-rename .../loader/direct/loader/DirectLoader.java | 76 ++++++ .../loader/direct/loader/HBaseDirectLoader.java | 268 +++++++++++++++++++++ .../hugegraph/loader/direct/util/SinkToHBase.java | 165 +++++++++++++ .../hugegraph/loader/executor/LoadOptions.java | 2 +- .../hugegraph/loader/mapping/LoadMapping.java | 2 +- .../loader/spark/HugeGraphSparkLoader.java | 49 ++-- .../apache/hugegraph/loader/util/MappingUtil.java | 2 +- .../loader/test/unit/MappingConverterTest.java | 18 +- pom.xml | 6 +- 9 files changed, 554 insertions(+), 34 deletions(-) diff --cc hugegraph-loader/src/main/java/org/apache/hugegraph/loader/executor/LoadOptions.java index 0d6d5317,00000000..162fd157 mode 100644,000000..100644 --- a/hugegraph-loader/src/main/java/org/apache/hugegraph/loader/executor/LoadOptions.java +++ b/hugegraph-loader/src/main/java/org/apache/hugegraph/loader/executor/LoadOptions.java @@@ -1,373 -1,0 +1,373 @@@ +/* + * 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 org.apache.hugegraph.loader.executor; + +import java.io.File; +import java.io.Serializable; +import java.util.Set; + +import org.apache.hugegraph.loader.util.LoadUtil; +import org.apache.hugegraph.loader.mapping.BackendStoreInfo; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; + +import org.apache.hugegraph.loader.constant.Constants; +import org.apache.hugegraph.util.E; +import org.apache.hugegraph.util.Log; +import com.beust.jcommander.IParameterValidator; +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import com.beust.jcommander.ParameterException; +import com.google.common.collect.ImmutableSet; + +public class LoadOptions implements Serializable { + + private static final Logger LOG = Log.logger(LoadOptions.class); + + public static final String HTTPS_SCHEMA = "https"; + 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") + public String file; + + @Parameter(names = {"-s", "--schema"}, arity = 1, + 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") + public String graph; + + @Parameter(names = {"-h", "--host"}, arity = 1, + 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") + public int port = 8080; + + @Parameter(names = {"--username"}, arity = 1, + 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") + public String protocol = "http"; + + @Parameter(names = {"--trust-store-file"}, arity = 1, + 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") + public String trustStoreToken = null; + + @Parameter(names = {"--token"}, arity = 1, + 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") + public boolean clearAllData = false; + + @Parameter(names = {"--clear-timeout"}, arity = 1, + 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") + 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") + public boolean failureMode = false; + + @Parameter(names = {"--batch-insert-threads"}, arity = 1, + 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") + public int singleInsertThreads = 8; + + @Parameter(names = {"--max-conn"}, arity = 1, + 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") + public int maxConnectionsPerRoute = CPUS * 2; + + @Parameter(names = {"--batch-size"}, arity = 1, + 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") + public int flushIntervalMs = 30000; + + @Parameter(names = {"--cdc-sink-parallelism"}, arity = 1, + 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") + public int shutdownTimeout = 10; + + @Parameter(names = {"--check-vertex"}, arity = 1, + 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") + 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") + 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") + public int maxInsertErrors = 500; + + @Parameter(names = {"--timeout"}, arity = 1, + 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") + public int retryTimes = 3; + + @Parameter(names = {"--retry-interval"}, arity = 1, + 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") + public long maxReadLines = -1L; + + @Parameter(names = {"--dry-run"}, arity = 1, + 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") + public boolean printProgress = true; + + @Parameter(names = {"--test-mode"}, arity = 1, + description = "Whether the hugegraph-loader work in test mode") + public boolean testMode = false; + + @Parameter(names = {"--help"}, help = true, + 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"; + } else if (this.failureMode) { + return "FAILURE MODE"; + } else { + return "NORMAL MODE"; + } + } + + public static LoadOptions parseOptions(String[] args) { + LoadOptions options = new LoadOptions(); + JCommander commander = JCommander.newBuilder() + .addObject(options) + .build(); + commander.parse(args); + // Print usage and exit + if (options.help) { + LoadUtil.exitWithUsage(commander, Constants.EXIT_CODE_NORM); + } + // Check options + // Check option "-f" + E.checkArgument(!StringUtils.isEmpty(options.file), + "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); + File mappingFile = new File(options.file); + if (!mappingFile.canRead()) { + LOG.error("The mapping file must be readable: '{}'", mappingFile); + LoadUtil.exitWithUsage(commander, Constants.EXIT_CODE_ERROR); + } + + // Check option "-g" + E.checkArgument(!StringUtils.isEmpty(options.graph), + "The graph must be specified"); + // Check option "-h" + if (!options.host.startsWith(Constants.HTTP_PREFIX)) { + if (options.protocol.equals(HTTP_SCHEMA)) { + options.host = Constants.HTTP_PREFIX + options.host; + } else { + options.host = Constants.HTTPS_PREFIX + options.host; + } + } + // Check option --incremental-mode and --failure-mode + E.checkArgument(!(options.incrementalMode && options.failureMode), + "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"); + options.maxReadErrors = Constants.NO_LIMIT; + options.maxParseErrors = Constants.NO_LIMIT; + options.maxInsertErrors = Constants.NO_LIMIT; + } + return options; + } + + public void copyBackendStoreInfo (BackendStoreInfo backendStoreInfo) { - E.checkArgument(null != backendStoreInfo,"the backendStoreInfo is null "); ++ E.checkArgument(null != backendStoreInfo, "The backendStoreInfo can't be 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_!~*'()-]+)$"; + if (!value.matches(regex)) { + throw new ParameterException(String.format( + "Invalid url value of args '%s': '%s'", name, value)); + } + } + } + + public static class ProtocolValidator implements IParameterValidator { + + private static final Set<String> SSL_PROTOCOL = ImmutableSet.of( + HTTP_SCHEMA, HTTPS_SCHEMA + ); + + @Override + 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)); + } + } + } + + public static class DirectoryValidator implements IParameterValidator { + + @Override + public void validate(String name, String value) { + 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)); + } + } + } + + public static class FileValidator implements IParameterValidator { + + @Override + public void validate(String name, String value) { + 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)); + } + } + } + + public static class PositiveValidator implements IParameterValidator { + + @Override + public void validate(String name, String value) { + int retry = Integer.parseInt(value); + if (retry <= 0) { + throw new ParameterException(String.format( + "Parameter '%s' should be positive, but got '%s'", + name, value)); + } + } + } +} diff --cc hugegraph-loader/src/main/java/org/apache/hugegraph/loader/mapping/LoadMapping.java index a83fd835,00000000..7a403d25 mode 100644,000000..100644 --- a/hugegraph-loader/src/main/java/org/apache/hugegraph/loader/mapping/LoadMapping.java +++ b/hugegraph-loader/src/main/java/org/apache/hugegraph/loader/mapping/LoadMapping.java @@@ -1,201 -1,0 +1,201 @@@ +/* + * 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 org.apache.hugegraph.loader.mapping; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.io.FileUtils; +import org.apache.commons.lang3.StringUtils; + +import org.apache.hugegraph.loader.constant.Checkable; +import org.apache.hugegraph.loader.constant.Constants; +import org.apache.hugegraph.loader.exception.LoadException; +import org.apache.hugegraph.loader.executor.LoadOptions; +import org.apache.hugegraph.loader.util.JsonUtil; +import org.apache.hugegraph.loader.util.LoadUtil; +import org.apache.hugegraph.loader.util.MappingUtil; +import org.apache.hugegraph.loader.source.file.FileSource; +import org.apache.hugegraph.util.E; +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.annotation.JsonPropertyOrder; + +@JsonPropertyOrder({"version", "structs"}) +public class LoadMapping implements Checkable { + + @JsonProperty("version") + 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); + LoadMapping mapping; + try { + String json = FileUtils.readFileToString(file, Constants.CHARSET); + mapping = MappingUtil.parse(json); + } catch (IOException e) { + throw new LoadException("Failed to read mapping mapping file '%s'", + e, filePath); + } catch (IllegalArgumentException e) { + throw new LoadException("Failed to parse mapping mapping file '%s'", + e, filePath); + } + try { + mapping.check(); + } catch (IllegalArgumentException e) { + throw new LoadException("Invalid mapping file '%s'", e, filePath); + } + return mapping; + } + + @JsonCreator + public LoadMapping(@JsonProperty("structs") List<InputStruct> structs) { + this.version = Constants.V2_STRUCT_VERSION; + this.structs = structs; + } + + @JsonCreator + public LoadMapping(@JsonProperty("structs") List<InputStruct> structs, - @JsonProperty("backendStoreInfo") BackendStoreInfo backendStoreInfo) { ++ @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), + "The version can't be null or empty"); + E.checkArgument(this.version.equals(Constants.V2_STRUCT_VERSION), + "The version must be '%s', but got '%s'", + Constants.V2_STRUCT_VERSION, this.version); + E.checkArgument(!CollectionUtils.isEmpty(this.structs), + "The structs can't be null or empty"); + this.structs.forEach(InputStruct::check); + 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"); + } + + public List<InputStruct> structs() { + return this.structs; + } + + public List<InputStruct> structsForFailure(LoadOptions options) { + List<InputStruct> targetStructs = new ArrayList<>(); + String dir = LoadUtil.getStructDirPrefix(options); + String path = Paths.get(dir, Constants.FAILURE_DATA).toString(); + File pathDir = FileUtils.getFile(path); + // It means no failure data if the path directory does not exist + if (!pathDir.exists()) { + return targetStructs; + } + Map<String, FailureFile> failureFiles = this.groupFailureFiles(pathDir); + for (String inputId : failureFiles.keySet()) { + InputStruct struct = this.struct(inputId); + String charset = struct.input().charset(); + FailureFile failureFile = failureFiles.get(inputId); + + FileSource source = struct.input().asFileSource(); + if (failureFile.headerFile != null) { + // It means that header file existed + String json; + try { + json = FileUtils.readFileToString(failureFile.headerFile, + charset); + } catch (IOException e) { + throw new LoadException("Failed to read header file %s", + failureFile.headerFile); + } + List<String> header = JsonUtil.convertList(json, String.class); + source.header(header.toArray(new String[] {})); + } + // Set failure data path + source.path(failureFile.dataFile.getAbsolutePath()); + source.skippedLine().regex(Constants.SKIPPED_LINE_REGEX); + struct.input(source); + // Add to target structs + targetStructs.add(struct); + } + return targetStructs; + } + + 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, " + + "and a header file if need it"); + Map<String, FailureFile> failureFiles = new LinkedHashMap<>(); + for (File subFile : subFiles) { + String inputId = LoadUtil.getFileNamePrefix(subFile); + String suffix = LoadUtil.getFileNameSuffix(subFile); + FailureFile failureFile = failureFiles.get(inputId); + if (failureFile == null) { + failureFile = new FailureFile(); + } + if (Constants.FAILURE_SUFFIX.equals(suffix)) { + failureFile.dataFile = subFile; + } else { + E.checkArgument(Constants.HEADER_SUFFIX.equals(suffix), + "The failure data file must end with %s or %s", + Constants.FAILURE_SUFFIX, + Constants.HEADER_SUFFIX); + failureFile.headerFile = subFile; + } + failureFiles.put(inputId, failureFile); + } + return failureFiles; + } + + public InputStruct struct(String id) { + for (InputStruct struct : this.structs) { + if (struct.id().equals(id)) { + return struct; + } + } + throw new IllegalArgumentException(String.format( + "There is no input struct with id '%s'", id)); + } + + private static class FailureFile { + + // Maybe null + private File headerFile; + // Can't be null + private File dataFile; + } +} diff --cc hugegraph-loader/src/main/java/org/apache/hugegraph/loader/spark/HugeGraphSparkLoader.java index 5a8c2a46,00000000..0a7d1f63 mode 100644,000000..100644 --- a/hugegraph-loader/src/main/java/org/apache/hugegraph/loader/spark/HugeGraphSparkLoader.java +++ b/hugegraph-loader/src/main/java/org/apache/hugegraph/loader/spark/HugeGraphSparkLoader.java @@@ -1,338 -1,0 +1,335 @@@ +/* + * 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 org.apache.hugegraph.loader.spark; + +import org.apache.hugegraph.driver.GraphManager; +import org.apache.hugegraph.loader.builder.EdgeBuilder; +import org.apache.hugegraph.loader.builder.ElementBuilder; +import org.apache.hugegraph.loader.builder.VertexBuilder; +import org.apache.hugegraph.loader.direct.loader.HBaseDirectLoader; +import org.apache.hugegraph.loader.executor.LoadContext; +import org.apache.hugegraph.loader.executor.LoadOptions; +import org.apache.hugegraph.loader.metrics.LoadDistributeMetrics; +import org.apache.hugegraph.loader.source.InputSource; +import org.apache.hugegraph.loader.source.SourceType; +import org.apache.hugegraph.loader.source.jdbc.JDBCSource; +import org.apache.hugegraph.loader.util.Printer; +import org.apache.hugegraph.loader.mapping.EdgeMapping; +import org.apache.hugegraph.loader.mapping.ElementMapping; +import org.apache.hugegraph.loader.mapping.InputStruct; +import org.apache.hugegraph.loader.mapping.LoadMapping; +import org.apache.hugegraph.loader.mapping.VertexMapping; +import org.apache.hugegraph.loader.source.file.Compression; +import org.apache.hugegraph.loader.source.file.FileFilter; +import org.apache.hugegraph.loader.source.file.FileFormat; +import org.apache.hugegraph.loader.source.file.FileSource; +import org.apache.hugegraph.loader.source.file.SkippedLine; +import org.apache.hugegraph.structure.GraphElement; +import org.apache.hugegraph.structure.graph.BatchEdgeRequest; +import org.apache.hugegraph.structure.graph.BatchVertexRequest; +import org.apache.hugegraph.structure.graph.Edge; +import org.apache.hugegraph.structure.graph.UpdateStrategy; +import org.apache.hugegraph.structure.graph.Vertex; +import org.apache.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; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +import scala.collection.JavaConverters; + +public class HugeGraphSparkLoader implements Serializable { + + public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class); + + private final LoadOptions loadOptions; + private final Map<ElementBuilder, List<GraphElement>> builders; + + public static void main(String[] args) { + HugeGraphSparkLoader loader; + try { + loader = new HugeGraphSparkLoader(args); + } catch (Throwable e) { + Printer.printError("Failed to start loading", e); + return; + } + loader.load(); + } + + public HugeGraphSparkLoader(String[] args) { + this.loadOptions = LoadOptions.parseOptions(args); + this.builders = new HashMap<>(); + } + + public void load() { + LoadMapping mapping = LoadMapping.of(this.loadOptions.file); + List<InputStruct> structs = mapping.structs(); + boolean sinkType = this.loadOptions.sinkType; - if(!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$") - }); ++ 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(); + + LongAccumulator totalInsertSuccess = sc.longAccumulator("totalInsertSuccess"); + for (InputStruct struct : structs) { + LOG.info("\n Initializes the accumulator corresponding to the {} ", - struct.input().asFileSource().path()); ++ struct.input().asFileSource().path()); + LoadDistributeMetrics loadDistributeMetrics = new LoadDistributeMetrics(struct); + loadDistributeMetrics.init(sc); - LOG.info("\n Start to load data, data info is: \t {} ", - struct.input().asFileSource().path()); ++ LOG.info("\n Start to load data, data info is: \t {} ", ++ struct.input().asFileSource().path()); + Dataset<Row> ds = read(session, struct); + if (sinkType) { + LOG.info("\n Start to load data using spark apis \n"); + ds.foreachPartition((Iterator<Row> p) -> { + LoadContext context = initPartition(this.loadOptions, struct); + p.forEachRemaining((Row row) -> { + loadRow(struct, row, p, context); + }); + context.close(); + }); + + } else { + LOG.info("\n Start to load data using spark bulkload \n"); + // gen-hfile + HBaseDirectLoader directLoader = new HBaseDirectLoader(loadOptions, + struct,loadDistributeMetrics); + directLoader.bulkload(ds); + + } + collectLoadMetrics(loadDistributeMetrics,totalInsertSuccess); - LOG.info(" \n Finished load {} data ", - struct.input().asFileSource().path()); ++ LOG.info("\n Finished load {} data ", ++ struct.input().asFileSource().path()); + } + Long totalInsertSuccessCnt = totalInsertSuccess.value(); + LOG.info("\n ------------The data load task is complete-------------------\n" + - "\n insertSuccesscnt:\t {}" + - "\n ---------------------------------------------\n" - , totalInsertSuccessCnt); ++ "\n insertSuccesscnt:\t {}" + ++ "\n ---------------------------------------------\n" ++ , totalInsertSuccessCnt); + + 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<>()); + } + for (EdgeMapping edgeMapping : struct.edges()) { + this.builders.put(new EdgeBuilder(context, struct, edgeMapping), + new ArrayList<>()); + } + context.updateSchemaCache(); + return context; + } + + private void loadRow(InputStruct struct, Row row, Iterator<Row> p, + LoadContext context) { + for (Map.Entry<ElementBuilder, List<GraphElement>> builderMap : + this.builders.entrySet()) { + ElementMapping elementMapping = builderMap.getKey().mapping(); + // Parse + if (elementMapping.skip()) { + continue; + } + parse(row, builderMap, struct); + + // Insert + List<GraphElement> graphElements = builderMap.getValue(); + if (graphElements.size() >= elementMapping.batchSize() || + (!p.hasNext() && graphElements.size() > 0)) { + flush(builderMap, context.client().graph(), this.loadOptions.checkVertex); + } + } + } + + private Dataset<Row> read(SparkSession ss, InputStruct struct) { + InputSource input = struct.input(); + String charset = input.charset(); + DataFrameReader reader = ss.read(); + Dataset<Row> ds; + switch (input.type()) { + case FILE: + case HDFS: + FileSource fileSource = input.asFileSource(); + String[] header = fileSource.header(); + String delimiter = fileSource.delimiter(); + String path = fileSource.path(); + FileFilter filter = fileSource.filter(); + FileFormat format = fileSource.format(); + String dateFormat = fileSource.dateFormat(); + String timeZone = fileSource.timeZone(); + SkippedLine skippedLine = fileSource.skippedLine(); + Compression compression = fileSource.compression(); + int batchSize = fileSource.batchSize(); + switch (format) { + case TEXT: + ds = reader.text(path); + break; + case JSON: + ds = reader.json(path); + break; + case CSV: + ds = reader.csv(path); + break; + default: + throw new IllegalStateException( + "Unexpected format value: " + format); + } + break; + case JDBC: + JDBCSource jdbcSource = (JDBCSource) struct.input(); + String url = jdbcSource.url() + "/" + jdbcSource.database(); + String table = jdbcSource.table(); + String username = jdbcSource.username(); + String password = jdbcSource.password(); + Properties properties = new Properties(); + properties.put("user", username); + properties.put("password", password); + ds = reader.jdbc(url, table, properties); + break; + default: + throw new AssertionError(String.format( + "Unsupported input source '%s'", input.type())); + } + return ds; + } + + private void parse(Row row, + Map.Entry<ElementBuilder, List<GraphElement>> builderMap, + InputStruct struct) { + ElementBuilder builder = builderMap.getKey(); + List<GraphElement> graphElements = builderMap.getValue(); + if ("".equals(row.mkString())) { + return; + } + List<GraphElement> elements; + switch (struct.input().type()) { + case FILE: + case HDFS: + FileSource fileSource = struct.input().asFileSource(); + String delimiter = fileSource.delimiter(); + elements = builder.build(fileSource.header(), + row.mkString(delimiter).split(delimiter)); + break; + case JDBC: + Object[] structFields = JavaConverters.asJavaCollection(row.schema().toList()) + .toArray(); + int len = row.schema().length(); + String[] headers = new String[len]; + Object[] values = new Object[len]; + for (int i = 0; i < len; i++) { + headers[i] = ((StructField) structFields[i]).name(); + values[i] = row.get(i); + } + elements = builder.build(headers, values); + break; + default: + throw new AssertionError(String.format( + "Unsupported input source '%s'", + struct.input().type())); + } + graphElements.addAll(elements); + } + + private void flush(Map.Entry<ElementBuilder, List<GraphElement>> builderMap, + GraphManager g, boolean isCheckVertex) { + ElementBuilder builder = builderMap.getKey(); + ElementMapping elementMapping = builder.mapping(); + List<GraphElement> graphElements = builderMap.getValue(); + boolean isVertex = builder.mapping().type().isVertex(); + Map<String, UpdateStrategy> updateStrategyMap = elementMapping.updateStrategies(); + if (updateStrategyMap.isEmpty()) { + if (isVertex) { + g.addVertices((List<Vertex>) (Object) graphElements); + } else { + g.addEdges((List<Edge>) (Object) graphElements); + } + } else { + // CreateIfNotExist dose not support false now + if (isVertex) { + BatchVertexRequest.Builder req = + new BatchVertexRequest.Builder(); + req.vertices((List<Vertex>) (Object) graphElements) + .updatingStrategies(updateStrategyMap) + .createIfNotExist(true); + g.updateVertices(req.build()); + } else { + BatchEdgeRequest.Builder req = new BatchEdgeRequest.Builder(); + req.edges((List<Edge>) (Object) graphElements) + .updatingStrategies(updateStrategyMap) + .checkVertex(isCheckVertex) + .createIfNotExist(true); + g.updateEdges(req.build()); + } + } + graphElements.clear(); + } +} diff --cc hugegraph-loader/src/main/java/org/apache/hugegraph/loader/util/MappingUtil.java index 5bd405e4,00000000..47e7bf06 mode 100644,000000..100644 --- a/hugegraph-loader/src/main/java/org/apache/hugegraph/loader/util/MappingUtil.java +++ b/hugegraph-loader/src/main/java/org/apache/hugegraph/loader/util/MappingUtil.java @@@ -1,188 -1,0 +1,188 @@@ +/* + * 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 org.apache.hugegraph.loader.util; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.commons.io.FileUtils; + +import org.apache.hugegraph.loader.constant.Constants; +import org.apache.hugegraph.loader.exception.LoadException; +import org.apache.hugegraph.loader.mapping.EdgeMapping; +import org.apache.hugegraph.loader.mapping.ElementMapping; +import org.apache.hugegraph.loader.mapping.InputStruct; +import org.apache.hugegraph.loader.mapping.LoadMapping; +import org.apache.hugegraph.loader.mapping.VertexMapping; +import org.apache.hugegraph.loader.source.InputSource; +import org.apache.hugegraph.loader.source.SourceType; +import org.apache.hugegraph.loader.source.file.FileSource; +import org.apache.hugegraph.loader.struct.EdgeStructV1; +import org.apache.hugegraph.loader.struct.ElementStructV1; +import org.apache.hugegraph.loader.struct.GraphStructV1; +import org.apache.hugegraph.loader.struct.VertexStructV1; +import org.apache.hugegraph.util.E; +import org.apache.hugegraph.util.InsertionOrderUtil;; +import com.google.common.collect.ImmutableSet; + +@SuppressWarnings("deprecation") +public final class MappingUtil { + + private static final Set<String> ACCEPTABLE_VERSIONS = ImmutableSet.of( + Constants.V1_STRUCT_VERSION, Constants.V2_STRUCT_VERSION + ); + + public static void write(LoadMapping mapping, String path) { + File file = FileUtils.getFile(path); + String json = JsonUtil.toJson(mapping); + try { + FileUtils.write(file, json, Constants.CHARSET); + } catch (IOException e) { + throw new LoadException("Failed to write mapping %s to file '%s'", + e, mapping, file); + } + } + + public static LoadMapping parse(String json) { + Map<String, Object> map = JsonUtil.convertMap(json, String.class, + Object.class); + Object value = map.get(Constants.FIELD_VERSION); + if (value == null) { + value = Constants.V1_STRUCT_VERSION; + } else { + E.checkArgument(value instanceof String, + "The version value must be String class, " + + "but got '%s(%s)'", value, value.getClass()); + } + String version = (String) value; + + E.checkArgument(ACCEPTABLE_VERSIONS.contains(version), + "Invalid version '%s', the acceptable versions are %s", + version, ACCEPTABLE_VERSIONS); + if (version.equals(Constants.V2_STRUCT_VERSION)) { + return JsonUtil.fromJson(json, LoadMapping.class); + } else { + assert version.equals(Constants.V1_STRUCT_VERSION); + return parseV1(json); + } + } + + private static LoadMapping parseV1(String json) { + GraphStructV1 graphStruct = JsonUtil.fromJson(json, + GraphStructV1.class); + Map<FileSourceKey, InputStruct> fileSourceInputStructs = + InsertionOrderUtil.newMap(); + List<InputStruct> jdbcSourceInputStructs = new ArrayList<>(); + for (ElementStructV1 originStruct : graphStruct.structs()) { + InputSource inputSource = originStruct.input(); + ElementMapping targetStruct = convertV1ToV2(originStruct); + + SourceType type = inputSource.type(); + if (type == SourceType.FILE || type == SourceType.HDFS) { + FileSource source = (FileSource) inputSource; + FileSourceKey key = new FileSourceKey(type, source.path()); + fileSourceInputStructs.compute(key, (k, inputStruct) -> { + if (inputStruct == null) { + inputStruct = new InputStruct(null, null); + inputStruct.input(source); + } + inputStruct.add(targetStruct); + return inputStruct; + }); + } else { + assert type == SourceType.JDBC; + InputStruct inputStruct = new InputStruct(null, null); + inputStruct.input(inputSource); + inputStruct.add(targetStruct); + jdbcSourceInputStructs.add(inputStruct); + } + } + // Generate id for every input mapping + List<InputStruct> inputStructs = new ArrayList<>(); + int id = 0; + for (InputStruct inputStruct : fileSourceInputStructs.values()) { + inputStruct.id(String.valueOf(++id)); + inputStructs.add(inputStruct); + } + for (InputStruct inputStruct : jdbcSourceInputStructs) { + inputStruct.id(String.valueOf(++id)); + inputStructs.add(inputStruct); + } - return new LoadMapping(inputStructs,graphStruct.getBackendStoreInfo()); ++ return new LoadMapping(inputStructs, graphStruct.getBackendStoreInfo()); + } + + private static ElementMapping convertV1ToV2(ElementStructV1 origin) { + ElementMapping target; + if (origin.type().isVertex()) { + VertexStructV1 originVertex = (VertexStructV1) origin; + target = new VertexMapping(originVertex.idField(), + originVertex.unfold()); + } else { + EdgeStructV1 originEdge = (EdgeStructV1) origin; + target = new EdgeMapping(originEdge.sourceFields(), + originEdge.unfoldSource(), + originEdge.targetFields(), + originEdge.unfoldTarget()); + } + fill(origin, target); + return target; + } + + private static void fill(ElementStructV1 originStruct, + ElementMapping targetStruct) { + targetStruct.label(originStruct.label()); + targetStruct.skip(originStruct.skip()); + targetStruct.mappingFields(originStruct.mappingFields()); + targetStruct.mappingValues(originStruct.mappingValues()); + targetStruct.selectedFields(originStruct.selectedFields()); + targetStruct.ignoredFields(originStruct.ignoredFields()); + targetStruct.nullValues(originStruct.nullValues()); + targetStruct.updateStrategies(originStruct.updateStrategies()); + } + + private static class FileSourceKey { + + private final SourceType type; + private final String path; + + public FileSourceKey(SourceType type, String path) { + this.type = type; + this.path = path; + } + + @Override + public boolean equals(Object object) { + if (!(object instanceof FileSourceKey)) { + return false; + } + FileSourceKey other = (FileSourceKey) object; + return this.type == other.type && this.path.equals(other.path); + } + + @Override + public int hashCode() { + return this.type.hashCode() ^ this.path.hashCode(); + } + } +} diff --cc hugegraph-loader/src/test/java/org/apache/hugegraph/loader/test/unit/MappingConverterTest.java index 7f1dee26,00000000..fc71c74c mode 100644,000000..100644 --- a/hugegraph-loader/src/test/java/org/apache/hugegraph/loader/test/unit/MappingConverterTest.java +++ b/hugegraph-loader/src/test/java/org/apache/hugegraph/loader/test/unit/MappingConverterTest.java @@@ -1,125 -1,0 +1,139 @@@ +/* + * 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 org.apache.hugegraph.loader.test.unit; + +import java.io.File; +import java.io.IOException; +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; + +import org.apache.commons.io.FileUtils; +import org.apache.hugegraph.loader.MappingConverter; +import org.junit.Assert; +import org.junit.Test; + +public class MappingConverterTest { + + @Test + public void testConvertV1ToV2() throws IOException { + String v1Json = "{" + + " \"vertices\": [" + + " {" + + " \"label\": \"user\"," + + " \"input\": {" + + " \"type\": \"file\"," + + " \"path\": \"users.dat\"," + + " \"format\": \"TEXT\"," + + " \"delimiter\": \"::\"," + + " \"header\": [\"UserID\", \"Gender\", \"Age\", " + + "\"Occupation\", \"Zip-code\"]" + + " }," + + " \"ignored\": [\"Gender\", \"Age\", \"Occupation\", " + + "\"Zip-code\"]," + + " \"field_mapping\": {" + + " \"UserID\": \"id\"" + + " }" + + " }" + + " ]," + + " \"edges\": [" + + " {" + + " \"label\": \"rating\"," + + " \"source\": [\"UserID\"]," + + " \"target\": [\"MovieID\"]," + + " \"input\": {" + + " \"type\": \"file\"," + + " \"path\": \"ratings.dat\"," + + " \"format\": \"TEXT\"," + + " \"delimiter\": \"::\"," + + " \"header\": [\"UserID\", \"MovieID\", \"Rating\", " + + "\"Timestamp\"]" + + " }," + + " \"ignored\": [\"Timestamp\"]," + + " \"field_mapping\": {" + + " \"UserID\": \"id\"," + + " \"MovieID\": \"id\"," + + " \"Rating\": \"rate\"" + + " }" + + " }" + - " ]" + ++ " ]," + ++ " \"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\"" + ++ " }" + + "}"; + String input = "struct.json"; + File inputFile = new File(input); + Charset charset = StandardCharsets.UTF_8; + FileUtils.writeStringToFile(inputFile, v1Json, charset); + MappingConverter.main(new String[]{input}); + + File outputFile = FileUtils.getFile("struct-v2.json"); + String actualV2Json = FileUtils.readFileToString(outputFile, charset); + String expectV2Json = "{\"version\":\"2.0\"," + + "\"structs\":[{\"id\":\"1\",\"skip\":false," + + "\"input\":{\"type\":\"FILE\",\"path\":\"users.dat\"," + + "\"file_filter\":{\"extensions\":[\"*\"]}," + + "\"format\":\"TEXT\",\"delimiter\":\"::\"," + + "\"date_format\":\"yyyy-MM-dd HH:mm:ss\"," + + "\"time_zone\":\"GMT+8\",\"skipped_line\":{\"regex\":\"" + + "(^#|^//).*|\"},\"compression\":\"NONE\"," + + "\"batch_size\":500,\"header\":[\"UserID\",\"Gender\"," + + "\"Age\",\"Occupation\",\"Zip-code\"]," + + "\"charset\":\"UTF-8\",\"list_format\":null}," + + "\"vertices\":[{\"label\":\"user\",\"skip\":false," + + "\"id\":null,\"unfold\":false," + + "\"field_mapping\":{\"UserID\":\"id\"}," + + "\"value_mapping\":{},\"selected\":[]," + + "\"ignored\":[\"Occupation\",\"Zip-code\",\"Gender\"," + + "\"Age\"],\"null_values\":[\"\"]," + + "\"update_strategies\":{},\"batch_size\":500}],\"edges\":[]},{\"id\":\"2\"," + + "\"skip\":false,\"input\":{\"type\":\"FILE\"," + + "\"path\":\"ratings.dat\"," + + "\"file_filter\":{\"extensions\":[\"*\"]}," + + "\"format\":\"TEXT\",\"delimiter\":\"::\"," + + "\"date_format\":\"yyyy-MM-dd HH:mm:ss\"," + + "\"time_zone\":\"GMT+8\",\"skipped_line\":{\"regex\":\"" + + "(^#|^//).*|\"},\"compression\":\"NONE\"," + + "\"batch_size\":500,\"header\":[\"UserID\",\"MovieID\"," + + "\"Rating\",\"Timestamp\"],\"charset\":\"UTF-8\"," + + "\"list_format\":null},\"vertices\":[]," + + "\"edges\":[{\"label\":\"rating\",\"skip\":false," + + "\"source\":[\"UserID\"],\"unfold_source\":false," + + "\"target\":[\"MovieID\"],\"unfold_target\":false," + + "\"field_mapping\":{\"UserID\":\"id\",\"MovieID\":\"id\"," + + "\"Rating\":\"rate\"},\"value_mapping\":{},\"selected\":[]," + + "\"ignored\":[\"Timestamp\"],\"null_values\":[\"\"]," + - "\"update_strategies\":{},\"batch_size\":500}]}]}"; ++ "\"update_strategies\":{},\"batch_size\":500}]}]," + ++ "\"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\"}}"; + Assert.assertEquals(expectV2Json, actualV2Json); + + FileUtils.forceDelete(inputFile); + FileUtils.forceDelete(outputFile); + } +}
