You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by mp...@apache.org on 2019/05/30 16:56:26 UTC
[kudu] 04/04: [backup] KUDU-2787 Pt 2 Allow single table failures
for restore
This is an automated email from the ASF dual-hosted git repository.
mpercy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit b406853c279ee48471e0ce800d285a99a247458e
Author: Will Berkeley <wd...@gmail.com>
AuthorDate: Thu May 23 15:44:30 2019 -0700
[backup] KUDU-2787 Pt 2 Allow single table failures for restore
This path enhances restore to be able to survive the failures of
individual tables. It adds a new option `--failOnFirstError` to the
restore job, which defaults to false, and which when true causes the
restore job to fail as soon as any table has an error (the previous
behavior).
Change-Id: I9dcc0be78e637b47f35e494dce4a6df274c8d559
Reviewed-on: http://gerrit.cloudera.org:8080/13416
Reviewed-by: Mike Percy <mp...@apache.org>
Tested-by: Mike Percy <mp...@apache.org>
---
.../scala/org/apache/kudu/backup/KuduRestore.scala | 227 ++++++++++++---------
.../scala/org/apache/kudu/backup/Options.scala | 10 +-
.../org/apache/kudu/backup/TestKuduBackup.scala | 36 +++-
.../scala/org/apache/kudu/backup/TestOptions.scala | 1 +
4 files changed, 168 insertions(+), 106 deletions(-)
diff --git a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduRestore.scala b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduRestore.scala
index 1a5886f..7ea7599 100644
--- a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduRestore.scala
+++ b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/KuduRestore.scala
@@ -29,8 +29,10 @@ import org.apache.yetus.audience.InterfaceAudience
import org.apache.yetus.audience.InterfaceStability
import org.slf4j.Logger
import org.slf4j.LoggerFactory
-
import scala.collection.JavaConverters._
+import scala.util.Failure
+import scala.util.Success
+import scala.util.Try
/**
* The main class for a Kudu restore spark job.
@@ -40,7 +42,109 @@ import scala.collection.JavaConverters._
object KuduRestore {
val log: Logger = LoggerFactory.getLogger(getClass)
- def run(options: RestoreOptions, session: SparkSession): Unit = {
+ private def doRestore(
+ tableName: String,
+ context: KuduContext,
+ session: SparkSession,
+ io: BackupIO,
+ options: RestoreOptions,
+ backupMap: Map[String, BackupGraph]): Unit = {
+ if (!backupMap.contains(tableName)) {
+ throw new RuntimeException(s"No valid backups found for table: $tableName")
+ }
+ val graph = backupMap(tableName)
+ val lastMetadata = graph.restorePath.backups.last.metadata
+ val restoreName = s"${lastMetadata.getTableName}${options.tableSuffix}"
+ graph.restorePath.backups.foreach { backup =>
+ log.info(s"Restoring table $tableName from path: ${backup.path}")
+ val metadata = backup.metadata
+ val isFullRestore = metadata.getFromMs == 0
+ // TODO (KUDU-2788): Store the full metadata to compare/validate for each applied partial.
+
+ // On the full restore we may need to create the table.
+ if (isFullRestore) {
+ if (options.createTables) {
+ log.info(s"Creating restore table $restoreName")
+ // We use the last schema in the restore path when creating the table to
+ // ensure the table is created in its final state.
+ createTableRangePartitionByRangePartition(restoreName, lastMetadata, context)
+ }
+ }
+ val backupSchema = BackupUtils.dataSchema(TableMetadata.getKuduSchema(metadata))
+ val rowActionCol = backupSchema.fields.last.name
+ val table = context.syncClient.openTable(restoreName)
+
+ var data = session.sqlContext.read
+ .format(metadata.getDataFormat)
+ .schema(backupSchema)
+ .load(backup.path.toString)
+ // Default the the row action column with a value of "UPSERT" so that the
+ // rows from a full backup, which don't have a row action, are upserted.
+ .na
+ .fill(RowAction.UPSERT.getValue, Seq(rowActionCol))
+
+ // Adjust for dropped and renamed columns.
+ data = adjustSchema(data, metadata, lastMetadata, rowActionCol)
+ val restoreSchema = data.schema
+
+ // Write the data to Kudu.
+ data.queryExecution.toRdd.foreachPartition { internalRows =>
+ val table = context.syncClient.openTable(restoreName)
+ val converter = new RowConverter(table.getSchema, restoreSchema, false)
+ val partitioner = createPartitionFilter(metadata, lastMetadata)
+ val session = context.syncClient.newSession
+ session.setFlushMode(FlushMode.AUTO_FLUSH_BACKGROUND)
+ // In the case of task retries we need to ignore NotFound errors for deleted rows.
+ // TODO(KUDU-1563): Implement server side ignore capabilities to improve performance
+ // and reliability.
+ session.setIgnoreAllNotFoundRows(true)
+ try for (internalRow <- internalRows) {
+ // Convert the InternalRows to Rows.
+ // This avoids any corruption as reported in SPARK-26880.
+ val row = converter.toRow(internalRow)
+ // Get the operation type based on the row action column.
+ // This will always be the last column in the row.
+ val rowActionValue = row.getByte(row.length - 1)
+ val rowAction = RowAction.fromValue(rowActionValue)
+ // Generate an operation based on the row action.
+ val operation = rowAction match {
+ case RowAction.UPSERT => table.newUpsert()
+ case RowAction.DELETE => table.newDelete()
+ case _ => throw new IllegalStateException(s"Unsupported RowAction: $rowAction")
+ }
+ // Convert the Spark row to a partial row and set it on the operation.
+ val partialRow = converter.toPartialRow(row)
+ // Drop rows that are not covered by the partitioner. This is how we
+ // detect a partition which was dropped between backups and filter
+ // out the rows from that dropped partition.
+ if (partitioner.isCovered(partialRow)) {
+ operation.setRow(partialRow)
+ session.apply(operation)
+ }
+ } finally {
+ session.close()
+ }
+ // Fail the task if there are any errors.
+ // It is important to capture all of the errors via getRowErrors and then check
+ // the length because each call to session.getPendingErrors clears the ErrorCollector.
+ val pendingErrors = session.getPendingErrors
+ if (pendingErrors.getRowErrors.nonEmpty) {
+ val errors = pendingErrors.getRowErrors
+ val sample = errors.take(5).map(_.getErrorStatus).mkString
+ if (pendingErrors.isOverflowed) {
+ throw new RuntimeException(
+ s"PendingErrors overflowed. Failed to write at least ${errors.length} rows " +
+ s"to Kudu; Sample errors: $sample")
+ } else {
+ throw new RuntimeException(
+ s"Failed to write ${errors.length} rows to Kudu; Sample errors: $sample")
+ }
+ }
+ }
+ }
+ }
+
+ def run(options: RestoreOptions, session: SparkSession): Int = {
log.info(s"Restoring from path: ${options.rootPath}")
val context =
new KuduContext(
@@ -57,102 +161,33 @@ object KuduRestore {
.mapValues(_.maxBy(_.restorePath.toMs))
// TODO (KUDU-2786): Make parallel so each table isn't processed serially.
- // TODO (KUDU-2787): Handle single table failures.
- options.tables.foreach { tableName =>
- if (!backupMap.contains(tableName)) {
- throw new RuntimeException(s"No valid backups found for table: $tableName")
+ // TODO (KUDU-2832): If the job fails to restore a table it may still create the table, which
+ // will cause subsequent restores to fail unless the table is deleted or the restore suffix is
+ // changed. We ought to try to clean up the mess when a failure happens.
+ val restoreResults = options.tables.map { tableName =>
+ val restoreResult =
+ Try(doRestore(tableName: String, context, session, io, options, backupMap))
+ restoreResult match {
+ case Success(()) =>
+ log.info(s"Successfully restored table $tableName")
+ case Failure(ex) =>
+ if (options.failOnFirstError)
+ throw ex
+ else
+ log.error(s"Failed to restore table $tableName", ex)
}
- val graph = backupMap(tableName)
- val lastMetadata = graph.restorePath.backups.last.metadata
- val restoreName = s"${lastMetadata.getTableName}${options.tableSuffix}"
- graph.restorePath.backups.foreach { backup =>
- log.info(s"Restoring table $tableName from path: ${backup.path}")
- val metadata = backup.metadata
- val isFullRestore = metadata.getFromMs == 0
- // TODO (KUDU-2788): Store the full metadata to compare/validate for each applied partial.
-
- // On the full restore we may need to create the table.
- if (isFullRestore) {
- if (options.createTables) {
- log.info(s"Creating restore table $restoreName")
- // We use the last schema in the restore path when creating the table to
- // ensure the table is created in its final state.
- createTableRangePartitionByRangePartition(restoreName, lastMetadata, context)
- }
- }
- val backupSchema = BackupUtils.dataSchema(TableMetadata.getKuduSchema(metadata))
- val rowActionCol = backupSchema.fields.last.name
- val table = context.syncClient.openTable(restoreName)
-
- var data = session.sqlContext.read
- .format(metadata.getDataFormat)
- .schema(backupSchema)
- .load(backup.path.toString)
- // Default the the row action column with a value of "UPSERT" so that the
- // rows from a full backup, which don't have a row action, are upserted.
- .na
- .fill(RowAction.UPSERT.getValue, Seq(rowActionCol))
-
- // Adjust for dropped and renamed columns.
- data = adjustSchema(data, metadata, lastMetadata, rowActionCol)
- val restoreSchema = data.schema
+ (tableName, restoreResult)
+ }
- // Write the data to Kudu.
- data.queryExecution.toRdd.foreachPartition { internalRows =>
- val table = context.syncClient.openTable(restoreName)
- val converter = new RowConverter(table.getSchema, restoreSchema, false)
- val partitioner = createPartitionFilter(metadata, lastMetadata)
- val session = context.syncClient.newSession
- session.setFlushMode(FlushMode.AUTO_FLUSH_BACKGROUND)
- // In the case of task retries we need to ignore NotFound errors for deleted rows.
- // TODO(KUDU-1563): Implement server side ignore capabilities to improve performance
- // and reliability.
- session.setIgnoreAllNotFoundRows(true)
- try for (internalRow <- internalRows) {
- // Convert the InternalRows to Rows.
- // This avoids any corruption as reported in SPARK-26880.
- val row = converter.toRow(internalRow)
- // Get the operation type based on the row action column.
- // This will always be the last column in the row.
- val rowActionValue = row.getByte(row.length - 1)
- val rowAction = RowAction.fromValue(rowActionValue)
- // Generate an operation based on the row action.
- val operation = rowAction match {
- case RowAction.UPSERT => table.newUpsert()
- case RowAction.DELETE => table.newDelete()
- case _ => throw new IllegalStateException(s"Unsupported RowAction: $rowAction")
- }
- // Convert the Spark row to a partial row and set it on the operation.
- val partialRow = converter.toPartialRow(row)
- // Drop rows that are not covered by the partitioner. This is how we
- // detect a partition which was dropped between backups and filter
- // out the rows from that dropped partition.
- if (partitioner.isCovered(partialRow)) {
- operation.setRow(partialRow)
- session.apply(operation)
- }
- } finally {
- session.close()
- }
- // Fail the task if there are any errors.
- // It is important to capture all of the errors via getRowErrors and then check
- // the length because each call to session.getPendingErrors clears the ErrorCollector.
- val pendingErrors = session.getPendingErrors
- if (pendingErrors.getRowErrors.nonEmpty) {
- val errors = pendingErrors.getRowErrors
- val sample = errors.take(5).map(_.getErrorStatus).mkString
- if (pendingErrors.isOverflowed) {
- throw new RuntimeException(
- s"PendingErrors overflowed. Failed to write at least ${errors.length} rows " +
- s"to Kudu; Sample errors: $sample")
- } else {
- throw new RuntimeException(
- s"Failed to write ${errors.length} rows to Kudu; Sample errors: $sample")
- }
- }
- }
- }
+ restoreResults.filter(_._2.isFailure).foreach {
+ case (tableName, ex) =>
+ log.error(
+ s"Failed to restore table $tableName: Look back in the logs for the full exception. Error: ${ex.toString}")
}
+ if (restoreResults.exists(_._2.isFailure))
+ 1
+ else
+ 0
}
// Kudu isn't good at creating a lot of tablets at once, and by default tables may only be created
@@ -261,6 +296,6 @@ object KuduRestore {
.appName("Kudu Table Restore")
.getOrCreate()
- run(options, session)
+ System.exit(run(options, session))
}
}
diff --git a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/Options.scala b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/Options.scala
index 3db04a4..80fcbdc 100644
--- a/java/kudu-backup/src/main/scala/org/apache/kudu/backup/Options.scala
+++ b/java/kudu-backup/src/main/scala/org/apache/kudu/backup/Options.scala
@@ -165,10 +165,12 @@ case class RestoreOptions(
kuduMasterAddresses: String = InetAddress.getLocalHost.getCanonicalHostName,
tableSuffix: String = "",
createTables: Boolean = RestoreOptions.DefaultCreateTables,
- timestampMs: Long = System.currentTimeMillis())
+ timestampMs: Long = System.currentTimeMillis(),
+ failOnFirstError: Boolean = RestoreOptions.DefaultFailOnFirstError)
object RestoreOptions {
val DefaultCreateTables: Boolean = true
+ val DefaultFailOnFirstError = false
val ClassName: String = KuduRestore.getClass.getCanonicalName.dropRight(1) // Remove trailing `$`
val ProgramName: String = "spark-submit --class " + ClassName + " [spark-options] " +
@@ -204,6 +206,12 @@ object RestoreOptions {
"selecting restore candidates. Default: `System.currentTimeMillis()`")
.optional()
+ opt[Unit]("failOnFirstError")
+ .action((v, o) => o.copy(failOnFirstError = true))
+ .text("Whether to fail the restore job as soon as a single table restore fails. " +
+ "Default: " + DefaultFailOnFirstError)
+ .optional()
+
help("help").text("prints this usage text")
arg[String]("<table>...")
diff --git a/java/kudu-backup/src/test/scala/org/apache/kudu/backup/TestKuduBackup.scala b/java/kudu-backup/src/test/scala/org/apache/kudu/backup/TestKuduBackup.scala
index cdc96b3..255d27e 100644
--- a/java/kudu-backup/src/test/scala/org/apache/kudu/backup/TestKuduBackup.scala
+++ b/java/kudu-backup/src/test/scala/org/apache/kudu/backup/TestKuduBackup.scala
@@ -122,7 +122,7 @@ class TestKuduBackup extends KuduTestSuite {
}
@Test
- def testFailedTableDoesNotFailOtherTables() {
+ def testFailedTableBackupDoesNotFailOtherTableBackups() {
insertRows(table, 100) // Insert data into the default test table.
// Run a fail-fast backup. It should fail because the first table doesn't exist.
@@ -150,14 +150,32 @@ class TestKuduBackup extends KuduTestSuite {
@Test
def testRestoreWithNoBackup(): Unit = {
+ // Check that a restore of a table with no backups fails fast with an exception if the
+ // fail-on-first-error option is set.
+ val failFastOptions = createRestoreOptions(Seq(tableName)).copy(failOnFirstError = true)
try {
- val options = createRestoreOptions(Seq(tableName))
- doRestore(options)
+ assertEquals(1, runRestore(failFastOptions))
fail()
} catch {
case e: RuntimeException =>
assertEquals(e.getMessage, s"No valid backups found for table: $tableName")
}
+
+ // Check that a restore of a table with no backups does not fail fast or throw an exception if
+ // default no-fail-fast option is set.
+ assertEquals(1, runRestore(createRestoreOptions(Seq("missingTable"))))
+ }
+
+ @Test
+ def testFailedTableRestoreDoesNotFailOtherTableRestores() {
+ insertRows(table, 100)
+ KuduBackup.run(createBackupOptions(Seq(tableName)), ss)
+
+ // There's no guarantee about the order restores run in, so it doesn't work to test fail-fast
+ // and then the default no-fail-fast because the actual table may have been restored.
+ captureLogs(
+ () => assertEquals(1, runRestore(createRestoreOptions(Seq("missingTable", tableName)))))
+ .contains("Failed to restore table")
}
@Test
@@ -227,7 +245,7 @@ class TestKuduBackup extends KuduTestSuite {
backupAndValidateTable(tableName, incrementalRows.length, true)
}
- doRestore(createRestoreOptions(Seq(tableName)))
+ assertEquals(0, runRestore(createRestoreOptions(Seq(tableName))))
validateTablesMatch(tableName, s"$tableName-restore")
}
@@ -244,7 +262,7 @@ class TestKuduBackup extends KuduTestSuite {
insertRows(table2, numRows)
assertEquals(0, runBackup(createBackupOptions(Seq(table1Name, table2Name))))
- doRestore(createRestoreOptions(Seq(table1Name, table2Name)))
+ assertEquals(0, runRestore(createRestoreOptions(Seq(table1Name, table2Name))))
val rdd1 =
kuduContext.kuduRDD(ss.sparkContext, s"$table1Name-restore", List("key"))
@@ -367,7 +385,7 @@ class TestKuduBackup extends KuduTestSuite {
backupAndValidateTable(tableName, 10, true)
// Restore the table and validate.
- doRestore(createRestoreOptions(Seq(tableName)))
+ assertEquals(0, runRestore(createRestoreOptions(Seq(tableName))))
val restoreTable = kuduClient.openTable(s"$tableName-restore")
val scanner = kuduClient.newScannerBuilder(restoreTable).build()
@@ -430,7 +448,7 @@ class TestKuduBackup extends KuduTestSuite {
backupAndValidateTable(tableName, 15, true)
// Restore the table and validate.
- doRestore(createRestoreOptions(Seq(tableName)))
+ assertEquals(0, runRestore(createRestoreOptions(Seq(tableName))))
val restoreTable = kuduClient.openTable(s"$tableName-restore")
val scanner = kuduClient.newScannerBuilder(restoreTable).build()
@@ -631,14 +649,14 @@ class TestKuduBackup extends KuduTestSuite {
def restoreAndValidateTable(tableName: String, expectedRowCount: Long) = {
val options = createRestoreOptions(Seq(tableName))
- doRestore(options)
+ assertEquals(0, runRestore(options))
// Verify the table data.
val rdd = kuduContext.kuduRDD(ss.sparkContext, s"$tableName-restore")
assertEquals(rdd.collect.length, expectedRowCount)
}
- def doRestore(options: RestoreOptions): Unit = {
+ def runRestore(options: RestoreOptions): Int = {
KuduRestore.run(options, ss)
}
diff --git a/java/kudu-backup/src/test/scala/org/apache/kudu/backup/TestOptions.scala b/java/kudu-backup/src/test/scala/org/apache/kudu/backup/TestOptions.scala
index fdb24c6..a844f39 100644
--- a/java/kudu-backup/src/test/scala/org/apache/kudu/backup/TestOptions.scala
+++ b/java/kudu-backup/src/test/scala/org/apache/kudu/backup/TestOptions.scala
@@ -55,6 +55,7 @@ class TestOptions extends KuduTestSuite {
| --createTables <value> If true, create the tables during restore. Set to false if the target tables already exist. Default: true
| --tableSuffix <value> If set, the suffix to add to the restored table names. Only used when createTables is true.
| --timestampMs <value> A UNIX timestamp in milliseconds that defines the latest time to use when selecting restore candidates. Default: `System.currentTimeMillis()`
+ | --failOnFirstError Whether to fail the restore job as soon as a single table restore fails. Default: false
| --help prints this usage text
| <table>... A list of tables to be restored.""".stripMargin
assertEquals(expectedStr, RestoreOptions.parser.renderTwoColumnsUsage)