imbajin commented on code in PR #785:
URL:
https://github.com/apache/hugegraph-toolchain/pull/785#discussion_r4218784088
##########
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:
Fixed: FILE/HDFS CSV, JSON and TEXT mappings now reject non-UTF-8 charsets
before Spark initialization or partition submission; UTF-8 aliases remain
accepted. The existing SparkRowTest exercises the real load() path, including a
valid mapping before the rejected mapping. Java 17 targeted tests and
independent review passed. Updated-head CI is pending.
##########
.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:
Fixed: added setup-java-env/**, .github/configs/settings.xml and
start-hugegraph-servers.sh to the compatibility workflow PR paths. actionlint
passed.
##########
hugegraph-loader/assembly/static/bin/get-params.sh:
##########
@@ -17,40 +17,145 @@
#
function get_params() {
- echo "params: $*"
- ENGINE_PARAMS=""
- HUGEGRAPH_PARAMS=""
+ ENGINE_ARGS=()
+ HUGEGRAPH_ARGS=()
+ local file="" mode="" option="" value="" files="" file_set=false
files_set=false inline=false
+ local conf_mode=""
while (("$#")); do
- case "$1" in
- --graph | --schema | --host | --port | --username | --token | --protocol
| \
+ option=${1%%=*}
+ inline=false
+ if [[ "$1" == *=* ]]; then
+ inline=true
+ value=${1#*=}
+ fi
+ case "$option" in
+ -g | --graph | -s | --schema | -h | -i | --host | -p | --port |
--username | --password | --token | --protocol | \
+ --pd-peers | --pd-token | --meta-endpoints | --direct | --route-type |
--cluster | --graphspace | --create-graph | \
--trust-store-file | --trust-store-password | --clear-all-data |
--clear-timeout | \
--incremental-mode | --failure-mode | --batch-insert-threads |
--single-insert-threads | \
- --max-conn | --max-conn-per-route | --batch-size | --max-parse-errors |
--max-insert-errors | \
+ --max-conn | --max-conn-per-route | --batch-size | --parallel-count |
--parser-threads | \
+ --start-file | --end-file | --scatter-sources | --cdc-flush-interval |
--cdc-sink-parallelism | \
+ --max-read-errors | --max-parse-errors | --max-insert-errors |
--max-read-lines | \
--timeout | --shutdown-timeout | --retry-times | --retry-interval |
--check-vertex | \
- --print-progress | --dry-run | --sink-type | --vertex-partitions |
--edge-partitions | --help )
- HUGEGRAPH_PARAMS="$HUGEGRAPH_PARAMS $1 $2"
- shift 2
+ --print-progress | --dry-run | --test-mode | --use-prefilter |
--short-id | --vertex-edge-limit | \
+ --sink-type | --vertex-partitions | --edge-partitions |
--vertex-table-name | --edge-table-name | \
+ --hbase-zk-quorum | --hbase-zk-port | --hbase-zk-parent | --restore |
--backend | --serializer | \
+ --scheduler-type | --batch-failure-fallback)
+ # LoadOptions uses arity=1 even for booleans; ShortIdConfig lists also
consume one value.
+ if [[ "$inline" == true ]]; then
+ shift
+ else
+ if (( $# < 2 )); then
+ echo "Missing value for $option" >&2
+ return 2
Review Comment:
Fixed in this PR: the legacy CDC launcher now enables strict shell error
handling, so rejected option values stop execution before command construction.
Existing launcher regression tests cover missing password, file and batch-size
values. The combined merge with the rebased #786 is conflict-free.
--
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]
