[HUDI-438] Merge duplicated code fragment #1114
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -73,7 +73,6 @@ private[hudi] object HoodieSparkSqlWriter { | |
| parameters(OPERATION_OPT_KEY) | ||
| } | ||
|
|
||
| var commitTime: String = null | ||
| var writeStatuses: JavaRDD[WriteStatus] = null | ||
|
|
||
|
|
@@ -85,7 +84,7 @@ private[hudi] object HoodieSparkSqlWriter { | |
| // Running into issues wrt generic type conversion from Java to Scala. Couldn't make common code paths for | ||
| // write and deletes. Specifically, instantiating client of type HoodieWriteClient<T extends HoodieRecordPayload> | ||
| // is having issues. Hence some codes blocks are same in both if and else blocks. | ||
| val writeSuccessful = if (!operation.equalsIgnoreCase(DELETE_OPERATION_OPT_VAL)) { | ||
| // register classes & schemas | ||
| val structName = s"${tblName.get}_record" | ||
| val nameSpace = s"hoodie.${tblName.get}" | ||
|
|
@@ -148,52 +147,7 @@ private[hudi] object HoodieSparkSqlWriter { | |
| commitTime = client.startCommit() | ||
| writeStatuses = DataSourceUtils.doWriteOperation(client, hoodieRecords, commitTime, operation) | ||
| // Check for errors and commit the write. | ||
| checkWriteStatus(writeStatuses, parameters, client, commitTime, basePath, operation, jsc) | ||
| } else { | ||
|
|
||
| // Handle save modes | ||
|
|
@@ -225,52 +179,7 @@ private[hudi] object HoodieSparkSqlWriter { | |
| // Issue deletes | ||
| commitTime = client.startCommit() | ||
| writeStatuses = DataSourceUtils.doDeleteOperation(client, hoodieKeysToDelete, commitTime) | ||
| checkWriteStatus(writeStatuses, parameters, client, commitTime, basePath, operation, jsc) | ||
|
Contributor
There was a problem hiding this comment. this is good enough. But did you get a chance to try to get checkWriteStatus(writeStatuses, parameters, client, commitTime, basePath, operation, jsc) outside the if else block. in other words, instead of calling both inside if and else, can we call it after if else block. If thats doable.
Contributor
Author
There was a problem hiding this comment.
@nsivabalan After some try, it was move out to the outside of if else block. |
||
| } | ||
|
|
||
| (writeSuccessful, common.util.Option.ofNullable(commitTime)) | ||
|
|
@@ -339,4 +248,58 @@ private[hudi] object HoodieSparkSqlWriter { | |
| hiveSyncConfig.partitionValueExtractorClass = parameters(HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY) | ||
| hiveSyncConfig | ||
| } | ||
|
|
||
| private def checkWriteStatus(writeStatuses: JavaRDD[WriteStatus], | ||
| parameters: Map[String, String], | ||
| client: HoodieWriteClient[_], | ||
| commitTime: String, | ||
| basePath: Path, | ||
| operation: String, | ||
| jsc: JavaSparkContext): Boolean = { | ||
| val errorCount = writeStatuses.rdd.filter(ws => ws.hasErrors).count() | ||
| if (errorCount == 0) { | ||
| log.info("No errors. Proceeding to commit the write.") | ||
| val metaMap = parameters.filter(kv => | ||
| kv._1.startsWith(parameters(COMMIT_METADATA_KEYPREFIX_OPT_KEY))) | ||
| val commitSuccess = if (metaMap.isEmpty) { | ||
| client.commit(commitTime, writeStatuses) | ||
| } else { | ||
| client.commit(commitTime, writeStatuses, | ||
| common.util.Option.of(new util.HashMap[String, String](mapAsJavaMap(metaMap)))) | ||
| } | ||
|
|
||
| if (commitSuccess) { | ||
| log.info("Commit " + commitTime + " successful!") | ||
| } | ||
| else { | ||
| log.info("Commit " + commitTime + " failed!") | ||
| } | ||
|
|
||
| val hiveSyncEnabled = parameters.get(HIVE_SYNC_ENABLED_OPT_KEY).exists(r => r.toBoolean) | ||
| val syncHiveSucess = if (hiveSyncEnabled) { | ||
| log.info("Syncing to Hive Metastore (URL: " + parameters(HIVE_URL_OPT_KEY) + ")") | ||
| val fs = FSUtils.getFs(basePath.toString, jsc.hadoopConfiguration) | ||
| syncHive(basePath, fs, parameters) | ||
| } else { | ||
| true | ||
| } | ||
| client.close() | ||
| commitSuccess && syncHiveSucess | ||
| } else { | ||
| log.error(s"$operation failed with ${errorCount} errors :"); | ||
| if (log.isTraceEnabled) { | ||
| log.trace("Printing out the top 100 errors") | ||
| writeStatuses.rdd.filter(ws => ws.hasErrors) | ||
| .take(100) | ||
| .foreach(ws => { | ||
| log.trace("Global error :", ws.getGlobalError) | ||
| if (ws.getErrors.size() > 0) { | ||
| ws.getErrors.foreach(kt => | ||
| log.trace(s"Error for key: ${kt._1}", kt._2)) | ||
| } | ||
| }) | ||
| } | ||
| false | ||
| } | ||
| } | ||
| } | ||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
can you remove the above comment
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Move to checkWriteStatus's javadoc.