imbajin commented on code in PR #785:
URL:
https://github.com/apache/hugegraph-toolchain/pull/785#discussion_r4213962990
##########
hugegraph-loader/src/main/java/org/apache/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -282,25 +313,29 @@ private Dataset<Row> read(SparkSession ss, InputStruct
struct) {
return ds;
}
- private void parse(Row row, Map.Entry<ElementBuilder, List<GraphElement>>
builderMap,
+ private static void parse(Row row, Map.Entry<ElementBuilder,
List<GraphElement>> builderMap,
InputStruct struct) {
ElementBuilder builder = builderMap.getKey();
List<GraphElement> graphElements = builderMap.getValue();
- if ("".equals(row.mkString())) {
- return;
- }
List<GraphElement> elements;
switch (struct.input().type()) {
case FILE:
case HDFS:
FileSource fileSource = struct.input().asFileSource();
- String delimiter = fileSource.delimiter();
- if (Optional.ofNullable(delimiter).isPresent()) {
- elements = builder.build(fileSource.header(),
- row.mkString(delimiter).split(delimiter));
+ if (fileSource.format() == FileFormat.TEXT) {
+ if (row.isNullAt(0) || "".equals(row.getString(0))) {
+ return;
+ }
+ Line line = new TextLineParser(fileSource.delimiter())
+ .parse(fileSource.header(), row.getString(0));
+ elements = builder.build(line.names(), line.values());
Review Comment:
Fixed: driver preflight now rejects non-UTF-8 TEXT mappings for both FILE
and HDFS before Spark initialization. UTF-8 aliases remain valid. CSV behavior
is unchanged; regression cases use the existing SparkRowTest suite.
##########
hugegraph-loader/src/main/java/org/apache/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -360,4 +395,32 @@ private void flush(Map.Entry<ElementBuilder,
List<GraphElement>> builderMap,
}
graphElements.clear();
}
+
+ static final class PartitionWriter implements
ForeachPartitionFunction<Row> {
+
+ final String optionsJson;
+ final String structJson;
+
+ PartitionWriter(LoadOptions options, InputStruct struct) {
+ // Spark uses Java serialization for closures, independently of
its data serializer.
+ this.optionsJson = JsonUtil.toJson(options);
+ this.structJson = JsonUtil.toJson(struct);
+ }
+
+ @Override
+ public void call(Iterator<Row> rows) {
+ LoadOptions options = JsonUtil.fromJson(this.optionsJson,
LoadOptions.class);
+ InputStruct struct = JsonUtil.fromJson(this.structJson,
InputStruct.class);
+ Map<ElementBuilder, List<GraphElement>> builders = new HashMap<>();
+ LoadContext context = new LoadContext(options);
+ try {
+ initPartition(context, struct, builders);
+ rows.forEachRemaining(row -> loadRow(struct, row, context,
builders,
Review Comment:
Fixed: Spark load preflight now rejects enabled incremental-mode and
failure-mode before reading the mapping or initializing Spark. Defaults remain
supported. The existing SparkRowTest suite covers both flags, and documentation
states this supported boundary.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
