boneanxs commented on code in PR #6046:
URL: https://github.com/apache/hudi/pull/6046#discussion_r974817505
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/HoodieDatasetBulkInsertHelper.scala:
##########
@@ -105,6 +108,52 @@ object HoodieDatasetBulkInsertHelper extends Logging {
partitioner.repartitionRecords(trimmedDF,
config.getBulkInsertShuffleParallelism)
}
+ /**
+ * Perform bulk insert for [[Dataset<Row>]], will not change timeline/index,
return
+ * information about write files.
+ */
+ def bulkInsert(dataset: Dataset[Row],
+ instantTime: String,
+ table: HoodieTable[_ <: HoodieRecordPayload[_ <:
HoodieRecordPayload[_ <: AnyRef]], _, _, _],
+ writeConfig: HoodieWriteConfig,
+ partitioner: BulkInsertPartitioner[Dataset[Row]],
+ parallelism: Int,
+ shouldPreserveHoodieMetadata: Boolean):
HoodieData[WriteStatus] = {
+ val repartitionedDataset = partitioner.repartitionRecords(dataset,
parallelism)
+ val arePartitionRecordsSorted = partitioner.arePartitionRecordsSorted
+ val schema = dataset.schema
+ val writeStatuses =
repartitionedDataset.queryExecution.toRdd.mapPartitions(iter => {
+ val taskContextSupplier: TaskContextSupplier =
table.getTaskContextSupplier
+ val taskPartitionId = taskContextSupplier.getPartitionIdSupplier.get
+ val taskId = taskContextSupplier.getStageIdSupplier.get.toLong
+ val taskEpochId = taskContextSupplier.getAttemptIdSupplier.get
+ val writer = new BulkInsertDataInternalWriterHelper(
+ table,
+ writeConfig,
+ instantTime,
+ taskPartitionId,
+ taskId,
+ taskEpochId,
+ schema,
+ writeConfig.populateMetaFields,
+ arePartitionRecordsSorted,
+ shouldPreserveHoodieMetadata)
+
+ try {
+ iter.foreach(writer.write)
+ } catch {
+ case t: Throwable =>
+ writer.abort()
+ throw t
+ } finally {
+ writer.close()
+ }
+
+ writer.getWriteStatuses.asScala.map(_.toWriteStatus).iterator
+ }).collect()
+ table.getContext.parallelize(writeStatuses.toList.asJava)
Review Comment:
`writeStatuses` is an `Array`, but parallelize func needs `list`
--
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: commits-unsubscr...@hudi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
