imbajin commented on code in PR #785:
URL:
https://github.com/apache/hugegraph-toolchain/pull/785#discussion_r4209777156
##########
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:
⚠️ Blocking: no. Summary: This parser consumes Spark-decoded TEXT rows, but
the Spark reader uses `reader.text(path)` without applying
`FileSource.charset()`. Non-UTF-8 TEXT mappings can therefore silently corrupt
IDs or properties. Please reject non-UTF-8 TEXT during driver preflight or
decode with the configured charset. Evidence: `FileSource.charset()` is
configurable; Spark 3.5.8 says text files must be UTF-8: [DataFrameReader
documentation](https://spark.apache.org/docs/3.5.8/api/scala/org/apache/spark/sql/DataFrameReader.html#text(paths:String*)).
##########
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:
⚠️ Blocking: no. Summary: The Spark API path accepts `--incremental-mode`
and `--failure-mode` through shared `LoadOptions`, but this partition loop
processes every row without consulting saved progress or failure records;
reruns therefore perform a full import despite those flags. Please reject these
options before submission or implement their Spark semantics. Evidence: this
loop calls `loadRow` for every Spark row, and `docs/spark-java17.md` documents
these modes as unsupported.
--
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]
