hudi-agent commented on code in PR #20352: URL: https://github.com/apache/hudi/pull/20352#discussion_r4286622420
########## scripts/release/rc_validation/streamer_e2e_datagen.scala: ########## @@ -0,0 +1,71 @@ +/* + * 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. + */ + +// Feeds a ParquetDFSSource directory with parquet batches at a fixed interval, run through spark-shell -i. +// Batch 0 inserts `initial` keys; every later batch upserts up to `updates` existing keys and inserts +// `inserts` new keys. Each batch is staged elsewhere and then renamed into the source directory so the +// streamer never sees a partially written batch. The expected final key count is written to +// <dir>/expected_total. +// Required confs: spark.hudi.rc.e2e.{dir,batches,initial,updates,inserts,interval.seconds} + +import org.apache.hadoop.fs.{FileSystem, Path} +import org.apache.spark.sql.functions._ + +def conf(key: String) = spark.conf.get(s"spark.hudi.rc.e2e.$key") +val dir = conf("dir") +val numBatches = conf("batches").toInt +val initial = conf("initial").toLong +val updates = conf("updates").toLong +val inserts = conf("inserts").toLong +val intervalMs = conf("interval.seconds").toLong * 1000 + +val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration) +var nextKey = 0L + +for (b <- 0 until numBatches) { + val start = System.currentTimeMillis() + val ids = if (b == 0) { + spark.range(0, initial).toDF("id") + } else { + val updated = spark.range(0, updates).select(pmod(hash(lit(b), col("id")), lit(nextKey)).cast("long").as("id")).distinct() + updated.union(spark.range(nextKey, nextKey + inserts).toDF("id")) Review Comment: 🤖 nit: could you add a short comment explaining that this picks pseudo-random existing keys via a hash of the batch number, and that `distinct` means fewer than `updates` keys may be updated? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## scripts/release/rc_validation/quickstart.scala: ########## @@ -0,0 +1,125 @@ +/* + * 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. + */ + +// Quickstart-style sanity checks for a release candidate, run through spark-shell -i. +// Covers COW and MOR via the datasource API and Spark SQL. Prints "QS PASS|FAIL <check>" +// lines and exits non-zero if any check fails. +// Required conf: spark.hudi.rc.quickstart.dir (local working directory). + +import org.apache.spark.sql.SaveMode._ +import org.apache.spark.sql.functions._ + +val workDir = spark.conf.get("spark.hudi.rc.quickstart.dir") +val base = s"file://$workDir/tables" Review Comment: 🤖 non-blocking: A relative `--work_dir` (e.g. `--work_dir=rc-work`) makes this `file://rc-work/...`, and Hadoop then reads `rc-work` as the authority and throws `Wrong FS`. Quickstart and the e2e run both build `file://$DIR` paths, so both would fail in a way that looks like an RC problem. Could `validate_rc.sh` make the path absolute right after `mkdir -p`, e.g. `WORK_DIR=$(cd "$WORK_DIR" && pwd)`? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## scripts/release/rc_validation/streamer_e2e_datagen.scala: ########## @@ -0,0 +1,71 @@ +/* + * 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. + */ + +// Feeds a ParquetDFSSource directory with parquet batches at a fixed interval, run through spark-shell -i. +// Batch 0 inserts `initial` keys; every later batch upserts up to `updates` existing keys and inserts +// `inserts` new keys. Each batch is staged elsewhere and then renamed into the source directory so the +// streamer never sees a partially written batch. The expected final key count is written to +// <dir>/expected_total. +// Required confs: spark.hudi.rc.e2e.{dir,batches,initial,updates,inserts,interval.seconds} + +import org.apache.hadoop.fs.{FileSystem, Path} +import org.apache.spark.sql.functions._ + +def conf(key: String) = spark.conf.get(s"spark.hudi.rc.e2e.$key") +val dir = conf("dir") +val numBatches = conf("batches").toInt +val initial = conf("initial").toLong +val updates = conf("updates").toLong +val inserts = conf("inserts").toLong +val intervalMs = conf("interval.seconds").toLong * 1000 + +val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration) +var nextKey = 0L + +for (b <- 0 until numBatches) { + val start = System.currentTimeMillis() + val ids = if (b == 0) { + spark.range(0, initial).toDF("id") + } else { + val updated = spark.range(0, updates).select(pmod(hash(lit(b), col("id")), lit(nextKey)).cast("long").as("id")).distinct() + updated.union(spark.range(nextKey, nextKey + inserts).toDF("id")) + } Review Comment: 🤖 nit: `nextKey` is advanced after it is used as the upper bound for updates and the start of inserts, so the order matters; could you move the update to after `val df` or add a comment to make that clear? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## scripts/release/rc_validation/streamer_e2e_datagen.scala: ########## @@ -0,0 +1,71 @@ +/* + * 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. + */ + +// Feeds a ParquetDFSSource directory with parquet batches at a fixed interval, run through spark-shell -i. +// Batch 0 inserts `initial` keys; every later batch upserts up to `updates` existing keys and inserts +// `inserts` new keys. Each batch is staged elsewhere and then renamed into the source directory so the +// streamer never sees a partially written batch. The expected final key count is written to +// <dir>/expected_total. +// Required confs: spark.hudi.rc.e2e.{dir,batches,initial,updates,inserts,interval.seconds} + +import org.apache.hadoop.fs.{FileSystem, Path} +import org.apache.spark.sql.functions._ + +def conf(key: String) = spark.conf.get(s"spark.hudi.rc.e2e.$key") +val dir = conf("dir") +val numBatches = conf("batches").toInt +val initial = conf("initial").toLong +val updates = conf("updates").toLong +val inserts = conf("inserts").toLong +val intervalMs = conf("interval.seconds").toLong * 1000 + +val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration) +var nextKey = 0L + +for (b <- 0 until numBatches) { + val start = System.currentTimeMillis() + val ids = if (b == 0) { + spark.range(0, initial).toDF("id") + } else { + val updated = spark.range(0, updates).select(pmod(hash(lit(b), col("id")), lit(nextKey)).cast("long").as("id")).distinct() + updated.union(spark.range(nextKey, nextKey + inserts).toDF("id")) + } + nextKey = if (b == 0) initial else nextKey + inserts + val df = ids.select( + format_string("key-%012d", col("id")).as("uuid"), + lit(start).as("ts"), + lit(s"rider-$b").as("rider"), + lit(s"driver-$b").as("driver"), + (rand(b) * 100).as("fare"), + lit(b).as("batch"), + when(col("id") % 3 === 0, "san_francisco").when(col("id") % 3 === 1, "sao_paulo").otherwise("chennai").as("city")) + val staged = s"$dir/staging/batch_$b" + val batchSize = if (b == 0) initial else updates + inserts + df.repartition(math.max(1L, batchSize / 500000L).toInt).write.mode("overwrite").parquet(staged) + val files = fs.listStatus(new Path(staged)).map(_.getPath).filter(_.getName.endsWith(".parquet")) + files.zipWithIndex.foreach { case (p, i) => fs.rename(p, new Path(f"$dir/source/batch_$b%04d_$i%03d.parquet")) } Review Comment: 🤖 non-blocking: When a batch has more than one file (batch 0 at `large`), a streamer sync that runs between these renames checkpoints the first file's mtime. `DFSPathSelector` only takes files with mtime > checkpoint, so if the second file was written earlier it is skipped for good and the count check fails spuriously. Renaming the whole staged directory into `source/` would make each batch appear at once, since the selector recurses and ignores `_SUCCESS` and `.crc`: ```scala fs.rename(new Path(staged), new Path(f"$dir/source/batch_$b%04d")) ``` <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- 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]
