After the GA of Apache Kudu in Cloudera CDH 5.10, we take a look at the Apache Spark on Kudu integration, share code snippets, and explain how to get up and running quickly, as Kudu is already a first-class citizen in Spark’s ecosystem.
As the Apache Kudu development team celebrates the initial 1.0 release launched on September 19, and the most recent 1.2.0 version now GA as part of Cloudera’s CDH 5.10 release, we take a look at Apache Spark and the capabilities already in place for working with Kudu.
The Spark integration with Kudu supports:
-
DDL operations (Create/Delete)
-
Native Kudu RDD
-
Native Kudu Data Source, for DataFrame integration
-
Reading from Kudu
-
Performing insert/update/upsert/delete from Kudu
-
Predicate pushdown
-
Schema mapping between Kudu and Spark SQL
Kudu IS:
-
a replicated and distributed storage engine for fast analytics and fast data
-
a storage engine that provides a balance between high throughput for large scans and low latency for random access and updates
-
a storage engine that provides database-like semantics and a relational data model
Kudu is NOT:
-
a file system
-
an application running on HDFS
-
a replacement for HDFS nor for HBase
Kudu is configured given directories on pre-defined, typical Linux file systems where the table data actually resides. You may dedicate file systems to Kudu alone, or even assign a directory for Kudu table data next to an already existing directory servicing HDFS. For example, HDFS may be assigned /data1/dfs for HDFS data while Kudu may be configured to store its data in /data1/kudu.
SQL access is available for Kudu tables using SQL engines written that support Kudu as the storage layer. Currently, Impala and Spark SQL provide that capability. Kudu is a complementary technology to HDFS and HBase as it provides fast sequential scans and fast random access though not scans as fast as sequential scans as Parquet on HDFS or random access as fast as HBase. It also does not provide the NoSQL ad hoc column creation capabilities of HBase and the variety of data formats stored in HDFS, as Kudu mandates structure and strong typing on the content it stores.
Spark is a processing engine running on top of Kudu, allowing one to integrate various datasets, whether they be on HDFS, HBase, Kudu or other storage engines, into a single application providing a unified view of your data. Spark SQL in particular nicely aligns with Kudu as Kudu tables already contain a strongly-typed, relational data model.
Always refer to the latest documentation found in the Developing Applications with Apache Kudu online documentation. All examples in this blog post may be found on Github.
At a high level, start your Spark application development by defining the following in your pom.xml file as we build our project using Maven.
Maven repository element
</repository>
Next, include the Maven dependencies for the kudu-client and kudu-spark_2.10 (2.10 referring to Scala level) artifacts.
Maven artifact dependencies
Make note that the version specified in the example above may change with upcoming releases.
By now you may have heard about several contexts such as SparkContext, SQLContext, HiveContext, SparkSession, and now, with Kudu, we introduce a KuduContext. This is the primary serializable object that can be broadcasted in your Spark application. This class interacts with the Kudu Java client on your behalf in your Spark executors.
KuduContext provides the methods needed to perform DDL operations, interface with the native Kudu RDD, perform updates/inserts/deletes on your data, convert data types from Kudu to Spark, and more.
Much of the implementation provided here you don’t need to worry about. Just know that such a context exists, and that you will likely interact with this and the DataFrame APIs when working with Kudu in Spark.
Common preamble code
| // Create a Spark and SQL contextval sc = new SparkContext(sparkConf)val sqlContext = new SQLContext(sc) // Comma-separated list of Kudu masters with port numbersval master1 = “ip-10-13-4-249.ec2.internal:7051”val master2 = “ip-10-13-5-150.ec2.internal:7051”val master3 = “ip-10-13-5-56.ec2.internal:7051”val kuduMasters = Seq(master1, master2, master3).mkString(“,”) // Create an instance of a KuduContextval kuduContext = new KuduContext(kuduMasters) |
val sc = new SparkContext(sparkConf)
val master1 = “ip-10-13-4-249.ec2.internal:7051”
val master3 = “ip-10-13-5-56.ec2.internal:7051”
val kuduContext = new KuduContext(kuduMasters)
In the above example, we start as usual defining a SparkContext with a SQLContext. To get started with the KuduContext, you simply supply it your list of Kudu Master hostnames with associated port numbers and you’re off to the races. If you are using the default port number in your installation, 7051, then you do not need to specify the port numbers in this list.
We start with examples on how to define your Kudu tables via Spark. First, the following show simple, yet ever useful, table ‘exists’ and ‘delete’ methods.
Table exists and delete
| // Specify a table namevar kuduTableName = “spark_kudu_tbl” // Check if the table exists, and drop it if it doesif (kuduContext.tableExists(kuduTableName)) { kuduContext.deleteTable(kuduTableName)} |
var kuduTableName = “spark_kudu_tbl”
// Check if the table exists, and drop it if it does
kuduContext.deleteTable(kuduTableName)
We now define our Kudu table in five steps:
-
Provide the table name
-
Provide the schema
-
Provide the primary key
-
Define important options like describing your partitioning schema
-
Call the create table API.
Be sure to refer to Apache Kudu Schema Design documentation for hints and tips on defining your table appropriately for your use case. Keep in mind that schema design is the single most important thing within your control to maximize the performance of your Kudu cluster.
Create table
| 1234567891011121314151617181920212223 | // 1. Give your table a namekuduTableName = “spark_kudu_tbl” // 2. Define a schemaval kuduTableSchema = StructType( // col name type nullable? StructField(“name”, StringType , false) :: StructField(“age” , IntegerType, true ) :: StructField(“city”, StringType , true ) :: Nil) // 3. Define the primary keyval kuduPrimaryKey = Seq(“name”) // 4. Specify any further optionsval kuduTableOptions = new CreateTableOptions()kuduTableOptions. setRangePartitionColumns(List(“name”).asJava). setNumReplicas(3) // 5. Call create table APIkuduContext.createTable( // Table name, schema, primary key and options kuduTableName, kuduTableSchema, kuduPrimaryKey, kuduTableOptions) |
2
4
6
8
10
12
14
16
18
20
22
kuduTableName = “spark_kudu_tbl”
// 2. Define a schema
// col name type nullable?
StructField(“age” , IntegerType, true ) ::
val kuduPrimaryKey = Seq(“name”)
// 4. Specify any further options
kuduTableOptions.
setNumReplicas(3)
// 5. Call create table API
// Table name, schema, primary key and options
One item to note when defining your table is in the Kudu table options values. You’ll notice we call the “asJava” method when specifying a List of column names making up the range partition columns. This is because here, we’re calling into the Kudu Java client itself, which requires Java objects (ie. java.util.List) instead of Scala’s List object.
To make the “asJava” method available, remember to import the JavaConverters libraries.
Import JavaConverters to specify Java types
| import scala.collection.JavaConverters._ |
After creating your table, take a look at the Kudu master UI by pointing your browser to http://
Next, you can see the list of tablets representing this table along with which host is currently acting as the leader tablet.
Finally, if you do decide in the future to create this table using Impala, the CREATE TABLE statement is shown for you as reference.
Kudu comes with a custom, native Data Source for Kudu tables. Hence, DataFrame APIs are tightly integrated. To demonstrate this, we define a DataFrame we’re going to work with, then show the capabilities available through this API.
Define your DataFrame
DataFrames can be created from many sources, including an existing RDD, Hive table, or from Spark data. Here, we will define a tiny dataset of Customers, convert into an RDD, and from there get our DataFrame.
Creating a simple dataset, converting it into a DataFrame
| 12345678910111213141516171819 | // Define your case class outside your main methodcase class Customer(name:String, age:Int, city:String) // This allows us to implicitly convert RDD to a DataFrameimport sqlContext.implicits. // Define a list of customers based on the case class already defined aboveval customers = Array( Customer(“jane”, 30, “new york”), Customer(“jordan”, 18, “toronto”)) // Create RDD out of the customers Arrayval customersRDD = sc.parallelize(customers) // Now, using reflection, this RDD can easily be converted to a DataFrame// Ensure to do the :// import sqlContext.implicits.// above to have the toDF() function available to youval customersDF = customersRDD.toDF() |
2
4
6
8
10
12
14
16
18
case class Customer(name:String, age:Int, city:String)
// This allows us to implicitly convert RDD to a DataFrame
val customers = Array(
Customer(“jordan”, 18, “toronto”))
// Create RDD out of the customers Array
// Ensure to do the :
// above to have the toDF() function available to you
Kudu supports a number of DML type operations, several of which are included in the Spark on Kudu integration. Supported Spark operations on Kudu DataFrame objects include:
-
INSERT – Insert rows of the DataFrame into the Kudu table. Note that although the API fully supports INSERT, the use of it within Spark is discouraged. It is risky to use INSERT because Spark tasks may require re-execution, which means rows inserted already may be requested to be inserted again. Doing so will result in failure, since INSERT will not allow rows to be inserted if they already exist (causes a failure). Instead, we encourage the use of INSERT_IGNORE described below.
-
INSERT-IGNORE – Insert rows of the DataFrame into the Kudu table. Ignore records if they already exist in the Kudu table.
-
DELETE – Delete rows found in the DataFrame from the Kudu table
-
UPSERT – Rows in the DataFrame are updated in the Kudu table if they exist, otherwise they are inserted.
-
UPDATE – Rows in the DataFrame are updated in the Kudu table
It is recommended to use the KuduContext for these operations, although as you will see later, many of these can also be done through the DataFrame API.
Insert data
| 12345678910111213141516171819 | // Define Kudu options used by various operationsval kuduOptions: Map[String, String] = Map( “kudu.table” -> kuduTableName, “kudu.master” -> kuduMasters) // 1. Specify your Kudu table namekuduTableName = “spark_kudu_tbl” // 2. Insert our customer DataFrame data set into the Kudu tablekuduContext.insertRows(customersDF, kuduTableName) // 3. Read back the records from the Kudu table to see them dumpedsqlContext.read.options(kuduOptions).kudu.show+——+—+——–+ | name | age | city | +——+—+——–+ | jane | 30 | new york | jordan | 18 | toronto | +——+—+——–+ |
2
4
6
8
10
12
14
16
18
val kuduOptions: Map[String, String] = Map(
“kudu.master” -> kuduMasters)
// 1. Specify your Kudu table name
kuduContext.insertRows(customersDF, kuduTableName)
// 3. Read back the records from the Kudu table to see them dumped
+——+—+——–+
| jordan | 18 | toronto |
Next, we will filter rows from our customers DataFrame, those who are older than 20, and delete those records, which removes ‘jane’ (aged 30) from our table.
Delete data
| 1234567891011121314151617181920 | // 1. Specify your Kudu table namekuduTableName = “spark_kudu_tbl” // 2. Let’s register our customer dataframe as a temporary table so we// refer to it in Spark SQLcustomersDF.registerTempTable(“customers”) // 3. Filter and create a keys-only DataFrame to be deleted from our tableval deleteKeysDF = sqlContext.sql(“select name from customers where age > 20”) // 4. Delete the rows from our Kudu tablekuduContext.deleteRows(deleteKeysDF, kuduTableName) // 5. Read data from Kudu tablesqlContext.read.options(kuduOptions).kudu.show+——+—+——-+ | name | age | city | +——+—+——-+ | jordan | 18 | toronto | +——+—+——-+ |
2
4
6
8
10
12
14
16
18
20
kuduTableName = “spark_kudu_tbl”
// 2. Let’s register our customer dataframe as a temporary table so we
customersDF.registerTempTable(“customers”)
// 3. Filter and create a keys-only DataFrame to be deleted from our table
kuduContext.deleteRows(deleteKeysDF, kuduTableName)
// 5. Read data from Kudu table
+——+—+——-+
At this point, we’ve removed the row for ‘jane’.
Our customer Jordan just had a birthday, and we’ve onboarded a number of new customers. We want to perform an upsert now, where ‘jordan’ will get an updated record, and we’ll have several new customers inserted.
Upsert data
| 12345678910111213141516171819202122232425 | // 1. Specify your Kudu table namekuduTableName = “spark_kudu_tbl” // 2. Define the dataset we want to upsertval newAndChangedCustomers = Array( Customer(“michael”, 25, “chicago”), Customer(“denise” , 43, “winnipeg”), Customer(“jordan” , 19, “toronto”)) // 3. Create our dataframeval newAndChangedRDD = sc.parallelize(newAndChangedCustomers)val newAndChangedDF = newAndChangedRDD.toDF() // 4. Call upsert with our new and changed customers DataFramekuduContext.upsertRows(newAndChangedDF, kuduTableName) // 5. Show contents of Kudu tablesqlContext.read.options(kuduOptions).kudu.show+——-+—+——–+ | name | age | city | +——-+—+——–+ | denise | 43 | winnipeg | jordan | 19 | toronto | michael | 25 | chicago | +——-+—+——–+ |
2
4
6
8
10
12
14
16
18
20
22
24
kuduTableName = “spark_kudu_tbl”
// 2. Define the dataset we want to upsert
Customer(“michael”, 25, “chicago”),
Customer(“jordan” , 19, “toronto”))
// 3. Create our dataframe
val newAndChangedDF = newAndChangedRDD.toDF()
// 4. Call upsert with our new and changed customers DataFrame
sqlContext.read.options(kuduOptions).kudu.show
| name | age | city |
| denise | 43 | winnipeg |
| michael | 25 | chicago |
Since Toronto is such a great city, let’s have Michael move to Toronto. To do so, we need a key and the column(s) that we want to update.
Update data
| 1234567891011121314151617181920 | // 1. Specify your Kudu table namekuduTableName = “spark_kudu_tbl” // 2. Create a DataFrame of updated rowsval modifiedCustomers = Array(Customer(“michael”, 25, “toronto”))val modifiedCustomersRDD = sc.parallelize(modifiedCustomers)val modifiedCustomersDF = modifiedCustomersRDD.toDF() // 3. Call update with our new and changed customers DataFramekuduContext.updateRows(modifiedCustomersDF, kuduTableName) // 4. Show contents of Kudu tablesqlContext.read.options(kuduOptions).kudu.show+——-+—+——–+ | name | age | city | +——-+—+——–+ | denise | 43 | winnipeg | jordan | 19 | toronto | michael | 25 | toronto | +——-+—+——–+ |
2
4
6
8
10
12
14
16
18
20
kuduTableName = “spark_kudu_tbl”
// 2. Create a DataFrame of updated rows
val modifiedCustomersRDD = sc.parallelize(modifiedCustomers)
kuduContext.updateRows(modifiedCustomersDF, kuduTableName)
// 4. Show contents of Kudu table
+——-+—+——–+
| jordan | 19 | toronto |
+——-+—+——–+
Spark integration with Kudu also provides you with a native Kudu RDD. Reading in the RDD provides you with a RDD[Row] type of objects. The only element you want to supply is the list of columns you want to project from the underlying table and away you go.
Reading with Native Kudu RDD
| 123456789101112131415161718 | // 1. Specify a table namekuduTableName = “spark_kudu_tbl” // 2. Specify the columns you want to projectval kuduTableProjColumns = Seq(“name”, “age”) // 3. Read table, represented now as RDDval custRDD = kuduContext.kuduRDD(sc, kuduTableName, kuduTableProjColumns) // We get a RDD[Row] coming back to us. Lets send through a map to pull// out the name and age into the form of a tupleval custTuple = custRDD.map { case Row(name: String, age: Int) => (name, age) } // Print it on the screen just for funcustTuple.collect().foreach(println(_))(jordan,19)(michael,25)(denise,43) |
2
4
6
8
10
12
14
16
18
kuduTableName = “spark_kudu_tbl”
// 2. Specify the columns you want to project
val custRDD = kuduContext.kuduRDD(sc, kuduTableName, kuduTableProjColumns)
// We get a RDD[Row] coming back to us. Lets send through a map to pull
val custTuple = custRDD.map { case Row(name: String, age: Int) => (name, age) }
// Print it on the screen just for fun
(jordan,19)
(denise,43)
While we can perform a number of manipulations through the KuduContext shown above, we also have the ability to call the read/write APIs straight from the default data source itself.
To setup a read, we need to specify options for the Kudu table naming the table we want to read alongside the list of Kudu master servers of the Kudu cluster servicing the table.
DataFrame read
| // Read our table into a DataFrame - reusing kuduOptions specified// aboveval customerReadDF = sqlContext.read.options(kuduOptions).kudu // Show our table to the screen.customerReadDF.show()+——-+—+——–+ | name | age | city | +——-+—+——–+ | jordan | 19 | toronto | michael | 25 | toronto | denise | 43 | winnipeg | +——-+—+——–+ |
// above
customerReadDF.show()
| name | age | city |
| jordan | 19 | toronto |
| denise | 43 | winnipeg |
When writing through the DataFrame API, currently only one mode, “append” is supported. The “overwrite” mode, not yet implemented, will likely be treated as a traditional TRUNCATE command in SQL, removing all contents of the table before inserting the contents of the DataFrame.
In any case, “append” mode with Kudu defaults behaviour to “upsert”; rows will be updated if the key already exists otherwise rows are inserted into the table.
DataFrame write
| 12345678910111213141516171819202122232425 | // Create a small dataset to write (append) to the Kudu tableval customersAppend = Array( Customer(“bob”, 30, “boston”), Customer(“charlie”, 23, “san francisco”)) // Create our DataFrame our of our datasetval customersAppendDF = sc.parallelize(customersAppend).toDF() // Specify the table namekuduTableName = “spark_kudu_tbl” // Call the write method on our DataFrame directly in “append” modecustomersAppendDF.write.options(kuduOptions).mode(“append”).kudu // See results of our appendsqlContext.read.options(kuduOptions).kudu.show()+——-+—+————-+ | name | age | city | +——-+—+————-+ | bob | 30 | boston | charlie | 23 | san francisco | denise | 43 | winnipeg | jordan | 19 | toronto | michael | 25 | toronto | +——-+—+————-+ |
2
4
6
8
10
12
14
16
18
20
22
24
val customersAppend = Array(
Customer(“charlie”, 23, “san francisco”))
// Create our DataFrame our of our dataset
kuduTableName = “spark_kudu_tbl”
// Call the write method on our DataFrame directly in “append” mode
sqlContext.read.options(kuduOptions).kudu.show()
| name | age | city |
| bob | 30 | boston |
| denise | 43 | winnipeg |
| michael | 25 | toronto |
You can also choose to write to a Kudu table using Spark SQL directly with an INSERT statement. Similar to ‘append’, the INSERT statement will actually be treated with UPSERT semantics by default. The INSERT OVERWRITE statement is not implemented, but when it is, will likely be treated as a TRUNCATE statement (ie. table’s contents are removed, and DataFrame contents will be fully written)
Spark SQL INSERT
| 12345678910111213141516171819202122232425262728293031323334353637 | // Quickly prepare a Kudu table we will use as our source table in Spark// SQL.// First, some sample dataval srcTableData = Array( Customer(“enzo”, 43, “oakland”), Customer(“laura”, 27, “vancouver”)) // Create our DataFrameval srcTableDF = sc.parallelize(srcTableData).toDF() // Register our source tablesrcTableDF.registerTempTable(“source_table”) // Specify Kudu table name we will be inserting intokuduTableName = “spark_kudu_tbl” // Register your table as a Spark SQL table.// Remember that kuduOptions stores the kuduTableName already as well as// the list of Kudu masters.sqlContext.read.options(kuduOptions).kudu.registerTempTable(kuduTableName) // Use Spark SQL to INSERT (treated as UPSERT by default) into Kudu tablesqlContext.sql(s”INSERT INTO TABLE $kuduTableName SELECT * FROM source_table”) // See results of our insertsqlContext.read.options(kuduOptions).kudu.show()+——-+—+————-+ | name | age | city | +——-+—+————-+ | michael | 25 | toronto | bob | 30 | boston | charlie | 23 | san francisco | denise | 43 | winnipeg | enzo | 43 | oakland | jordan | 19 | toronto | laura | 27 | vancouver | +——-+—+————-+ |
2
4
6
8
10
12
14
16
18
20
22
24
26
28
30
32
34
36
// SQL.
val srcTableData = Array(
Customer(“laura”, 27, “vancouver”))
// Create our DataFrame
srcTableDF.registerTempTable(“source_table”)
// Specify Kudu table name we will be inserting into
// Remember that kuduOptions stores the kuduTableName already as well as
sqlContext.read.options(kuduOptions).kudu.registerTempTable(kuduTableName)
// Use Spark SQL to INSERT (treated as UPSERT by default) into Kudu table
sqlContext.read.options(kuduOptions).kudu.show()
| name | age | city |
| michael | 25 | toronto |
| charlie | 23 | san francisco |
| enzo | 43 | oakland |
| laura | 27 | vancouver |
Pushing predicate evaluation down into the Kudu engine improves performance as it reduces the amount of data that needs to flow back to the Spark engine for further evaluation and processing.
The set of predicates that are currently supported for predicate pushdown through the Spark API include:
-
Equal to (=)
-
Greater than (>)
-
Greater than or equal (>=)
-
Less than (<)
-
Less than or equal (<=)
Hence, such statements in Spark SQL will push the predicate evaluation down into Kudu’s storage engine, improving overall performance.
Predicate pushdown
| 1234567891011121314151617181920 | // Kudu table namekuduTableName = “spark_kudu_tbl” // Register Kudu table as a Spark SQL temp tablesqlContext.read.options(kuduOptions).kudu. registerTempTable(kuduTableName) // Now refer to that temp table name in our Spark SQL statementval customerNameAgeDF = sqlContext. sql(s”"”SELECT name, age FROM $kuduTableName WHERE age >= 30”””) // Show the resultscustomerNameAgeDF.show()+——+—+ | name | age | +——+—+ | bob | 30 | denise | 43 | enzo | 43 | +——+—+ |
2
4
6
8
10
12
14
16
18
20
kuduTableName = “spark_kudu_tbl”
// Register Kudu table as a Spark SQL temp table
registerTempTable(kuduTableName)
// Now refer to that temp table name in our Spark SQL statement
sql(s”"”SELECT name, age FROM $kuduTableName WHERE age >= 30”””)
// Show the results
+——+—+
| denise | 43 |
+——+—+
If you point your browser to a tablet server such as at, http://
This would have to be reviewed while the query is running, which of course, is not practical especially when scans are too quick to spot in the UI.
Using Spark’s explain() function, you can also validate that your predicates are being pushed down as so (continued from the previous example of the predicate age >= 30)
| customerNameAgeDF.explain()== Physical Plan ==Scan org.apache.kudu.spark.kudu.KuduRelation@682615b3[name#6,age#7] PushedFilters: [GreaterThanOrEqual(age,30)] |
== Physical Plan ==
If you have multiple filters being pushed down to the Kudu storage layer, they get added to the array of PushedFilters seen above, similar to:
| PushedFilters: [GreaterThanOrEqual(age,30), GreaterThan(name,a)] |
Kudu and Spark SQL are altogether separate entities and engines. Spark is a processing framework, while Kudu a storage engine. Therefore, they have their own data types and schemas. Integration is already in place where Spark SQL schemas will be mapped accordingly to Kudu schemas under the covers already for you (with a few current limitations, see Spark Integration Known Issues and Limitations for details). Because of this, no additional work to be done on your end!
In this blog post, we’ve walked you through several aspects of the Apache Spark on Kudu integration. We have shown examples from setting up your application build properties, to defining your Kudu tables to showing various ways of how to interact with your Kudu tables through Spark. Kudu is now a first-class citizen in the Spark ecosystem, and hopefully by now you can start processing all your data through Spark whether it exists in Kudu or any other Hadoop storage engine.
