You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hugegraph.apache.org by GitBox <gi...@apache.org> on 2022/05/24 03:20:40 UTC
[GitHub] [incubator-hugegraph-toolchain] simon824 opened a new pull request, #282: feat: introduce HugeGraphSparkLoader
simon824 opened a new pull request, #282:
URL: https://github.com/apache/incubator-hugegraph-toolchain/pull/282
closed #281
basic version of HugeGraphSparkLoader
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-hugegraph-toolchain] simon824 commented on a diff in pull request #282: feat: introduce HugeGraphSparkLoader
Posted by GitBox <gi...@apache.org>.
simon824 commented on code in PR #282:
URL: https://github.com/apache/incubator-hugegraph-toolchain/pull/282#discussion_r880124096
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
Review Comment:
do you mean like this?
```java
ds.foreachPartition(p -> {
loadPartition();
p.forEachRemaining((Row row) -> {
loadRow();
});
});
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-hugegraph-toolchain] imbajin commented on pull request #282: feat: introduce HugeGraphSparkLoader
Posted by GitBox <gi...@apache.org>.
imbajin commented on PR #282:
URL: https://github.com/apache/incubator-hugegraph-toolchain/pull/282#issuecomment-1135488173
could enable check-style plugin to check code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-hugegraph-toolchain] imbajin merged pull request #282: feat: introduce HugeGraphSparkLoader
Posted by GitBox <gi...@apache.org>.
imbajin merged PR #282:
URL: https://github.com/apache/incubator-hugegraph-toolchain/pull/282
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-hugegraph-toolchain] imbajin commented on a diff in pull request #282: feat: introduce HugeGraphSparkLoader
Posted by GitBox <gi...@apache.org>.
imbajin commented on code in PR #282:
URL: https://github.com/apache/incubator-hugegraph-toolchain/pull/282#discussion_r882550011
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.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;
+
+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();
+
+ SparkSession session =
+ SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ 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);
+ });
+ context.close();
+ });
+ }
+ session.close();
+ session.stop();
+ }
+
+ 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<>());
+ }
+ 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)) {
+ sink(builderMap, context.client().graph(),
+ this.loadOptions.checkVertex);
+ }
+ }
+ }
+
+ private Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ 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 (row.mkString().equals("")) {
+ return;
+ }
+ List<GraphElement> elements;
+ switch (struct.input().type()) {
+ case FILE:
+ case HDFS:
+ FileSource fileSource = struct.input().asFileSource();
+ elements = builder.build(fileSource.header(),
+ row.mkString()
+ .split(fileSource.delimiter()));
+ break;
+ case JDBC:
+ //TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'",
+ struct.input().type()));
+ }
+ graphElements.addAll(elements);
+ }
+
+ private void sink(
+ 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();
Review Comment:
same problem, seems one line is better? @javeme
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.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;
+
+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();
+
+ SparkSession session =
+ SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ 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);
+ });
+ context.close();
+ });
+ }
+ session.close();
+ session.stop();
+ }
+
+ 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<>());
+ }
+ 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)) {
+ sink(builderMap, context.client().graph(),
+ this.loadOptions.checkVertex);
+ }
+ }
+ }
+
+ private Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
Review Comment:
move `todo` to previous line
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.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;
+
+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();
+
+ SparkSession session =
+ SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ 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);
+ });
+ context.close();
+ });
+ }
+ session.close();
+ session.stop();
+ }
+
+ 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<>());
+ }
+ 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)) {
+ sink(builderMap, context.client().graph(),
+ this.loadOptions.checkVertex);
+ }
+ }
+ }
+
+ private Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ 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 (row.mkString().equals("")) {
Review Comment:
seems could revert it to `"".equals(row.mkString())`
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.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;
+
+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();
+
+ SparkSession session =
+ SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ 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);
+ });
+ context.close();
+ });
+ }
+ session.close();
+ session.stop();
+ }
+
+ 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<>());
+ }
+ 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();
Review Comment:
need align? @javeme
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.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;
+
+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();
+
+ SparkSession session =
+ SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ 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);
+ });
+ context.close();
+ });
+ }
+ session.close();
+ session.stop();
+ }
+
+ 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<>());
+ }
+ 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();
Review Comment:
ditto
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.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;
+
+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();
+
+ SparkSession session =
+ SparkSession.builder().getOrCreate();
Review Comment:
ditto
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.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;
+
+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();
+
+ SparkSession session =
+ SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ 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);
+ });
+ context.close();
+ });
+ }
+ session.close();
+ session.stop();
+ }
+
+ 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<>());
+ }
+ 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)) {
+ sink(builderMap, context.client().graph(),
+ this.loadOptions.checkVertex);
+ }
+ }
+ }
+
+ private Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ 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 (row.mkString().equals("")) {
+ return;
+ }
+ List<GraphElement> elements;
+ switch (struct.input().type()) {
+ case FILE:
+ case HDFS:
+ FileSource fileSource = struct.input().asFileSource();
+ elements = builder.build(fileSource.header(),
+ row.mkString()
+ .split(fileSource.delimiter()));
+ break;
+ case JDBC:
+ //TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'",
+ struct.input().type()));
+ }
+ graphElements.addAll(elements);
+ }
+
+ private void sink(
+ Map.Entry<ElementBuilder, List<GraphElement>> builderMap,
+ GraphManager g, boolean isCheckVertex) {
Review Comment:
seems we could move 214 to 213 to avoid confuse wrap line (maybe chars < 100 is ok)
then 215 align with it
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.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;
+
+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();
+
+ SparkSession session =
+ SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ 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);
+ });
+ context.close();
+ });
+ }
+ session.close();
+ session.stop();
+ }
+
+ 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<>());
+ }
+ 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)) {
+ sink(builderMap, context.client().graph(),
+ this.loadOptions.checkVertex);
+ }
+ }
+ }
+
+ private Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ 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 (row.mkString().equals("")) {
+ return;
+ }
+ List<GraphElement> elements;
+ switch (struct.input().type()) {
+ case FILE:
+ case HDFS:
+ FileSource fileSource = struct.input().asFileSource();
+ elements = builder.build(fileSource.header(),
+ row.mkString()
+ .split(fileSource.delimiter()));
Review Comment:
the wrap looks strange @javeme
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-hugegraph-toolchain] javeme commented on a diff in pull request #282: feat: introduce HugeGraphSparkLoader
Posted by GitBox <gi...@apache.org>.
javeme commented on code in PR #282:
URL: https://github.com/apache/incubator-hugegraph-toolchain/pull/282#discussion_r883215740
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -175,7 +173,8 @@ private Dataset<Row> read(SparkSession ss, InputStruct struct) {
"Unexpected format value: " + format);
}
break;
- case JDBC: // TODO
+ case JDBC:
+ // TODO
Review Comment:
improve to "TODO: implement jdbc"?
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -210,15 +209,13 @@ private void parse(Row row,
graphElements.addAll(elements);
}
- private void sink(
- Map.Entry<ElementBuilder, List<GraphElement>> builderMap,
+ private void sink(Map.Entry<ElementBuilder, List<GraphElement>> builderMap,
Review Comment:
aldo update alignment of line 213
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-hugegraph-toolchain] codecov[bot] commented on pull request #282: feat: introduce HugeGraphSparkLoader
Posted by GitBox <gi...@apache.org>.
codecov[bot] commented on PR #282:
URL: https://github.com/apache/incubator-hugegraph-toolchain/pull/282#issuecomment-1135367498
# [Codecov](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#282](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7f35cf5) into [master](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/commit/7c39453ddf9855681fca304f4bd409d4a3952ab0?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7c39453) will **decrease** coverage by `4.56%`.
> The diff coverage is `1.76%`.
```diff
@@ Coverage Diff @@
## master #282 +/- ##
============================================
- Coverage 75.74% 71.17% -4.57%
+ Complexity 1847 877 -970
============================================
Files 235 82 -153
Lines 7766 3813 -3953
Branches 685 456 -229
============================================
- Hits 5882 2714 -3168
+ Misses 1501 895 -606
+ Partials 383 204 -179
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...m/baidu/hugegraph/loader/executor/LoadContext.java](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-aHVnZWdyYXBoLWxvYWRlci9zcmMvbWFpbi9qYXZhL2NvbS9iYWlkdS9odWdlZ3JhcGgvbG9hZGVyL2V4ZWN1dG9yL0xvYWRDb250ZXh0LmphdmE=) | `67.90% <0.00%> (ø)` | |
| [...om/baidu/hugegraph/loader/mapping/InputStruct.java](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-aHVnZWdyYXBoLWxvYWRlci9zcmMvbWFpbi9qYXZhL2NvbS9iYWlkdS9odWdlZ3JhcGgvbG9hZGVyL21hcHBpbmcvSW5wdXRTdHJ1Y3QuamF2YQ==) | `93.02% <ø> (ø)` | |
| [.../baidu/hugegraph/loader/source/AbstractSource.java](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-aHVnZWdyYXBoLWxvYWRlci9zcmMvbWFpbi9qYXZhL2NvbS9iYWlkdS9odWdlZ3JhcGgvbG9hZGVyL3NvdXJjZS9BYnN0cmFjdFNvdXJjZS5qYXZh) | `83.33% <ø> (ø)` | |
| [...baidu/hugegraph/loader/source/file/FileFilter.java](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-aHVnZWdyYXBoLWxvYWRlci9zcmMvbWFpbi9qYXZhL2NvbS9iYWlkdS9odWdlZ3JhcGgvbG9hZGVyL3NvdXJjZS9maWxlL0ZpbGVGaWx0ZXIuamF2YQ==) | `75.00% <ø> (ø)` | |
| [...aidu/hugegraph/loader/source/file/SkippedLine.java](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-aHVnZWdyYXBoLWxvYWRlci9zcmMvbWFpbi9qYXZhL2NvbS9iYWlkdS9odWdlZ3JhcGgvbG9hZGVyL3NvdXJjZS9maWxlL1NraXBwZWRMaW5lLmphdmE=) | `91.66% <ø> (ø)` | |
| [...u/hugegraph/loader/spark/HugeGraphSparkLoader.java](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-aHVnZWdyYXBoLWxvYWRlci9zcmMvbWFpbi9qYXZhL2NvbS9iYWlkdS9odWdlZ3JhcGgvbG9hZGVyL3NwYXJrL0h1Z2VHcmFwaFNwYXJrTG9hZGVyLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...baidu/hugegraph/loader/mapping/ElementMapping.java](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-aHVnZWdyYXBoLWxvYWRlci9zcmMvbWFpbi9qYXZhL2NvbS9iYWlkdS9odWdlZ3JhcGgvbG9hZGVyL21hcHBpbmcvRWxlbWVudE1hcHBpbmcuamF2YQ==) | `87.83% <50.00%> (-1.06%)` | :arrow_down: |
| [...m/baidu/hugegraph/loader/executor/LoadOptions.java](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-aHVnZWdyYXBoLWxvYWRlci9zcmMvbWFpbi9qYXZhL2NvbS9iYWlkdS9odWdlZ3JhcGgvbG9hZGVyL2V4ZWN1dG9yL0xvYWRPcHRpb25zLmphdmE=) | `74.41% <100.00%> (ø)` | |
| [...ava/com/baidu/hugegraph/api/traverser/RaysAPI.java](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-aHVnZWdyYXBoLWNsaWVudC9zcmMvbWFpbi9qYXZhL2NvbS9iYWlkdS9odWdlZ3JhcGgvYXBpL3RyYXZlcnNlci9SYXlzQVBJLmphdmE=) | | |
| [.../com/baidu/hugegraph/driver/HugeClientBuilder.java](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-aHVnZWdyYXBoLWNsaWVudC9zcmMvbWFpbi9qYXZhL2NvbS9iYWlkdS9odWdlZ3JhcGgvZHJpdmVyL0h1Z2VDbGllbnRCdWlsZGVyLmphdmE=) | | |
| ... and [152 more](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [7c39453...7f35cf5](https://codecov.io/gh/apache/incubator-hugegraph-toolchain/pull/282?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-hugegraph-toolchain] javeme commented on a diff in pull request #282: feat: introduce HugeGraphSparkLoader
Posted by GitBox <gi...@apache.org>.
javeme commented on code in PR #282:
URL: https://github.com/apache/incubator-hugegraph-toolchain/pull/282#discussion_r880029517
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
Review Comment:
prefer name `SparkSession session`
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
Review Comment:
can we use interface type for var declaration: Map\<ElementBuilder, List\<GraphElement>>
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
+ builders.entrySet()) {
+ ElementMapping elementMapping =
+ builderMap.getKey().mapping();
+ // Parse
+ if (elementMapping.skip()) continue;
+ parse(row, builderMap, struct);
+
+ // Insert
+ ArrayList<GraphElement> graphElements =
+ builderMap.getValue();
+ if (graphElements.size() > elementMapping.batchSize() ||
+ (!p.hasNext() && graphElements.size() > 0)) {
+ sink(builderMap, context.client().graph(),
+ isCheckVertex);
+ }
+ }
+ });
+ context.close();
+ });
+ }
+ ss.close();
+ ss.stop();
+ }
+
+ public Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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);
Review Comment:
align with IllegalStateException
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
+ builders.entrySet()) {
+ ElementMapping elementMapping =
+ builderMap.getKey().mapping();
+ // Parse
+ if (elementMapping.skip()) continue;
+ parse(row, builderMap, struct);
+
+ // Insert
+ ArrayList<GraphElement> graphElements =
+ builderMap.getValue();
+ if (graphElements.size() > elementMapping.batchSize() ||
+ (!p.hasNext() && graphElements.size() > 0)) {
+ sink(builderMap, context.client().graph(),
+ isCheckVertex);
+ }
+ }
+ });
+ context.close();
+ });
+ }
+ ss.close();
+ ss.stop();
+ }
+
+ public Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
Review Comment:
can we rename ss to session for reading?
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
+ builders.entrySet()) {
+ ElementMapping elementMapping =
+ builderMap.getKey().mapping();
+ // Parse
+ if (elementMapping.skip()) continue;
+ parse(row, builderMap, struct);
+
+ // Insert
+ ArrayList<GraphElement> graphElements =
+ builderMap.getValue();
+ if (graphElements.size() > elementMapping.batchSize() ||
+ (!p.hasNext() && graphElements.size() > 0)) {
+ sink(builderMap, context.client().graph(),
+ isCheckVertex);
+ }
+ }
+ });
+ context.close();
+ });
+ }
+ ss.close();
+ ss.stop();
+ }
+
+ public Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'", input.type()));
+ }
+ return ds;
+ }
+
+ public void parse(Row row,
+ Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap,
+ InputStruct struct) {
+ ElementBuilder builder = builderMap.getKey();
+ ArrayList<GraphElement> graphElements = builderMap.getValue();
+ if (!row.mkString().equals("")) {
+ List<GraphElement> build;
+ switch (struct.input().type()) {
+ case FILE:
+ case HDFS:
+ FileSource fileSource = struct.input().asFileSource();
+ build = builder.build(fileSource.header(),
+ row.mkString()
+ .split(fileSource.delimiter()));
+ break;
+ case JDBC:
+ //TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'",
+ struct.input().type()));
+ }
+
+ graphElements.addAll(build);
+ }
+ }
+
+ public void sink(
+ Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap,
+ GraphManager g, boolean isCheckVertex) {
+ ElementBuilder builder = builderMap.getKey();
+ ElementMapping elementMapping = builder.mapping();
+ ArrayList<GraphElement> graphElements = builderMap.getValue();
+ boolean isVertex = builder.mapping().type().isVertex();
+ Map<String, UpdateStrategy> stringUpdateStrategyMap =
+ elementMapping.updateStrategies();
+ if (stringUpdateStrategyMap.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(stringUpdateStrategyMap)
+ .createIfNotExist(true);
+ g.updateVertices(req.build());
+ } else {
+ BatchEdgeRequest.Builder req = new BatchEdgeRequest.Builder();
+ req.edges((List<Edge>) (Object) graphElements)
+ .updatingStrategies(stringUpdateStrategyMap)
+ .checkVertex(isCheckVertex)
+ .createIfNotExist(true);
+
+ g.updateEdges(req.build());
+ }
+ }
+ graphElements.clear();
+ }
+
Review Comment:
remove useless blank line
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
+ builders.entrySet()) {
+ ElementMapping elementMapping =
+ builderMap.getKey().mapping();
+ // Parse
+ if (elementMapping.skip()) continue;
+ parse(row, builderMap, struct);
+
+ // Insert
+ ArrayList<GraphElement> graphElements =
+ builderMap.getValue();
+ if (graphElements.size() > elementMapping.batchSize() ||
+ (!p.hasNext() && graphElements.size() > 0)) {
+ sink(builderMap, context.client().graph(),
+ isCheckVertex);
+ }
+ }
+ });
+ context.close();
+ });
+ }
+ ss.close();
+ ss.stop();
+ }
+
+ public Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'", input.type()));
+ }
+ return ds;
+ }
+
+ public void parse(Row row,
+ Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap,
+ InputStruct struct) {
+ ElementBuilder builder = builderMap.getKey();
+ ArrayList<GraphElement> graphElements = builderMap.getValue();
+ if (!row.mkString().equals("")) {
+ List<GraphElement> build;
+ switch (struct.input().type()) {
+ case FILE:
+ case HDFS:
+ FileSource fileSource = struct.input().asFileSource();
+ build = builder.build(fileSource.header(),
+ row.mkString()
+ .split(fileSource.delimiter()));
+ break;
+ case JDBC:
+ //TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'",
+ struct.input().type()));
+ }
+
+ graphElements.addAll(build);
+ }
+ }
+
+ public void sink(
+ Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap,
+ GraphManager g, boolean isCheckVertex) {
+ ElementBuilder builder = builderMap.getKey();
+ ElementMapping elementMapping = builder.mapping();
+ ArrayList<GraphElement> graphElements = builderMap.getValue();
+ boolean isVertex = builder.mapping().type().isVertex();
+ Map<String, UpdateStrategy> stringUpdateStrategyMap =
Review Comment:
updateStrategyMap name is more concise
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
+ builders.entrySet()) {
+ ElementMapping elementMapping =
+ builderMap.getKey().mapping();
+ // Parse
+ if (elementMapping.skip()) continue;
+ parse(row, builderMap, struct);
+
+ // Insert
+ ArrayList<GraphElement> graphElements =
Review Comment:
ditto
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
+ builders.entrySet()) {
+ ElementMapping elementMapping =
+ builderMap.getKey().mapping();
+ // Parse
+ if (elementMapping.skip()) continue;
Review Comment:
expect style like:
```java
if (x) {
xx
}
```
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
+ builders.entrySet()) {
+ ElementMapping elementMapping =
+ builderMap.getKey().mapping();
+ // Parse
+ if (elementMapping.skip()) continue;
+ parse(row, builderMap, struct);
+
+ // Insert
+ ArrayList<GraphElement> graphElements =
+ builderMap.getValue();
+ if (graphElements.size() > elementMapping.batchSize() ||
+ (!p.hasNext() && graphElements.size() > 0)) {
+ sink(builderMap, context.client().graph(),
+ isCheckVertex);
+ }
+ }
+ });
+ context.close();
+ });
+ }
+ ss.close();
+ ss.stop();
+ }
+
+ public Dataset<Row> read(SparkSession ss, InputStruct struct) {
Review Comment:
we can keep `private` mark as possible for better encapsulation
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
Review Comment:
can create and move to a new method like loadRow()?
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
Review Comment:
ditto
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
+ builders.entrySet()) {
+ ElementMapping elementMapping =
+ builderMap.getKey().mapping();
+ // Parse
+ if (elementMapping.skip()) continue;
+ parse(row, builderMap, struct);
+
+ // Insert
+ ArrayList<GraphElement> graphElements =
+ builderMap.getValue();
+ if (graphElements.size() > elementMapping.batchSize() ||
+ (!p.hasNext() && graphElements.size() > 0)) {
+ sink(builderMap, context.client().graph(),
+ isCheckVertex);
+ }
+ }
+ });
+ context.close();
+ });
+ }
+ ss.close();
+ ss.stop();
+ }
+
+ public Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'", input.type()));
+ }
+ return ds;
+ }
+
+ public void parse(Row row,
+ Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap,
Review Comment:
ArrayList => List\<GraphElement>
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
Review Comment:
can create and move to a new method like loadPartition()?
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
+ builders.entrySet()) {
+ ElementMapping elementMapping =
+ builderMap.getKey().mapping();
+ // Parse
+ if (elementMapping.skip()) continue;
+ parse(row, builderMap, struct);
+
+ // Insert
+ ArrayList<GraphElement> graphElements =
+ builderMap.getValue();
+ if (graphElements.size() > elementMapping.batchSize() ||
+ (!p.hasNext() && graphElements.size() > 0)) {
+ sink(builderMap, context.client().graph(),
+ isCheckVertex);
+ }
+ }
+ });
+ context.close();
+ });
+ }
+ ss.close();
+ ss.stop();
+ }
+
+ public Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'", input.type()));
+ }
+ return ds;
+ }
+
+ public void parse(Row row,
Review Comment:
we can keep private mark
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
+ builders.entrySet()) {
+ ElementMapping elementMapping =
+ builderMap.getKey().mapping();
+ // Parse
+ if (elementMapping.skip()) continue;
+ parse(row, builderMap, struct);
+
+ // Insert
+ ArrayList<GraphElement> graphElements =
+ builderMap.getValue();
+ if (graphElements.size() > elementMapping.batchSize() ||
+ (!p.hasNext() && graphElements.size() > 0)) {
+ sink(builderMap, context.client().graph(),
+ isCheckVertex);
+ }
+ }
+ });
+ context.close();
+ });
+ }
+ ss.close();
+ ss.stop();
+ }
+
+ public Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'", input.type()));
Review Comment:
ditto
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
+ builders.entrySet()) {
+ ElementMapping elementMapping =
+ builderMap.getKey().mapping();
+ // Parse
+ if (elementMapping.skip()) continue;
+ parse(row, builderMap, struct);
+
+ // Insert
+ ArrayList<GraphElement> graphElements =
+ builderMap.getValue();
+ if (graphElements.size() > elementMapping.batchSize() ||
+ (!p.hasNext() && graphElements.size() > 0)) {
+ sink(builderMap, context.client().graph(),
+ isCheckVertex);
+ }
+ }
+ });
+ context.close();
+ });
+ }
+ ss.close();
+ ss.stop();
+ }
+
+ public Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'", input.type()));
+ }
+ return ds;
+ }
+
+ public void parse(Row row,
+ Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap,
+ InputStruct struct) {
+ ElementBuilder builder = builderMap.getKey();
+ ArrayList<GraphElement> graphElements = builderMap.getValue();
+ if (!row.mkString().equals("")) {
Review Comment:
we can just return if `row.mkString().equals(""))` to be more concise
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
+ builders.entrySet()) {
+ ElementMapping elementMapping =
+ builderMap.getKey().mapping();
+ // Parse
+ if (elementMapping.skip()) continue;
+ parse(row, builderMap, struct);
+
+ // Insert
+ ArrayList<GraphElement> graphElements =
+ builderMap.getValue();
+ if (graphElements.size() > elementMapping.batchSize() ||
+ (!p.hasNext() && graphElements.size() > 0)) {
+ sink(builderMap, context.client().graph(),
+ isCheckVertex);
+ }
+ }
+ });
+ context.close();
+ });
+ }
+ ss.close();
+ ss.stop();
+ }
+
+ public Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'", input.type()));
+ }
+ return ds;
+ }
+
+ public void parse(Row row,
+ Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap,
+ InputStruct struct) {
+ ElementBuilder builder = builderMap.getKey();
+ ArrayList<GraphElement> graphElements = builderMap.getValue();
+ if (!row.mkString().equals("")) {
+ List<GraphElement> build;
+ switch (struct.input().type()) {
+ case FILE:
+ case HDFS:
+ FileSource fileSource = struct.input().asFileSource();
+ build = builder.build(fileSource.header(),
+ row.mkString()
+ .split(fileSource.delimiter()));
+ break;
+ case JDBC:
+ //TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'",
Review Comment:
ditto
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
+ builders.entrySet()) {
+ ElementMapping elementMapping =
+ builderMap.getKey().mapping();
+ // Parse
+ if (elementMapping.skip()) continue;
+ parse(row, builderMap, struct);
+
+ // Insert
+ ArrayList<GraphElement> graphElements =
+ builderMap.getValue();
+ if (graphElements.size() > elementMapping.batchSize() ||
+ (!p.hasNext() && graphElements.size() > 0)) {
+ sink(builderMap, context.client().graph(),
+ isCheckVertex);
+ }
+ }
+ });
+ context.close();
+ });
+ }
+ ss.close();
+ ss.stop();
+ }
+
+ public Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'", input.type()));
+ }
+ return ds;
+ }
+
+ public void parse(Row row,
+ Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap,
+ InputStruct struct) {
+ ElementBuilder builder = builderMap.getKey();
+ ArrayList<GraphElement> graphElements = builderMap.getValue();
+ if (!row.mkString().equals("")) {
+ List<GraphElement> build;
+ switch (struct.input().type()) {
+ case FILE:
+ case HDFS:
+ FileSource fileSource = struct.input().asFileSource();
+ build = builder.build(fileSource.header(),
+ row.mkString()
+ .split(fileSource.delimiter()));
+ break;
+ case JDBC:
+ //TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'",
+ struct.input().type()));
+ }
+
+ graphElements.addAll(build);
+ }
+ }
+
+ public void sink(
+ Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap,
+ GraphManager g, boolean isCheckVertex) {
+ ElementBuilder builder = builderMap.getKey();
+ ElementMapping elementMapping = builder.mapping();
+ ArrayList<GraphElement> graphElements = builderMap.getValue();
+ boolean isVertex = builder.mapping().type().isVertex();
+ Map<String, UpdateStrategy> stringUpdateStrategyMap =
+ elementMapping.updateStrategies();
Review Comment:
align with stringUpdateStrategyMap
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
+ builders.entrySet()) {
+ ElementMapping elementMapping =
+ builderMap.getKey().mapping();
+ // Parse
+ if (elementMapping.skip()) continue;
+ parse(row, builderMap, struct);
+
+ // Insert
+ ArrayList<GraphElement> graphElements =
+ builderMap.getValue();
+ if (graphElements.size() > elementMapping.batchSize() ||
+ (!p.hasNext() && graphElements.size() > 0)) {
+ sink(builderMap, context.client().graph(),
+ isCheckVertex);
+ }
+ }
+ });
+ context.close();
+ });
+ }
+ ss.close();
+ ss.stop();
+ }
+
+ public Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'", input.type()));
+ }
+ return ds;
+ }
+
+ public void parse(Row row,
+ Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap,
+ InputStruct struct) {
+ ElementBuilder builder = builderMap.getKey();
+ ArrayList<GraphElement> graphElements = builderMap.getValue();
+ if (!row.mkString().equals("")) {
+ List<GraphElement> build;
+ switch (struct.input().type()) {
+ case FILE:
+ case HDFS:
+ FileSource fileSource = struct.input().asFileSource();
+ build = builder.build(fileSource.header(),
+ row.mkString()
+ .split(fileSource.delimiter()));
+ break;
+ case JDBC:
+ //TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'",
+ struct.input().type()));
+ }
+
+ graphElements.addAll(build);
+ }
+ }
+
+ public void sink(
+ Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap,
Review Comment:
ditto
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.slf4j.Logger;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class HugeGraphSparkLoader implements Serializable {
+
+ public static final Logger LOG = Log.logger(HugeGraphSparkLoader.class);
+
+ private final LoadOptions loadOptions;
+
+ 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);
+ }
+
+ public void load() {
+ LoadMapping mapping = LoadMapping.of(loadOptions.file);
+ boolean isCheckVertex = loadOptions.checkVertex;
+ List<InputStruct> structs = mapping.structs();
+
+ SparkSession ss = SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ // Read
+ Dataset<Row> ds = read(ss, struct);
+ ds.foreachPartition(p -> {
+ LoadContext context = new LoadContext(loadOptions);
+ HashMap<ElementBuilder, ArrayList<GraphElement>> builders =
+ new HashMap<>();
+ for (VertexMapping vertexMapping : struct.vertices()) {
+ builders.put(
+ new VertexBuilder(context, struct, vertexMapping),
+ new ArrayList<>());
+ }
+ for (EdgeMapping edgeMapping : struct.edges()) {
+ builders.put(new EdgeBuilder(context, struct, edgeMapping),
+ new ArrayList<>());
+ }
+
+ p.forEachRemaining((Row row) -> {
+ for (Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap :
+ builders.entrySet()) {
+ ElementMapping elementMapping =
+ builderMap.getKey().mapping();
+ // Parse
+ if (elementMapping.skip()) continue;
+ parse(row, builderMap, struct);
+
+ // Insert
+ ArrayList<GraphElement> graphElements =
+ builderMap.getValue();
+ if (graphElements.size() > elementMapping.batchSize() ||
+ (!p.hasNext() && graphElements.size() > 0)) {
+ sink(builderMap, context.client().graph(),
+ isCheckVertex);
+ }
+ }
+ });
+ context.close();
+ });
+ }
+ ss.close();
+ ss.stop();
+ }
+
+ public Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'", input.type()));
+ }
+ return ds;
+ }
+
+ public void parse(Row row,
+ Map.Entry<ElementBuilder, ArrayList<GraphElement>> builderMap,
+ InputStruct struct) {
+ ElementBuilder builder = builderMap.getKey();
+ ArrayList<GraphElement> graphElements = builderMap.getValue();
+ if (!row.mkString().equals("")) {
+ List<GraphElement> build;
Review Comment:
prefer to rename build to elements
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-hugegraph-toolchain] javeme commented on a diff in pull request #282: feat: introduce HugeGraphSparkLoader
Posted by GitBox <gi...@apache.org>.
javeme commented on code in PR #282:
URL: https://github.com/apache/incubator-hugegraph-toolchain/pull/282#discussion_r882608495
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.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;
+
+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();
+
+ SparkSession session =
+ SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ 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);
+ });
+ context.close();
+ });
+ }
+ session.close();
+ session.stop();
+ }
+
+ 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<>());
+ }
+ 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)) {
+ sink(builderMap, context.client().graph(),
+ this.loadOptions.checkVertex);
+ }
+ }
+ }
+
+ private Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ 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 (row.mkString().equals("")) {
+ return;
+ }
+ List<GraphElement> elements;
+ switch (struct.input().type()) {
+ case FILE:
+ case HDFS:
+ FileSource fileSource = struct.input().asFileSource();
+ elements = builder.build(fileSource.header(),
+ row.mkString()
+ .split(fileSource.delimiter()));
+ break;
+ case JDBC:
+ //TODO
+ default:
+ throw new AssertionError(String.format(
+ "Unsupported input source '%s'",
+ struct.input().type()));
+ }
+ graphElements.addAll(elements);
+ }
+
+ private void sink(
+ 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();
Review Comment:
I think both are fine
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.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;
+
+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();
+
+ SparkSession session =
+ SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ 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);
+ });
+ context.close();
+ });
+ }
+ session.close();
+ session.stop();
+ }
+
+ 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<>());
+ }
+ 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)) {
+ sink(builderMap, context.client().graph(),
+ this.loadOptions.checkVertex);
+ }
+ }
+ }
+
+ private Dataset<Row> read(SparkSession ss, InputStruct struct) {
+ InputSource input = struct.input();
+ String charset = input.charset();
+ 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();
+
+ DataFrameReader reader = ss.read();
+ Dataset<Row> ds;
+ switch (input.type()) {
+ case FILE:
+ case HDFS:
+ 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: // TODO
+ 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 (row.mkString().equals("")) {
+ return;
+ }
+ List<GraphElement> elements;
+ switch (struct.input().type()) {
+ case FILE:
+ case HDFS:
+ FileSource fileSource = struct.input().asFileSource();
+ elements = builder.build(fileSource.header(),
+ row.mkString()
+ .split(fileSource.delimiter()));
Review Comment:
looks aligned on the big screen: <img width="629" alt="image" src="https://user-images.githubusercontent.com/9625821/170485297-7b69127b-34db-44bd-badf-dba7ad03c586.png">
##########
hugegraph-loader/src/main/java/com/baidu/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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.spark;
+
+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.executor.LoadContext;
+import com.baidu.hugegraph.loader.executor.LoadOptions;
+import com.baidu.hugegraph.loader.mapping.EdgeMapping;
+import com.baidu.hugegraph.loader.mapping.VertexMapping;
+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.source.InputSource;
+import com.baidu.hugegraph.loader.source.file.FileFilter;
+import com.baidu.hugegraph.loader.source.file.FileFormat;
+import com.baidu.hugegraph.loader.source.file.FileSource;
+import com.baidu.hugegraph.loader.source.file.SkippedLine;
+import com.baidu.hugegraph.loader.source.file.Compression;
+import com.baidu.hugegraph.loader.util.Printer;
+import com.baidu.hugegraph.structure.GraphElement;
+import com.baidu.hugegraph.structure.graph.UpdateStrategy;
+import com.baidu.hugegraph.structure.graph.Vertex;
+import com.baidu.hugegraph.structure.graph.Edge;
+import com.baidu.hugegraph.structure.graph.BatchEdgeRequest;
+import com.baidu.hugegraph.structure.graph.BatchVertexRequest;
+import com.baidu.hugegraph.util.Log;
+
+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.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;
+
+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();
+
+ SparkSession session =
+ SparkSession.builder().getOrCreate();
+ for (InputStruct struct : structs) {
+ 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);
+ });
+ context.close();
+ });
+ }
+ session.close();
+ session.stop();
+ }
+
+ 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<>());
+ }
+ 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();
Review Comment:
seems can keep in a line
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hugegraph.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org