You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by gr...@apache.org on 2019/03/22 18:06:45 UTC
[kudu] branch master updated (7acbdb3 -> 61546c0)
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git.
from 7acbdb3 KUDU-2687: fix ITClient retries (take two)
new 1356792 [scripts] Add initial test scripts for backup/restore testing
new 61546c0 [spark-tools] DistributedDataGenerator repartition support
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../spark/tools/DistributedDataGenerator.scala | 204 +++++++-----
.../spark/tools/DistributedDataGeneratorTest.scala | 74 ++++-
.../org/apache/kudu/spark/kudu/KuduContext.scala | 2 +-
src/kudu/scripts/backup-perf.py | 343 +++++++++++++++++++++
4 files changed, 552 insertions(+), 71 deletions(-)
create mode 100755 src/kudu/scripts/backup-perf.py
[kudu] 02/02: [spark-tools] DistributedDataGenerator repartition
support
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 61546c02517663bd4ac5879a9a4189e0a81e963b
Author: Grant Henke <gr...@apache.org>
AuthorDate: Fri Feb 8 10:52:42 2019 -0600
[spark-tools] DistributedDataGenerator repartition support
This patch adds support to the DistributedDataGenerator
to repartition the data to match the Kudu partitioning.
Because data generation is now decoupled from data
loading, this patch changes the collision handling
behavior. Instead of generating new data on collision,
now the collision is only tracked in the metrics.
Additionally this patch changes the default generation
type from random to sequential given that has been
shown to be the more common option and the type
of workload Kudu is better suited for.
Change-Id: I57bcc68d645c52b429ac6cf8bcdf0551a8244995
Reviewed-on: http://gerrit.cloudera.org:8080/12411
Tested-by: Kudu Jenkins
Reviewed-by: Adar Dembo <ad...@cloudera.com>
---
.../spark/tools/DistributedDataGenerator.scala | 204 ++++++++++++++-------
.../spark/tools/DistributedDataGeneratorTest.scala | 74 +++++++-
.../org/apache/kudu/spark/kudu/KuduContext.scala | 2 +-
3 files changed, 209 insertions(+), 71 deletions(-)
diff --git a/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/DistributedDataGenerator.scala b/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/DistributedDataGenerator.scala
index ee0954f..01af3ce 100644
--- a/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/DistributedDataGenerator.scala
+++ b/java/kudu-spark-tools/src/main/scala/org/apache/kudu/spark/tools/DistributedDataGenerator.scala
@@ -20,12 +20,17 @@ import java.math.BigDecimal
import java.math.BigInteger
import java.nio.charset.StandardCharsets
+import org.apache.kudu.Schema
import org.apache.kudu.Type
import org.apache.kudu.client.PartialRow
import org.apache.kudu.client.SessionConfiguration
import org.apache.kudu.spark.kudu.KuduContext
+import org.apache.kudu.spark.kudu.KuduWriteOptions
+import org.apache.kudu.spark.kudu.RowConverter
+import org.apache.kudu.spark.kudu.SparkUtil
import org.apache.kudu.spark.tools.DistributedDataGeneratorOptions._
import org.apache.kudu.util.DataGenerator
+import org.apache.spark.sql.Row
import org.apache.spark.sql.SparkSession
import org.apache.spark.util.LongAccumulator
import org.apache.spark.SparkConf
@@ -50,56 +55,138 @@ object GeneratorMetrics {
object DistributedDataGenerator {
val log: Logger = LoggerFactory.getLogger(getClass)
- def generateRows(
- context: KuduContext,
- options: DistributedDataGeneratorOptions,
- taskNum: Int,
- metrics: GeneratorMetrics) {
-
- val kuduClient = context.syncClient
- val session = kuduClient.newSession()
- session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND)
- val kuduTable = kuduClient.openTable(options.tableName)
-
- val generator = new DataGenerator.DataGeneratorBuilder()
- // Add taskNum to the seed otherwise each task will try to generate the same rows.
- .random(new java.util.Random(options.seed + taskNum))
- .stringLength(options.stringLength)
- .binaryLength(options.binaryLength)
- .build()
-
- val rowsToWrite = options.numRows / options.numTasks
- var currentRow: Long = rowsToWrite * taskNum
- var rowsWritten: Long = 0
- while (rowsWritten < rowsToWrite) {
- val insert = kuduTable.newInsert()
- if (options.generatorType == SequentialGenerator) {
- setRow(insert.getRow, currentRow)
- } else if (options.generatorType == RandomGenerator) {
- generator.randomizeRow(insert.getRow)
- }
- session.apply(insert)
+ def run(options: DistributedDataGeneratorOptions, ss: SparkSession): Unit = {
+ log.info(s"Running a DistributedDataGenerator with options: $options")
+ val sc = ss.sparkContext
+ val context = new KuduContext(options.masterAddresses, sc)
+ val metrics = GeneratorMetrics(sc)
+
+ // Generate the Inserts.
+ var rdd = sc
+ .parallelize(0 until options.numTasks, numSlices = options.numTasks)
+ .mapPartitions(
+ { taskNumIter =>
+ // We know there is only 1 task per partition because numSlices = options.numTasks above.
+ val taskNum = taskNumIter.next()
+ val generator = new DataGenerator.DataGeneratorBuilder()
+ // Add taskNum to the seed otherwise each task will try to generate the same rows.
+ .random(new java.util.Random(options.seed + taskNum))
+ .stringLength(options.stringLength)
+ .binaryLength(options.binaryLength)
+ .build()
+ val table = context.syncClient.openTable(options.tableName)
+ val schema = table.getSchema
+ val numRows = options.numRows / options.numTasks
+ val startRow: Long = numRows * taskNum
+ new GeneratedRowIterator(generator, options.generatorType, schema, startRow, numRows)
+ },
+ true
+ )
+
+ if (options.repartition) {
+ val table = context.syncClient.openTable(options.tableName)
+ val sparkSchema = SparkUtil.sparkSchema(table.getSchema)
+ rdd = context
+ .repartitionRows(rdd, options.tableName, sparkSchema, KuduWriteOptions(ignoreNull = true))
+ }
- // Synchronously flush on potentially the last iteration of the
- // loop, so we can check whether we need to retry any collisions.
- if (rowsWritten + 1 == rowsToWrite) session.flush()
+ // Write the rows to Kudu.
+ // TODO: Use context.writeRows while still tracking inserts/collisions.
+ rdd.foreachPartition { rows =>
+ val kuduClient = context.syncClient
+ val table = kuduClient.openTable(options.tableName)
+ val kuduSchema = table.getSchema
+ val sparkSchema = SparkUtil.sparkSchema(kuduSchema)
+ val converter = new RowConverter(kuduSchema, sparkSchema, ignoreNull = true)
+ val session = kuduClient.newSession()
+ session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND)
+
+ var rowsWritten = 0
+ rows.foreach { row =>
+ val insert = table.newInsert()
+ val partialRow = converter.toPartialRow(row)
+ insert.setRow(partialRow)
+ session.apply(insert)
+ rowsWritten += 1
+ }
+ // Synchronously flush after the last record is written.
+ session.flush()
+
+ // Track the collisions.
+ var collisions = 0
for (error <- session.getPendingErrors.getRowErrors) {
if (error.getErrorStatus.isAlreadyPresent) {
// Because we can't check for collisions every time, but instead
// only when the rows are flushed, we subtract any rows that may
// have failed from the counter.
rowsWritten -= 1
- metrics.collisions.add(1)
+ collisions += 1
} else {
throw new RuntimeException("Kudu write error: " + error.getErrorStatus.toString)
}
}
- currentRow += 1
- rowsWritten += 1
+ metrics.rowsWritten.add(rowsWritten)
+ metrics.collisions.add(collisions)
+ session.close()
}
- metrics.rowsWritten.add(rowsWritten)
- session.close()
+
+ log.info(s"Rows written: ${metrics.rowsWritten.value}")
+ log.info(s"Collisions: ${metrics.collisions.value}")
+ }
+
+ /**
+ * Entry point for testing. SparkContext is a singleton,
+ * so tests must create and manage their own.
+ */
+ @InterfaceAudience.LimitedPrivate(Array("Test"))
+ def testMain(args: Array[String], ss: SparkSession): Unit = {
+ DistributedDataGeneratorOptions.parse(args) match {
+ case None => throw new IllegalArgumentException("Could not parse arguments")
+ case Some(config) => run(config, ss)
+ }
+ }
+
+ def main(args: Array[String]): Unit = {
+ val conf = new SparkConf().setAppName("DistributedDataGenerator")
+ val ss = SparkSession.builder().config(conf).getOrCreate()
+ testMain(args, ss)
+ }
+}
+
+private class GeneratedRowIterator(
+ generator: DataGenerator,
+ generatorType: String,
+ schema: Schema,
+ startRow: Long,
+ numRows: Long)
+ extends Iterator[Row] {
+
+ val sparkSchema = SparkUtil.sparkSchema(schema)
+ // ignoreNull values so unset/defaulted rows can be passed through.
+ val converter = new RowConverter(schema, sparkSchema, ignoreNull = true)
+
+ var currentRow: Long = startRow
+ var rowsGenerated: Long = 0
+
+ override def hasNext: Boolean = rowsGenerated < numRows
+
+ override def next(): Row = {
+ if (rowsGenerated >= numRows) {
+ throw new IllegalStateException("Already generated all of the rows.")
+ }
+
+ val partialRow = schema.newPartialRow()
+ if (generatorType == SequentialGenerator) {
+ setRow(partialRow, currentRow)
+ } else if (generatorType == RandomGenerator) {
+ generator.randomizeRow(partialRow)
+ } else {
+ throw new IllegalArgumentException(s"Generator type of $generatorType is unsupported")
+ }
+ currentRow += 1
+ rowsGenerated += 1
+ converter.toRow(partialRow)
}
/**
@@ -142,35 +229,6 @@ object DistributedDataGenerator {
}
}
}
-
- def run(options: DistributedDataGeneratorOptions, ss: SparkSession): Unit = {
- log.info(s"Running a DistributedDataGenerator with options: $options")
- val sc = ss.sparkContext
- val context = new KuduContext(options.masterAddresses, sc)
- val metrics = GeneratorMetrics(sc)
- sc.parallelize(0 until options.numTasks, numSlices = options.numTasks)
- .foreachPartition(taskNum => generateRows(context, options, taskNum.next(), metrics))
- log.info(s"Rows written: ${metrics.rowsWritten.value}")
- log.info(s"Collisions: ${metrics.collisions.value}")
- }
-
- /**
- * Entry point for testing. SparkContext is a singleton,
- * so tests must create and manage their own.
- */
- @InterfaceAudience.LimitedPrivate(Array("Test"))
- def testMain(args: Array[String], ss: SparkSession): Unit = {
- DistributedDataGeneratorOptions.parse(args) match {
- case None => throw new IllegalArgumentException("Could not parse arguments")
- case Some(config) => run(config, ss)
- }
- }
-
- def main(args: Array[String]): Unit = {
- val conf = new SparkConf().setAppName("DistributedDataGenerator")
- val ss = SparkSession.builder().config(conf).getOrCreate()
- testMain(args, ss)
- }
}
@InterfaceAudience.Private
@@ -183,7 +241,8 @@ case class DistributedDataGeneratorOptions(
numTasks: Int = DistributedDataGeneratorOptions.DefaultNumTasks,
stringLength: Int = DistributedDataGeneratorOptions.DefaultStringLength,
binaryLength: Int = DistributedDataGeneratorOptions.DefaultStringLength,
- seed: Long = System.currentTimeMillis())
+ seed: Long = System.currentTimeMillis(),
+ repartition: Boolean = DistributedDataGeneratorOptions.DefaultRepartition)
@InterfaceAudience.Private
@InterfaceStability.Unstable
@@ -194,7 +253,8 @@ object DistributedDataGeneratorOptions {
val DefaultBinaryLength: Int = 128
val RandomGenerator: String = "random"
val SequentialGenerator: String = "sequential"
- val DefaultGeneratorType: String = RandomGenerator
+ val DefaultGeneratorType: String = SequentialGenerator
+ val DefaultRepartition: Boolean = false
private val parser: OptionParser[DistributedDataGeneratorOptions] =
new OptionParser[DistributedDataGeneratorOptions]("LoadRandomData") {
@@ -220,7 +280,8 @@ object DistributedDataGeneratorOptions {
opt[Int]("num-tasks")
.action((v, o) => o.copy(numTasks = v))
- .text(s"The total number of Spark tasks to generate. Default: ${DefaultNumTasks}")
+ .text(s"The total number of Spark tasks to use when generating data. " +
+ s"Default: ${DefaultNumTasks}")
.optional()
opt[Int]("string-length")
@@ -237,6 +298,11 @@ object DistributedDataGeneratorOptions {
.action((v, o) => o.copy(seed = v))
.text(s"The seed to use in the random data generator. " +
s"Default: `System.currentTimeMillis()`")
+
+ opt[Boolean]("repartition")
+ .action((v, o) => o.copy(repartition = v))
+ .text(s"Repartition the data to ensure each spark task talks to a minimal " +
+ s"set of tablet servers.")
}
def parse(args: Seq[String]): Option[DistributedDataGeneratorOptions] = {
diff --git a/java/kudu-spark-tools/src/test/scala/org/apache/kudu/spark/tools/DistributedDataGeneratorTest.scala b/java/kudu-spark-tools/src/test/scala/org/apache/kudu/spark/tools/DistributedDataGeneratorTest.scala
index e8c2e0d..37e2624 100644
--- a/java/kudu-spark-tools/src/test/scala/org/apache/kudu/spark/tools/DistributedDataGeneratorTest.scala
+++ b/java/kudu-spark-tools/src/test/scala/org/apache/kudu/spark/tools/DistributedDataGeneratorTest.scala
@@ -17,11 +17,14 @@
package org.apache.kudu.spark.tools
import org.apache.kudu.Type
+import org.apache.kudu.client.KuduPartitioner
import org.apache.kudu.spark.kudu.KuduTestSuite
import org.apache.kudu.test.RandomUtils
import org.apache.kudu.util.DecimalUtil
import org.apache.kudu.util.SchemaGenerator
import org.apache.spark.rdd.RDD
+import org.apache.spark.scheduler.SparkListener
+import org.apache.spark.scheduler.SparkListenerTaskEnd
import org.apache.spark.sql.Row
import org.junit.Test
import org.junit.Assert.assertEquals
@@ -51,7 +54,9 @@ class DistributedDataGeneratorTest extends KuduTestSuite {
randomTableName,
harness.getMasterAddressesAsString)
val rdd = runGeneratorTest(args)
- assertEquals(numRows, rdd.collect.length)
+ val collisions = ss.sparkContext.longAccumulator("row_collisions").value
+ // Collisions could cause the number of row to be less than the number set.
+ assertEquals(numRows - collisions, rdd.collect.length)
}
@Test
@@ -67,6 +72,73 @@ class DistributedDataGeneratorTest extends KuduTestSuite {
assertEquals(numRows, rdd.collect.length)
}
+ @Test
+ def testRepartitionData() {
+ val numRows = 100
+ val args = Array(
+ s"--num-rows=$numRows",
+ "--num-tasks=10",
+ "--type=sequential",
+ "--repartition=true",
+ randomTableName,
+ harness.getMasterAddressesAsString)
+ val rdd = runGeneratorTest(args)
+ assertEquals(numRows, rdd.collect.length)
+ }
+
+ @Test
+ def testNumTasks() {
+ // Add a SparkListener to count the number of tasks that end.
+ var actualNumTasks = 0
+ val listener = new SparkListener {
+ override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
+ actualNumTasks += 1
+ }
+ }
+ ss.sparkContext.addSparkListener(listener)
+
+ val numTasks = 8
+ val numRows = 100
+ val args = Array(
+ s"--num-rows=$numRows",
+ s"--num-tasks=$numTasks",
+ randomTableName,
+ harness.getMasterAddressesAsString)
+ runGeneratorTest(args)
+
+ assertEquals(numTasks, actualNumTasks)
+ }
+
+ @Test
+ def testNumTasksRepartition(): Unit = {
+ // Add a SparkListener to count the number of tasks that end.
+ var actualNumTasks = 0
+ val listener = new SparkListener {
+ override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
+ actualNumTasks += 1
+ }
+ }
+ ss.sparkContext.addSparkListener(listener)
+
+ val numTasks = 8
+ val numRows = 100
+ val args = Array(
+ s"--num-rows=$numRows",
+ s"--num-tasks=$numTasks",
+ "--repartition=true",
+ randomTableName,
+ harness.getMasterAddressesAsString)
+ runGeneratorTest(args)
+
+ val table = kuduContext.syncClient.openTable(randomTableName)
+ val numPartitions = new KuduPartitioner.KuduPartitionerBuilder(table).build().numPartitions()
+
+ // We expect the number of tasks to be equal to numTasks + numPartitions because numTasks tasks
+ // are run to generate the data then we repartition the data to match the table partitioning
+ // and numPartitions tasks load the data.
+ assertEquals(numTasks + numPartitions, actualNumTasks)
+ }
+
def runGeneratorTest(args: Array[String]): RDD[Row] = {
val schema = generator.randomSchema()
val options = generator.randomCreateTableOptions(schema)
diff --git a/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduContext.scala b/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduContext.scala
index 777e46f..28aec32 100644
--- a/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduContext.scala
+++ b/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduContext.scala
@@ -366,7 +366,7 @@ class KuduContext(val kuduMaster: String, sc: SparkContext, val socketReadTimeou
log.info(s"completed $operation ops: duration histogram: $durationHistogram")
}
- private def repartitionRows(
+ private[spark] def repartitionRows(
rdd: RDD[Row],
tableName: String,
schema: StructType,
[kudu] 01/02: [scripts] Add initial test scripts for backup/restore
testing
Posted by gr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
granthenke pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 13567924fcb4d0605d63c1ab96b7182755ea3ce5
Author: Grant Henke <gr...@apache.org>
AuthorDate: Mon Dec 17 16:39:05 2018 -0600
[scripts] Add initial test scripts for backup/restore testing
Adds a rough script to test the backup and restore
functionality. This script is rough around the edges but
I would like to get a base script committed for collaboration purposes.
Change-Id: I0c8efc3778b20687f0c1bc4c825f49f8f24e6d3b
Reviewed-on: http://gerrit.cloudera.org:8080/12212
Tested-by: Kudu Jenkins
Reviewed-by: Adar Dembo <ad...@cloudera.com>
---
src/kudu/scripts/backup-perf.py | 343 ++++++++++++++++++++++++++++++++++++++++
1 file changed, 343 insertions(+)
diff --git a/src/kudu/scripts/backup-perf.py b/src/kudu/scripts/backup-perf.py
new file mode 100755
index 0000000..430f6f4
--- /dev/null
+++ b/src/kudu/scripts/backup-perf.py
@@ -0,0 +1,343 @@
+#!/usr/bin/env python
+#
+# 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.
+#################################################################
+# WARNING: This code is not stable and is intended for internal testing only.
+#
+# Backup and Restore performance test driver.
+#
+# Example invocation:
+# backup-perf.py \
+# --spark-submit-command spark2-submit \
+# --kudu-spark-tools-jar kudu-spark2-tools_2.11-1.9.0-SNAPSHOT.jar
+# --kudu-backup-jar kudu-backup2_2.11-1.9.0-SNAPSHOT.jar
+# --num-tasks 20 \
+# --master-addresses a123.example.com \
+# --impalad-address a123.example.com \
+# --backup-path hdfs:///user/foo/backups \
+# --partitions 450 \
+# --table-data-size-mb 500000 \
+# test_table_1
+#################################################################
+
+import argparse
+import datetime
+import json
+import subprocess
+import sys
+import timeit
+
+from collections import OrderedDict
+
+class TickingTimer:
+ """ Timer to keep track of the period between ticks. """
+ def __init__(self):
+ self.last_tick_ = timeit.default_timer()
+
+ def tick(self):
+ """
+ Resets the tick timer and returns the duration between the last two calls
+ to tick(), or construction of this object, whichever was more recent.
+ """
+ prev_last_tick = self.last_tick_
+ self.last_tick_ = timeit.default_timer()
+ latest_tick_period = self.last_tick_ - prev_last_tick
+ return latest_tick_period
+
+ def last_tick_time(self):
+ """
+ Returns the clock time of the last tick, or construction of this object,
+ whichever was more recent.
+ """
+ return self.last_tick_
+
+def check_output(*popenargs, **kwargs):
+ r"""Run command with arguments and return its output as a byte string.
+ Backported from Python 2.7 as it's implemented as pure python on stdlib.
+ >>> check_output(['/usr/bin/python', '--version'])
+ Python 2.6.2
+ """
+ process = subprocess.Popen(stdout=subprocess.PIPE, *popenargs, **kwargs)
+ output, unused_err = process.communicate()
+ retcode = process.poll()
+ if retcode:
+ cmd = kwargs.get("args")
+ if cmd is None:
+ cmd = popenargs[0]
+ error = subprocess.CalledProcessError(retcode, cmd)
+ error.output = output
+ raise error
+ return output
+
+def parse_bool(s):
+ if s.lower() == 'true': return True
+ if s.lower() == 'false': return False
+ raise argparse.ArgumentTypeError('value must be true or false')
+
+def timestamp():
+ return datetime.datetime.now().isoformat()
+
+def run_command(opts, cmd):
+ """ Print the command and run it if not in dry-run mode. """
+ print(cmd)
+ if not opts.dryrun:
+ print(check_output(cmd, shell=True))
+
+def ensure_commands_available(opts):
+ run_command(opts, 'which impala-shell')
+ run_command(opts, 'which kudu')
+
+def get_restored_table_name(opts):
+ return opts.table_name + opts.table_restore_suffix
+
+# TODO: Change this to use the Kudu python API.
+# It's good dog-fooding and removes the Impala requirement.
+def create_table(opts, stats):
+ """ Create a Kudu table via impala-shell """
+ print("--------------------------------------")
+ print("Creating table %s" % (opts.table_name,))
+ print("--------------------------------------")
+ print(timestamp())
+ create_table_ddl = "CREATE TABLE %s (" % (opts.table_name,)
+ num_bigint_cols = opts.columns - opts.num_string_columns
+ assert(num_bigint_cols > 0)
+ for i in range(opts.columns):
+ coltype = 'STRING'
+ if i < num_bigint_cols: coltype = 'BIGINT'
+ if i > 0: create_table_ddl += ', '
+ create_table_ddl += "f%d %s" % (i, coltype)
+ if i == 0: create_table_ddl += ' PRIMARY KEY'
+ create_table_ddl += ") PARTITION BY HASH(f0) PARTITIONS %d STORED AS KUDU " % \
+ (opts.partitions, )
+ create_table_ddl += "TBLPROPERTIES ('kudu.num_tablet_replicas' = '%d')" % \
+ (opts.replication_factor, )
+
+ cmd = 'echo "%s" | impala-shell -i %s -f -' % (create_table_ddl, opts.impalad_address)
+ run_command(opts, cmd)
+
+def drop_created_table(opts, stats):
+ """ Drop the created Kudu table via impala-shell """
+ print("--------------------------------------")
+ print("Dropping created table %s" % (opts.table_name, ))
+ print("--------------------------------------")
+ print(timestamp())
+ sql = "DROP TABLE %s" % (opts.table_name, )
+ cmd = 'echo "%s" | impala-shell -i %s -f -' % (sql, opts.impalad_address)
+ run_command(opts, cmd)
+
+def drop_restored_table(opts, stats):
+ """ Drop the restored Kudu table via the kudu table delete command """
+ # TODO: This may no longer be needed if and when we integrate
+ # restoring HMS metadata and the table is restored as "Impala-managed".
+ print("--------------------------------------")
+ print("Dropping restored table %s" % (get_restored_table_name(opts), ))
+ print("--------------------------------------")
+ print(timestamp())
+ cmd = 'kudu table delete %s %s' % (opts.master_addresses, opts.table_prefix +
+ get_restored_table_name(opts))
+ run_command(opts, cmd)
+
+def load_table(opts, stats):
+ """ Load a table with data using the DistributedDataGenerator spark job """
+ print("--------------------------------------")
+ print("Loading table %s" % (opts.table_name,))
+ print("--------------------------------------")
+ print(timestamp())
+ # Example invocation:
+ # spark-submit --class org.apache.kudu.spark.tools.DistributedDataGenerator \
+ # kudu-spark2-tools_2.11-1.8.0-SNAPSHOT.jar \
+ # --type random \
+ # --num-rows 10000000 \
+ # --num-tasks 20 \
+ # impala::default.foo_test3 m123.example.com
+ CLASS_NAME = 'org.apache.kudu.spark.tools.DistributedDataGenerator'
+ # TODO: Non-string columns are assumed to be 8 bytes.
+ row_size_bytes = opts.num_string_columns * opts.string_field_len + \
+ (opts.columns - opts.num_string_columns) * 8
+ num_rows = opts.table_data_size_mb * 1024 * 1024 / row_size_bytes
+ print("INFO: Inserting %d rows of %d bytes each" % (num_rows, row_size_bytes))
+ stats['row_size_bytes'] = row_size_bytes
+ stats['num_rows'] = num_rows
+ cmd = "%s --class %s %s --type %s --num-rows %d --num-tasks %d %s %s" % \
+ (opts.spark_submit_command, CLASS_NAME, opts.kudu_spark_tools_jar,
+ opts.load_policy, num_rows, opts.load_num_tasks, opts.table_prefix + opts.table_name,
+ opts.master_addresses)
+ run_command(opts, cmd)
+
+def backup_table(opts, stats):
+ print("--------------------------------------")
+ print("Backing up table %s" % (opts.table_name,))
+ print("--------------------------------------")
+ print(timestamp())
+ CLASS_NAME = "org.apache.kudu.backup.KuduBackup"
+ cmd = "%s --class %s %s --kuduMasterAddresses %s --scanRequestTimeoutMs %d --path %s %s" % \
+ (opts.spark_submit_command, CLASS_NAME, opts.kudu_backup_jar,
+ opts.master_addresses, opts.scan_request_timeout_ms,
+ opts.backup_path, opts.table_prefix + opts.table_name)
+ run_command(opts, cmd)
+
+def restore_table(opts, stats):
+ print("--------------------------------------")
+ print("Restoring table %s as %s" % (opts.table_name, get_restored_table_name(opts)))
+ print("--------------------------------------")
+ print(timestamp())
+ CLASS_NAME = "org.apache.kudu.backup.KuduRestore"
+ cmd = "%s --class %s %s --tableSuffix %s --kuduMasterAddresses %s --path %s %s" % \
+ (opts.spark_submit_command, CLASS_NAME, opts.kudu_backup_jar,
+ opts.table_restore_suffix, opts.master_addresses, opts.backup_path, opts.table_prefix + opts.table_name)
+ run_command(opts, cmd)
+
+def parse_args():
+ """ Parse command-line arguments """
+ parser = argparse.ArgumentParser(description='Run a Kudu backup and restore performance test',
+ formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+
+ # Kudu Configuration
+ parser.add_argument('--master-addresses', required=True, help='The Kudu master addresses')
+
+ # Impala Configuration
+ parser.add_argument('--impalad-address', required=True, help='The Impala daemon address')
+ parser.add_argument('--table-prefix', default='impala::default.',
+ help='Kudu table name prefix in the Hive metastore')
+
+ # Spark Job Configuration
+ parser.add_argument('--spark-submit-command', default='spark-submit',
+ help='The name of the spark-submit binary')
+ ## Spark Loader Configuration
+ parser.add_argument('--kudu-spark-tools-jar', default='kudu-spark*-tools*.jar',
+ help='The path to the kudu-spark-tools jar (for --load-table)')
+ parser.add_argument('--load-num-tasks', type=int, default=20,
+ help='Number of Spark tasks to create when loading data')
+ parser.add_argument('--load-policy', default='sequential', choices=['sequential', 'random'],
+ help='The data loading policy for the data generator')
+ parser.add_argument('--string-field-len', type=int, default=128,
+ help='The length, in bytes, of generated string column values')
+
+ ## Spark Backup/Restore Job Configuration
+ parser.add_argument('--kudu-backup-jar', default='kudu-backup*.jar',
+ help='The path to the kudu-backup jar')
+ parser.add_argument('--backup-path', default='hdfs:///kudu-backup-tests',
+ help='The Hadoop-compatible path at which to store the backup')
+ parser.add_argument('--backup-file-format', default='parquet',
+ help='The file format of the backup: must be parquet')
+ parser.add_argument('--scan-request-timeout-ms', type=int, default=30000,
+ help='The default scan timeout for backup, in milliseconds')
+ parser.add_argument('--table-restore-suffix', default='-restore',
+ help='Kudu table name suffix to append on restore')
+
+ # Table Configuration
+ parser.add_argument('--columns', type=int, default=10,
+ help='The number of columns in the Kudu table')
+ parser.add_argument('--num-string-columns', type=int, default=9,
+ help='The number of string columns in the table; the rest will be bigints')
+ parser.add_argument('--partitions', type=int, default=10,
+ help='The number of hash partitions of the table. '
+ 'This script only supports hash partitions')
+ parser.add_argument('--table-data-size-mb', type=int, default=1024,
+ help='The uncompressed data size of the table, in MB')
+ parser.add_argument('--replication-factor', type=int, default=3,
+ help='The replication factor of the table')
+
+ # Positional
+ parser.add_argument('table_name', help='The name of the Kudu table to create/backup')
+
+ # Actions
+ parser.add_argument('--create-table', type=parse_bool, choices=[True, False], default=False,
+ help='Whether to create the table for loading')
+ parser.add_argument('--drop-created-table', type=parse_bool, choices=[True, False], default=False,
+ help='Whether to drop the created table after a successful test run')
+ parser.add_argument('--load-table', type=parse_bool, choices=[True, False], default=False,
+ help='Whether to load the table with data')
+ parser.add_argument('--backup-table', type=parse_bool, choices=[True, False], default=False,
+ help='Whether to back up the table')
+ parser.add_argument('--restore-table', type=parse_bool, choices=[True, False], default=False,
+ help='Whether to restore the table')
+ parser.add_argument('--drop-restored-table', type=parse_bool, choices=[True, False], default=False,
+ help='Whether to drop the restored table after a successful test run')
+
+ # Utility
+ parser.add_argument('--dryrun', action='store_true',
+ help='Do not execute any commands, only print what would be executed')
+
+ return parser.parse_args()
+
+def main():
+ start_timestamp = timestamp()
+ print(start_timestamp)
+ print("Starting perf test...")
+
+ opts = parse_args()
+
+ stats = OrderedDict()
+ stats['start_timestamp'] = start_timestamp
+
+ stats['columns'] = opts.columns
+ stats['num_string_columns'] = opts.num_string_columns
+ stats['partitions'] = opts.partitions
+ stats['table_data_size_mb'] = opts.table_data_size_mb
+ stats['replication_factor'] = opts.replication_factor
+ stats['load_num_tasks'] = opts.load_num_tasks
+ stats['load_policy'] = opts.load_policy
+ stats['string_field_len'] = opts.load_policy
+
+ timer = TickingTimer()
+ start = timer.last_tick_time()
+
+ if opts.create_table:
+ create_table(opts, stats)
+ stats['create_table_duration_sec'] = timer.tick()
+
+ if opts.load_table:
+ load_table(opts, stats)
+ stats['load_table_duration_sec'] = timer.tick()
+
+ if opts.backup_table:
+ backup_table(opts, stats)
+ stats['backup_table_duration_sec'] = timer.tick()
+
+ if opts.restore_table:
+ restore_table(opts, stats)
+ stats['restore_table_duration_sec'] = timer.tick()
+
+ if opts.drop_created_table:
+ drop_created_table(opts, stats)
+ stats['drop_created_table_duration_sec'] = timer.tick()
+
+ if opts.drop_restored_table:
+ drop_restored_table(opts, stats)
+ stats['drop_restored_table_duration_sec'] = timer.tick()
+
+ end = timer.last_tick_time()
+
+ stats['end_timestamp'] = timestamp()
+ print(stats['end_timestamp'])
+ print("Ending perf test")
+ total_duration = end - start
+ stats['total_duration_sec'] = total_duration
+ print("Total time elapsed: %s s" % (total_duration, ))
+
+ print("")
+ print("--------------------------------------")
+ print("[ BEGIN STATS ]")
+ print(json.dumps(stats,
+ indent=4, separators=(',', ': ')))
+ print("[ END STATS ]")
+ print("--------------------------------------")
+
+if __name__ == "__main__":
+ main()