yihua commented on code in PR #5156: URL: https://github.com/apache/hudi/pull/5156#discussion_r844586429
########## hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/SparkDSContinuousIngestTool.java: ########## @@ -0,0 +1,167 @@ +/* + * 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. + */ + +package org.apache.hudi.integ.testsuite; + +import org.apache.hudi.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.integ.testsuite.SparkDataSourceContinuousIngest; +import org.apache.hudi.utilities.HoodieRepairTool; +import org.apache.hudi.utilities.IdentitySplitter; +import org.apache.hudi.utilities.UtilHelpers; + +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import org.apache.hadoop.fs.Path; +import org.apache.log4j.LogManager; +import org.apache.log4j.Logger; +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.SparkSession; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + + +/** + * Sample command + * + * ./bin/spark-submit --packages org.apache.spark:spark-avro_2.11:2.4.4 --driver-memory 4g --executor-memory 4g \ + * --conf spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.sql.catalogImplementation=hive \ + * --class org.apache.hudi.integ.testsuite.SparkDSContinuousIngestTool \ + * /Users/nsb/Documents/personal/projects/nov26/hudi/packaging/hudi-integ-test-bundle/target/hudi-integ-test-bundle-0.11.0-SNAPSHOT.jar \ + * --source-path file:///Users/nsb/Documents/personal/datasets/spark_ds_continuous --checkpoint-file-path /tmp/hudi/checkpoint \ Review Comment: nit: similar here ########## hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/SparkDSContinuousIngestTool.java: ########## @@ -0,0 +1,167 @@ +/* + * 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. + */ + +package org.apache.hudi.integ.testsuite; + +import org.apache.hudi.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.integ.testsuite.SparkDataSourceContinuousIngest; +import org.apache.hudi.utilities.HoodieRepairTool; +import org.apache.hudi.utilities.IdentitySplitter; +import org.apache.hudi.utilities.UtilHelpers; + +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import org.apache.hadoop.fs.Path; +import org.apache.log4j.LogManager; +import org.apache.log4j.Logger; +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.SparkSession; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + + +/** + * Sample command + * + * ./bin/spark-submit --packages org.apache.spark:spark-avro_2.11:2.4.4 --driver-memory 4g --executor-memory 4g \ + * --conf spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.sql.catalogImplementation=hive \ + * --class org.apache.hudi.integ.testsuite.SparkDSContinuousIngestTool \ + * /Users/nsb/Documents/personal/projects/nov26/hudi/packaging/hudi-integ-test-bundle/target/hudi-integ-test-bundle-0.11.0-SNAPSHOT.jar \ Review Comment: nit: make this jar path general. ########## hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/SparkDSContinuousIngestTool.java: ########## @@ -0,0 +1,167 @@ +/* + * 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. + */ + +package org.apache.hudi.integ.testsuite; + +import org.apache.hudi.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.integ.testsuite.SparkDataSourceContinuousIngest; +import org.apache.hudi.utilities.HoodieRepairTool; +import org.apache.hudi.utilities.IdentitySplitter; +import org.apache.hudi.utilities.UtilHelpers; + +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import org.apache.hadoop.fs.Path; +import org.apache.log4j.LogManager; +import org.apache.log4j.Logger; +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.SparkSession; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + + +/** + * Sample command + * + * ./bin/spark-submit --packages org.apache.spark:spark-avro_2.11:2.4.4 --driver-memory 4g --executor-memory 4g \ + * --conf spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.sql.catalogImplementation=hive \ + * --class org.apache.hudi.integ.testsuite.SparkDSContinuousIngestTool \ + * /Users/nsb/Documents/personal/projects/nov26/hudi/packaging/hudi-integ-test-bundle/target/hudi-integ-test-bundle-0.11.0-SNAPSHOT.jar \ + * --source-path file:///Users/nsb/Documents/personal/datasets/spark_ds_continuous --checkpoint-file-path /tmp/hudi/checkpoint \ + * --base-path file:///tmp/hudi/tbl_path/ --props /tmp/hudi_props.out + * + * Contents of hudi_props.out + * + * hoodie.insert.shuffle.parallelism=4 + * hoodie.upsert.shuffle.parallelism=4 + * hoodie.bulkinsert.shuffle.parallelism=4 + * hoodie.delete.shuffle.parallelism=4 + * hoodie.datasource.write.recordkey.field=VendorID + * hoodie.datasource.write.partitionpath.field=date_col + * hoodie.datasource.write.operation=upsert + * hoodie.datasource.write.precombine.field=tpep_pickup_datetime + * hoodie.metadata.enable=false + * hoodie.table.name=hudi_tbl + */ + +public class SparkDSContinuousIngestTool { Review Comment: nit: rename to `SparkDataSourceContinuousIngestTool` to avoid confusion with DeltaStreamer ########## hudi-integ-test/src/main/scala/org/apache/hudi/integ/testsuite/SparkDataSourceContinuousIngest.scala: ########## @@ -0,0 +1,93 @@ +/* + * 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. + */ + +package org.apache.hudi.integ.testsuite + +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.{FileStatus, FileSystem, Path, PathFilter} +import org.apache.log4j.LogManager +import org.apache.spark.sql.{SaveMode, SparkSession} + +import java.io.Serializable + +class SparkDataSourceContinuousIngest(val spark: SparkSession, val conf: Configuration, val sourcePath: Path, + val sourceFormat: String, val checkpointFile: Path, hudiBasePath: Path, hudiOptions: java.util.Map[String, String], + minSyncIntervalSeconds: Long) extends Serializable { + + private val log = LogManager.getLogger(getClass) + + def startIngestion(): Unit = { + val fs = sourcePath.getFileSystem(conf) + var orderedBatch : Array[FileStatus] = null + if (fs.exists(checkpointFile)) { + log.info("Checkpoint file exists. ") + val checkpoint = spark.sparkContext.textFile(checkpointFile.toString).collect()(0) + log.warn("Checkpoint to resume from " + checkpoint) + + orderedBatch = fetchListOfFilesToConsume(fs, sourcePath, new PathFilter { + override def accept(path: Path): Boolean = { + path.getName.toLong > checkpoint.toLong + } + }) + if (log.isDebugEnabled) { + log.debug("List of batches to consume in order ") + orderedBatch.foreach(entry => log.warn(" " + entry.getPath.getName)) + } + + } else { + log.warn("No checkpoint file exists. Starting from scratch ") + orderedBatch = fetchListOfFilesToConsume(fs, sourcePath, new PathFilter { + override def accept(path: Path): Boolean = { + true + } + }) + if (log.isDebugEnabled) { + log.debug("List of batches to consume in order ") + orderedBatch.foreach(entry => log.warn(" " + entry.getPath.getName)) + } + } + + orderedBatch.foreach(entry => { + log.info("Consuming from batch " + entry) + val pathToConsume = new Path(sourcePath.toString + "/" + entry.getPath.getName) + val df = spark.read.format(sourceFormat).load(pathToConsume.toString) + + df.write.format("hudi").options(hudiOptions).mode(SaveMode.Append).save(hudiBasePath.toString) + writeToFile(checkpointFile, entry.getPath.getName, fs) + log.info("Completed batch " + entry + ". Moving to next batch. Sleeping for " + minSyncIntervalSeconds + " secs before next batch") + Thread.sleep(minSyncIntervalSeconds * 1000) + }) + } + + def fetchListOfFilesToConsume(fs: FileSystem, basePath: Path, pathFilter: PathFilter): Array[FileStatus] = { + val nextBatches = fs.listStatus(basePath, pathFilter) + nextBatches.sortBy(fileStatus => fileStatus.getPath.getName.toLong) + } + + def writeToFile(checkpointFilePath: Path, str: String, fs: FileSystem): Unit = { + if (!fs.exists(checkpointFilePath)) { + fs.create(checkpointFilePath) + } + val fsOutStream = fs.create(checkpointFilePath, true) + fsOutStream.writeBytes(str) + fsOutStream.flush() + fsOutStream.close() + } + Review Comment: nit: remove empty line. ########## hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/SparkDSContinuousIngestTool.java: ########## @@ -0,0 +1,167 @@ +/* + * 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. + */ + +package org.apache.hudi.integ.testsuite; + +import org.apache.hudi.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.integ.testsuite.SparkDataSourceContinuousIngest; +import org.apache.hudi.utilities.HoodieRepairTool; +import org.apache.hudi.utilities.IdentitySplitter; +import org.apache.hudi.utilities.UtilHelpers; + +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import org.apache.hadoop.fs.Path; +import org.apache.log4j.LogManager; +import org.apache.log4j.Logger; +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.SparkSession; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + + +/** + * Sample command + * + * ./bin/spark-submit --packages org.apache.spark:spark-avro_2.11:2.4.4 --driver-memory 4g --executor-memory 4g \ + * --conf spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.sql.catalogImplementation=hive \ + * --class org.apache.hudi.integ.testsuite.SparkDSContinuousIngestTool \ + * /Users/nsb/Documents/personal/projects/nov26/hudi/packaging/hudi-integ-test-bundle/target/hudi-integ-test-bundle-0.11.0-SNAPSHOT.jar \ + * --source-path file:///Users/nsb/Documents/personal/datasets/spark_ds_continuous --checkpoint-file-path /tmp/hudi/checkpoint \ + * --base-path file:///tmp/hudi/tbl_path/ --props /tmp/hudi_props.out + * + * Contents of hudi_props.out Review Comment: nit: rename this to `hudi_props.properties` or `hudi.properties` ########## hudi-integ-test/src/main/scala/org/apache/hudi/integ/testsuite/SparkDataSourceContinuousIngest.scala: ########## @@ -0,0 +1,93 @@ +/* + * 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. + */ + +package org.apache.hudi.integ.testsuite + +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.{FileStatus, FileSystem, Path, PathFilter} +import org.apache.log4j.LogManager +import org.apache.spark.sql.{SaveMode, SparkSession} + +import java.io.Serializable + +class SparkDataSourceContinuousIngest(val spark: SparkSession, val conf: Configuration, val sourcePath: Path, Review Comment: No change is required. Just wondering why these two classes are in two different languages, Java and Scala? ########## hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/SparkDSContinuousIngestTool.java: ########## @@ -0,0 +1,167 @@ +/* + * 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. + */ + +package org.apache.hudi.integ.testsuite; + +import org.apache.hudi.client.common.HoodieSparkEngineContext; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.integ.testsuite.SparkDataSourceContinuousIngest; +import org.apache.hudi.utilities.HoodieRepairTool; +import org.apache.hudi.utilities.IdentitySplitter; +import org.apache.hudi.utilities.UtilHelpers; + +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import org.apache.hadoop.fs.Path; +import org.apache.log4j.LogManager; +import org.apache.log4j.Logger; +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.SparkSession; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + + +/** + * Sample command + * + * ./bin/spark-submit --packages org.apache.spark:spark-avro_2.11:2.4.4 --driver-memory 4g --executor-memory 4g \ + * --conf spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.sql.catalogImplementation=hive \ + * --class org.apache.hudi.integ.testsuite.SparkDSContinuousIngestTool \ + * /Users/nsb/Documents/personal/projects/nov26/hudi/packaging/hudi-integ-test-bundle/target/hudi-integ-test-bundle-0.11.0-SNAPSHOT.jar \ + * --source-path file:///Users/nsb/Documents/personal/datasets/spark_ds_continuous --checkpoint-file-path /tmp/hudi/checkpoint \ + * --base-path file:///tmp/hudi/tbl_path/ --props /tmp/hudi_props.out + * + * Contents of hudi_props.out + * + * hoodie.insert.shuffle.parallelism=4 + * hoodie.upsert.shuffle.parallelism=4 + * hoodie.bulkinsert.shuffle.parallelism=4 + * hoodie.delete.shuffle.parallelism=4 + * hoodie.datasource.write.recordkey.field=VendorID + * hoodie.datasource.write.partitionpath.field=date_col + * hoodie.datasource.write.operation=upsert + * hoodie.datasource.write.precombine.field=tpep_pickup_datetime + * hoodie.metadata.enable=false + * hoodie.table.name=hudi_tbl + */ + +public class SparkDSContinuousIngestTool { + + private static final Logger LOG = LogManager.getLogger(SparkDSContinuousIngestTool.class); + + private final Config cfg; + // Properties with source, hoodie client, key generator etc. + private TypedProperties props; + private HoodieSparkEngineContext context; + private SparkSession sparkSession; + + public SparkDSContinuousIngestTool(JavaSparkContext jsc, Config cfg) { + if (cfg.propsFilePath != null) { + cfg.propsFilePath = FSUtils.addSchemeIfLocalPath(cfg.propsFilePath).toString(); + } + this.context = new HoodieSparkEngineContext(jsc); + this.sparkSession = SparkSession.builder().config(jsc.getConf()).getOrCreate(); + this.cfg = cfg; + this.props = cfg.propsFilePath == null + ? UtilHelpers.buildProperties(cfg.configs) + : readConfigFromFileSystem(jsc, cfg); + } + + public static void main(String[] args) { + final Config cfg = new Config(); + JCommander cmd = new JCommander(cfg, null, args); + if (cfg.help || args.length == 0) { + cmd.usage(); + System.exit(1); + } + final JavaSparkContext jsc = UtilHelpers.buildSparkContext("hudi-table-repair", cfg.sparkMaster, cfg.sparkMemory); Review Comment: nit: rename the job name -- 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]
