Spark 1.6.x

Use the kudu-spark_2.10 artifact if you are using Spark with Scala 2.10. For example:

Spark 2.x

Use the kudu-spark2_2.11 artifact if you are using Spark 2 with Scala 2.11. For example:

Inserting Data

Create some sample data.

Insert the DataFrame. Note the name of the table. This format is needed if the table was created in Impala. In this example, default is the name of the database and customers is the name of the table. This convention was adopted to prevent table name collision between tables created in Impala and tables created natively using the Kudu API or Spark.

Confirm that the data was successfully inserted.

Updating a Kudu Table

Create an updated data set. Note that we modified the last name and the age.

Update the table.

Confirm that the table was successfully updated.

Upserting Data

Create some sample data.

Upsert data – update all columns if the primary key exists, and insert rows if it doesn’t.

Confirm if the data was successfully upserted.

Deleting Data

Inspect the data in the table.

Register the table so that we can use the table in our SQL query.

Delete the data based on our query.

Confirm that the data was successfully deleted.

Selecting Data

Select the data in the table.

You can also run your query by registering the table and using it in a SQL query. Note that you should use createOrReplaceTempView if you are using Spark 2.x.

Creating a Kudu Table

This table will not be visible to Impala since this table was created in Spark. You have to create an external table in Impala and refer to this table.

Inserting CSV into Kudu

Let’s insert multiple rows from a CSV file.

Because we created the customer2 table via Spark we only had to specify the table name (customer2) instead of impala::default.customer2.

Inserting CSV into Kudu Using the spark-csv Package

We can also use the spark-csv package to process the sample CSV data. The command below will automatically download the spark-csv dependencies so make sure you have an Internet connection in your cluster node. Note the use of the comma separating the list of packages.

Because we created the customer2 table via Spark, we only had to specify the table name (customer2) instead of impala::default.customer2.

Insert CSV into Kudu by Programmatically Specifying the Schema

You can use StructType to define a schema for your data set. Programmatically specifying a schema is helpful when the schema cannot be determined ahead of time.

Read a CSV file from HDFS.

Map CSV data to an RDD

Create a schema.

Insert DataFrame to Kudu.

Remember that we created the customers table via Impala. Therefore, we need to use the format impala::database.table when referring to the customers table.

Inserting XML into Kudu Using the spark-xml Package

We’ll create an XML file as a sample data for this example. We need to copy the file to HDFS.

We’ll use the spark-xml package to process the sample XML data. This package works similarly to the spark-csv package. The command below will automatically download the spark-xml package so make sure you have an Internet connection in your cluster node. Include the kudu-spark dependency as well.

Create a DataFrame using Spark XML. In this example, we specify the row tag and the path in HDFS where the XML file is located.

Let’s also take a look at the data.

Let’s inspect the schema.

Now let’s go back to the impala-shell and compare the schema with the structure of the users table. As you can see, the data type of the age and zip columns in the users table is different from the corresponding columns in the data frame. We’ll get an error message if we try to insert this DataFrame into the Kudu table.

We’ll need to cast the data types before inserting the DataFrame to the Kudu table. We’re introducing the use of selectExpr method here to convert data types, but another option is to programmatically specify the schema using StructType.

Create the kudu context and insert the DataFrame to the destination table.

It looks like the DataFrame was successfully inserted into the Kudu table. Using the impala-shell, check the data in the table to confirm.

Inserting JSON into Kudu

We’ll create a JSON file as sample data for this example. Make sure the file is in a folder in HDFS called /jsondata.

Create a DataFrame from the JSON file.

Check the data.

Check the schema.

Convert the data type of the age column to tinyint to match the table’s data type.

Create the kudu context and insert the DataFrame to the destination table.

Using the impala-shell, check if the rows were successfully inserted.

Inserting from MySQL into Kudu

Let’s populate the users table in MySQL with some test data. Make sure the users table exists in the salesdb database. We will insert this data to a table in Kudu.

Start the spark-shell. Take note that I had to include the MySQL driver as a parameter in both the –driver-class-path and –jars.

Let’s set up the jdbc url and connection properties.

We can create a DataFrame from an entire table.

Let’s take advantage of pushdown optimization, running the query in the database and only returning the required results.

Let's specify an entire query. This is a more convenient and flexible method. Also, unlike the previous method, we are not required to specify the columns in our SELECT list if the column is specified in the WHERE clause.

We’ve just tried different ways on how to select data from a MySQL table.

Let’ just go ahead and insert the whole table to Kudu.

Verify the schema.

But first let’s convert age from integer to TINYINT. Otherwise you won’t be able to insert this DataFrame into Kudu. Again, we could’ve just defined a schema using StructType.

As you can see, the data type of the age column is now TINYINT. Let’s go ahead and insert the data to Kudu.

Now go to impala-shell and check if the data was successfully inserted.

Alternatively, you can also check using Spark.

Inserting from SQL Server into Kudu

The first thing you need to do is download the Microsoft JDBC driver for SQL Server. You can download the JDBC driver here: https://docs.microsoft.com/en-us/sql/connect/jdbc/microsoft-jdbc-driver-for-sql-server .

You should see a page similar to the one in Figure 6-1. Click the “Download JDBC Driver” link.

Choose the language and click “Download.”

Untar and unzip the Tarball. Choose the driver based on the version of your JRE.

I’m going to use SQL Server 2016 throughout this book. You will also need to install SQL Server Management Studio separately (Figure 6-2).

We’ll create the salesdb database and the users table. In the Object Explorer, right-click the Databases node and click “New Databases” (Figure 6-3). You will be shown a window where you can specify the database name and other database configuration options. Enter a database name “salesdb” and click OK. For testing purposes we’ll leave the other options with the default values.

Expand the salesdb node. Right-click “Tables,” click “New,” then “Table.” Fill in the column names and data types. To make it easier for you to follow the examples in the book, make sure you keep the column names and data types the same as the MySQL and Kudu tables. Click the “Save” icon near the upper-right corner of the window, and then enter the table name “users” (Figure 6-4).

Let’s insert some test data into the users table we just created. Click “New Query” button on the standard toolbar to open a new editor window (Figure 6-5).

We’re going to copy these four rows from SQL Server to Kudu.

Start the spark-shell. Don’t forget to pass the SQL Server driver as a parameter in both the -driver-class-path and -jars.

Let’s set up the jdbc url and connection properties.

We can create a DataFrame from an entire table.

Let’s take advantage of pushdown optimization to run the query in the database and only return the results.

We can specify an entire query.

Let’s just go ahead and insert the entire table to Kudu.

Checking the schema, it looks like the age was converted into an integer.

We need to convert age from integer to TINYINT. Otherwise you won’t be able to insert this DataFrame into kudu.

As you can see, the data type of the age column is now TINYINT. Let’s go ahead and insert the DataFrame to Kudu.

Let’s now go to impala-shell and confirm if the data was successfully inserted.

Alternatively, you can also check using Spark.

Inserting from HBase into Kudu

There are several ways to transfer data from HBase to Kudu. We can use the HBase client API. There’s the Spark-HBase connector developed by Hortonworks. ^iv Astro, developed by Huwawei, provides an SQL layer HBase using Spark SQL. ^v The SparkOnHBase project from Cloudera was integrated into HBase recently, but it will probably take time before it makes it into a release of HBase. ^vi

We’ll use the most straightforward way by using JDBC. This might not be the fastest way to move data from HBase to Kudu, but it should be adequate for most tasks. We’ll create a Hive table on top of the HBase table, and then we’ll create a Spark DataFrame using JDBC via Impala. Once we have the DataFrame, we can easily insert it to Kudu.

The first thing we need to do is download the Impala JDBC driver. Point your browser to https://www.cloudera.com/downloads.html (Figure 6-6).

Clicking the Impala JDBC Driver Downloads link near the bottom right of the page will bring you to the download page of the latest version of the JDBC driver for Impala. Download and unzip the files.

The first thing we need to do is to start “hbase shell” and create the HBase table. We also need to add test data to our HBase table. Consult the Apache HBase Reference Guide online if you are unfamiliar with the HBase commands.

Create the Hive external table on top of the HBase table.

Using the impala-shell, verify that you can see the Hive external table.

It’s not showing up. We need to invalidate the metadata to refresh Impala’s memory.

Start the spark-shell.

Create a DataFrame from the HBase table

We still need to cast age to TINYINT before we can insert the data to the Kudu users table. Defining the schema using StructType is an option here.

We can now insert the data to Kudu.

Confirm if the data was successfully inserted.

The rows were successfully inserted.

Inserting from Solr into Kudu

As discussed in Chapter 5, you can access Solr from Spark using SolrJ. ^vii

A much better way to access Solr collections from Spark is by using the spark-solr package. Lucidworks started the spark-solr project to provide Spark-Solr integration. ^viii Using spark-solr is so much easier and powerful compared to SolrJ, allowing you to create DataFrames from Solr collections and using SparkSQL to interact with them. You can download the jar file from Lucidworks’s website.

Start by importing the jar file from spark-shell.

Specify the collection and connection information.

Create a DataFrame.

Insert the data to Kudu.

Insert from Amazon S3 into Kudu

Amazon S3 is a popular object store frequently used as data store for transient clusters. It’s also a cost-effective storage for backups and cold data. Reading data from S3 is just like reading data from HDFS or any other file system.

Read a CSV file from Amazon S3. Make sure you’ve configured your S3 credentials.

Map CSV data to an RDD.

Create a schema.

Insert DataFrame to Kudu.

You’ve successfully inserted data from S3 into Kudu.

We’ve inserted data from different data sources into Kudu. Let’s now insert data from Kudu to different data sources.

Inserting from Kudu into MySQL

Start the spark-shell.

Connect to the Kudu master and check the data in the users table. We’re going to sync this Kudu table with a MySQL table.

Register the DataFrame so we can run SQL queries against it.

Set up the JDBC URL and connection properties of the MySQL database.

Let’s sync both tables by inserting all the rows with userid < 300 from Kudu to MySQL.

Check the MySQL table again and verify that the rows were added.

It looks like the rows were successfully inserted.

Inserting from Kudu into SQL Server

Start spark-shell.

Create a DataFrame from the users table in the default database.

Verify the contents of the DataFrame.

Register the DataFrame so we can run SQL queries against it.

Set up the JDBC URL and connection properties of the SQL Server database.

To make sure our test is consistent, make sure the users table in SQL Server is empty using SQL Server Management Studio (Figure 6-7).

Insert the data from Kudu to SQL Server.

Check the SQL Server table again and verify that the rows were added (Figure 6-8).

Congratulations! The data was successfully inserted.

Inserting from Kudu into Oracle

The first thing we need to do is set up the Oracle environment. We’ll create the user table in an existing pluggable database called EDWPDB. Log in as sysdba and start the instance. Consult the online Oracle documentation if you are unfamiliar with Oracle.

Open the EDWPDB pluggable database and set it as the current container.

Create the Oracle table.

Start the spark-shell. Don’t forget to include the oracle driver. I’m using ojdbc6.jar in this example.

Create a DataFrame from the users table in the default database.

Verify the contents of the DataFrame.

Register the DataFrame so we can run SQL queries against it.

Set up the JDBC URL and connection properties of the Oracle database.

Make sure the users table in Oracle is empty using Oracle SQL Developer (Figure 6-9).

Insert the data from Kudu to Oracle.

Check the Oracle table again and verify that the rows were added (Figure 6-10).

Congratulations! You’ve successfully copied rows from Kudu to Oracle.

Inserting from Kudu to HBase

We’re going to insert data HBase via Impala so that we can use SQL. This is not the fastest method to write to HBase. If performance is critical, I suggest you use the saveAsHadoopDataset method or the HBase Java client API to write to HBase. There are various other ways of importing data into HBase. ^ix

Start the spark-shell and create a data frame from the kudu table.

Verify the contents of the table.

Let’s register the table so we can use it in a query.

Using impala-shell, verify the contents of the destination HBase table.

Go back to the spark-shell and set up the Impala connection.

Insert only the selected rows to the destination HBase table.

Go back to impala-shell and confirm that the rows were added to the destination HBase table.

The data was successfully inserted into the HBase table.

Inserting Rows from Kudu to Parquet

Read the table.

Register the table and then create another DataFrame off of the result of a query.

Inspect the data.

Append the DataFrame to the parquet table. You can also overwrite the destination by using the keyword “overwrite” instead of “append.”

You will discover that Spark generates dozens or hundreds of small files when writing to HDFS. This is known as the “small file” problem. ^x This will eventually cause all sorts of performance problems on your cluster. If that happens you might want to use coalesce or repartition to specify how many files to write to HDFS. For example, you might want Spark to write 1 parquet file to HDFS.

Using coalesce and repartition may cause performance issues since you are essentially reducing the degree of parallelism when writing data. Coalesce and repartition also trigger a shuffle that could cause performance issues depending on the amount of data that you’re processing. You need to balance the number of generated files with processing performance. You may still have to perform regular compaction of your parquet tables after a period of time. This is a problem that you will not have with Kudu.

Insert SQL Server and Oracle DataFrames into Kudu

We will join data from SQL and Oracle and insert it into Kudu.

Start the spark-shell. Don’t forget to include the necessary drivers and dependencies.

Set up the Oracle connection.

Create a DataFrame from the Oracle table.

Register the table so we can run SQL against it.

Set up the SQL Server connection.

Create a DataFrame from the SQL Server table.

Register the table so that we can join it to the Oracle table.

Join both tables. We'll insert the results to a Kudu table.

You can also join both DataFrames using this method.

Create the destination Kudu table in Impala.

Go back to the spark-shell and set up the Kudu connection

Insert the data to Kudu.

Confirm that the data was successfully inserted into the Kudu table.

Looks good to me.

Insert Kudu and SQL Server DataFrames into Oracle

Create the destination table in Oracle using Oracle SQL Developer (Figure 6-11).

Start the spark-shell. Don’t forget to include the necessary drivers and dependencies.

Create a DataFrame from the Kudu users table in the default database.

Verify the contents of the DataFrame.

Register the DataFrame so we can run SQL queries against it.

Create a DataFrame from the SQL Server table.

Register the DataFrame as a temp table.

Join both tables. We'll insert the results to an Oracle database.

You can achieve the same result using this method.

Set up the JDBC URL and connection properties of the Oracle database.

Insert the DataFrame to Oracle.

Verify that the rows were successfully added to the Oracle database (Figure 6-12).

Spark Streaming and Kudu

There are use cases where you need to ingest data into Kudu in near real time. This is a requirement for Internet of Things (IoT) use cases, for example. In Listing 6-1, we show a sample Spark Streaming application that ingests sensor data from Flume. We perform basic event stream processing to tag the status of each event as NORMAL, WARNING, or CRITICAL depending on the temperature returned by the sensor. The status is then saved in a Kudu table together with the rest of the data.

Users can query the Kudu table as data is being inserted into it. In Chapter 9, we discuss a real-time data visualization tool called Zoomdata. You can use Zoomdata to visualize data stored in Kudu in real time.

This is my SBT file. Consult Chapter 5 for more information on building applications. Consult Maven’s online documentation if you are using Maven.

After executing sbt package to package the spark application, you can now launch the application using the spark-submit tool that comes with Spark. The parameters are for testing purposes only. Change the parameters based on your data ingestion requirements. Note that I’m using the jar file to include kudu-spark dependency. The parameters localhost and 9999 will be used as the Flume sink destination.

This is a sample flume.conf file using spooldir as a data source. The settings are adequate for testing purposes.

Kudu as a Feature Store for Spark MLlib

Kudu makes it easier for data scientists to prepare and clean data. Kudu can serve as a fast, highly scalable and mutable feature store for machine learning applications. It also smoothly integrates with Spark SQL and DataFrame API.

Let’s start. We’ll need to download the file, copy it to HDFS, and create an external table on top of it. We will then copy the data to a Kudu table.

Note that we used the Impala function uuid() to generate a unique primary key for our Kudu table. The Kudu table should now have some data in it. Let’s use the spark-shell to fit our model using Spark MLlib.

We can now fit the model.

Let’s evaluate the model.

You can now make some predictions on our data.

We used a very small data set (300+ observations) for our example, but imagine if the data set contains billions of rows. If you need to add or update your training data, you simply run the DML statement against the Kudu table. The ability to update a highly scalable storage engine such as Kudu greatly simplifies data preparation and feature engineering.

Strictly speaking, you can bypass Kudu’s 300-column limit by setting an unsafe flag. For example, if you need the ability to create a Kudu table with 1000 columns, you can start the Kudu master with the following flags: -unlock-unsafe-flags -max-num-columns=1000. This has not been thoroughly tested by the Kudu development team and is therefore not recommended for production use.