imbajin commented on code in PR #785:
URL:
https://github.com/apache/hugegraph-toolchain/pull/785#discussion_r4215080096
##########
.github/workflows/upstream-compatibility.yml:
##########
@@ -0,0 +1,85 @@
+name: Upstream compatibility
+
+on:
+ schedule:
+ - cron: '17 3 * * *'
+ workflow_dispatch:
+ pull_request:
+ paths:
+ - '.github/workflows/upstream-compatibility.yml'
+ - '.github/actions/get-hugegraph-commit/**'
+ - '.github/actions/setup-candidate-sdk/**'
+ - 'hugegraph-client/assembly/travis/install-candidate-sdk.sh'
+ - 'hugegraph-client/assembly/travis/checkout-server.sh'
+ - '.github/scripts/verify_candidate_image_sdk.py'
Review Comment:
⚠️ Blocking: no. Summary: This workflow consumes the local setup-java-env
action (which supplies Maven's settings file) and runs
start-hugegraph-servers.sh, but none of those paths is in pull_request.paths. A
change to either action, .github/configs/settings.xml, or the server startup
script can therefore skip this current-master compatibility gate; please add
all three paths. Evidence: workflow steps 32–36 and 58–61; setup-java-env
exports --settings .github/configs/settings.xml through MAVEN_ARGS.
##########
hugegraph-loader/src/main/java/org/apache/hugegraph/loader/spark/HugeGraphSparkLoader.java:
##########
@@ -125,66 +129,108 @@ private void registerKryoClasses(SparkConf conf) {
}
}
+ static void checkDeployment(SparkConf conf) {
+ if ("cluster".equals(conf.get("spark.submit.deployMode", "client")) &&
+ conf.get("spark.master", "").startsWith("spark://")) {
+ throw new LoadException("Standalone cluster deploy mode is not
supported; " +
+ "use client deploy mode");
+ }
+ }
+
public void load() throws ExecutionException, InterruptedException {
+ checkLoadOptions(this.loadOptions);
LoadMapping mapping = LoadMapping.of(this.loadOptions.file);
List<InputStruct> structs = mapping.structs();
+ checkHeaders(structs);
+ SparkConf conf = new SparkConf();
boolean sinkType = this.loadOptions.sinkType;
//if (!sinkType) {
//
this.loadOptions.copyBackendStoreInfo(mapping.getBackendStoreInfo());
//}
- SparkConf conf = new SparkConf();
registerKryoClasses(conf);
SparkSession session =
SparkSession.builder().config(conf).getOrCreate();
SparkContext sc = session.sparkContext();
+ LOG.info("Spark Loader runtime: Java {}, default charset {}",
+ System.getProperty("java.version"),
Charset.defaultCharset().name());
- LongAccumulator totalInsertSuccess =
sc.longAccumulator("totalInsertSuccess");
- List<Future<?>> futures = new ArrayList<>(structs.size());
+ try {
+ LongAccumulator totalInsertSuccess =
sc.longAccumulator("totalInsertSuccess");
+ List<Future<?>> futures = new ArrayList<>(structs.size());
- for (InputStruct struct : structs) {
- Future<?> future = this.executor.submit(() -> {
- LOG.info("\n Initializes the accumulator corresponding to the
{} ",
- struct.input().asFileSource().path());
- LoadDistributeMetrics loadDistributeMetrics = new
LoadDistributeMetrics(struct);
- loadDistributeMetrics.init(sc);
- LOG.info("\n Start to load data, data info is: \t {} ",
- struct.input().asFileSource().path());
- Dataset<Row> ds = read(session, struct);
- if (sinkType) {
- LOG.info("\n Start to load data using spark apis \n");
- ds.foreachPartition((Iterator<Row> p) -> {
- LoadContext context = initPartition(this.loadOptions,
struct);
- p.forEachRemaining((Row row) -> {
- loadRow(struct, row, p, context);
- });
- context.close();
- });
+ for (InputStruct struct : structs) {
+ Future<?> future = this.executor.submit(() -> {
+ LOG.info("\n Initializes the accumulator corresponding to
the {} ",
+ struct.input().asFileSource().path());
+ LoadDistributeMetrics loadDistributeMetrics = new
LoadDistributeMetrics(struct);
+ loadDistributeMetrics.init(sc);
+ LOG.info("\n Start to load data, data info is: \t {} ",
+ struct.input().asFileSource().path());
+ Dataset<Row> ds = read(session, struct);
+ if (sinkType) {
+ LOG.info("\n Start to load data using spark apis
\n");
+ ds.foreachPartition(new
PartitionWriter(this.loadOptions, struct));
- } else {
- LOG.info("\n Start to load data using spark bulkload \n");
- // gen-hfile
- HBaseDirectLoader directLoader = new
HBaseDirectLoader(loadOptions, struct,
-
loadDistributeMetrics);
- directLoader.bulkload(ds);
+ } else {
+ LOG.info("\n Start to load data using spark bulkload
\n");
+ // gen-hfile
+ HBaseDirectLoader directLoader = new
HBaseDirectLoader(loadOptions, struct,
+
loadDistributeMetrics);
+ directLoader.bulkload(ds);
- }
- collectLoadMetrics(loadDistributeMetrics, totalInsertSuccess);
- LOG.info("\n Finished load {} data ",
struct.input().asFileSource().path());
- });
- futures.add(future);
- }
- for (Future<?> future : futures) {
- future.get();
+ }
+ collectLoadMetrics(loadDistributeMetrics,
totalInsertSuccess);
+ LOG.info("\n Finished load {} data ",
struct.input().asFileSource().path());
+ });
+ futures.add(future);
+ }
+ for (Future<?> future : futures) {
+ future.get();
+ }
+
+ Long totalInsertSuccessCnt = totalInsertSuccess.value();
+ LOG.info("\n ------------The data load task is
complete-------------------\n" +
+ "\n insertSuccessCnt:\t {} \n
---------------------------------------------\n",
+ totalInsertSuccessCnt);
+ } finally {
+ this.executor.shutdownNow();
+ session.stop();
}
+ }
- Long totalInsertSuccessCnt = totalInsertSuccess.value();
- LOG.info("\n ------------The data load task is
complete-------------------\n" +
- "\n insertSuccessCnt:\t {} \n
---------------------------------------------\n",
- totalInsertSuccessCnt);
+ static void checkLoadOptions(LoadOptions options) {
+ if (options.incrementalMode || options.failureMode) {
+ throw new LoadException("Spark does not support --incremental-mode
or --failure-mode; " +
+ "see docs/spark-java17.md");
+ }
+ }
- sc.stop();
- session.close();
- session.stop();
+ static void checkHeaders(List<InputStruct> structs) {
+ for (InputStruct struct : structs) {
+ switch (struct.input().type()) {
+ case FILE:
+ case HDFS:
+ FileSource source = struct.input().asFileSource();
+ if (source.format() == FileFormat.TEXT &&
Review Comment:
⚠️ Blocking: yes. Summary: The new preflight rejects non-UTF-8 only for
TEXT, while the CSV and JSON readers below do not use the configured source
charset. A non-UTF-8 CSV mapping therefore passes preflight and can be decoded
as UTF-8, corrupting IDs or properties; please pass the configured encoding to
supported readers or reject unsupported encodings before Spark starts.
Evidence: AbstractSource exposes the mapping charset, and
HugeGraphSparkLoader.read calls reader.csv/json without an encoding option;
Spark 3.5.8 documents the [CSV encoding
option](https://spark.apache.org/docs/3.5.8/sql-data-sources-csv.html#data-source-option)
and [JSON encoding
option](https://spark.apache.org/docs/3.5.8/sql-data-sources-json.html#data-source-option).
--
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]
