Relational Data Management and Analytics

Kudu (when used with Impala) exhibits most of the characteristics of a relational database. It stores data in rows and columns and organizes them in databases and tables. Impala provides a highly scalable MPP SQL engine and allows you to interact with Kudu tables using ANSI SQL commands just as you would with a relational database. Relational database use cases can be classified into two main categories, online transactional processing (OLTP) and decision support systems (DSS) or as commonly referred to in modern nomenclature, data warehousing. Kudu was not designed for OLTP, but it can be used for data warehousing and other enterprise data warehouse (EDW) modernization use cases.

Internet of Things (IoT) and Time Series

Kudu is perfect for IoT and time series applications where real-time data ingestion, visualization, and complex event processing of sensor data is critical. Several large companies and government agencies such as Xiaomi, JD.com, ^v and Australia Department of Defense ^vi are successfully using Kudu for IoT use cases. I discuss IoT, real-time data ingestion, and complex event processing using Impala, Kudu, and StreamSets in Chapter 7. I discuss real-time data visualization with Zoomdata in Chapter 9.

Feature Store for Machine Learning Platforms

Data science teams usually create a centralized feature store where they can publish and share highly selected sets of authoritative features with other teams for creating machine learning models. Creating and maintaining feature stores using immutable data formats such as ORC and Parquet is time consuming, cumbersome, and requires too much unnecessary hard work, especially for large data sets. Using Kudu as a fast and highly scalable mutable feature store, data scientists and engineers can easily update and add features using familiar SQL statements. The ability to update feature stores in seconds or minutes is critical in an Agile environment where data scientists are constantly iterating in building, testing, and improving the accuracy of their predictive models. In Chapter 6, we use Kudu as a feature store for building a predictive machine learning model using Spark MLlib.

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.

Primary Key

Every Kudu table needs to have a primary key. Kudu’s primary key is implemented as a clustered index. With a clustered index, the rows are stored physically in the tablet in the same order as the index. Also note that Kudu doesn’t have an auto-increment feature so you will have to include a unique primary key value when inserting rows to a Kudu table. If you don’t have a primary key value, you can use Impala’s built-in uuid() function or another method to generate a unique value.

Data Types

You may notice that Kudu currently does not support the decimal data type. This is a key limitation in Kudu. The float and double data types only store a very close approximation of the value instead of the exact value as defined in the IEEE 754 specification. ^xv Because of this behaviour, float and double are not appropriate for storing financial data. At the time of writing, support for decimal data type is still under development (Apache Kudu 1.5 / CDH 5.13). Decimal support is coming in Kudu 1.7. Check KUDU-721 for more details. There are various workarounds available. You can store financial data as string then use Impala to cast the value to decimal every time you need to read the data. Since Parquet supports decimals, another workaround would be to use Parquet for your fact tables and Kudu for dimension tables.

As shown in Table 2-1, Kudu columns can use different encoding types depending on the type of column. Supported encoding types includes Plain, Bitshuffle, Run Length, Dictionary, and Prefix. By default, Kudu columns are uncompressed. Kudu supports column compression using Snappy, zlib, or LZ4 compression codecs. Consult Kudu’s documentation for more details on Kudu encoding and compression support.

Partitioning

Table partitioning is a common way to enhance performance, availability, and manageability of Kudu tables. Partitioning allows tables to be subdivided into smaller segments, or tablets. Partitioning enables Kudu to take advantage of partition pruning by allowing access to tables at a finer level of granularity. Table partitioning is required for all Kudu tables and is completely transparent to applications. Kudu supports Hash, Range, and Composite Hash-Range and Hash-Hash partitioning. Below are a few examples of partitioning in Kudu.

Kudu Context

You use a Kudu context in order to execute DML statements against a Kudu table. ^xvii For example, if we need to insert data into a Kudu table:

Insert the DataFrame into Kudu table. I assume the table already exists.

Confirm that the data was successfully inserted.

I discuss Spark and Kudu integration in more detail in Chapter 6.

Spark Streaming and Kudu

In our example shown in Listing 2-1, we will use Flafka (Flume and Kafka) and Spark Streaming to read data from a Flume spooldir source, store it in Kafka, and processing and writing the data to Kudu with Spark Streaming.

A new stream processing engine built on Spark SQL was included in Spark 2.0 called Structured Streaming. Starting with Spark 2.2.0, the experimental tag from Structured Streaming has been removed. However, Cloudera still does not support Structured Streaming as of this writing (CDH 5.13). Chapter 7 describes Flafka and Spark Streaming in more detail.

import org.apache.kudu.client.CreateTableOptions;

import org.apache.kudu.spark.kudu._

import org.apache.spark._

import org.apache.spark.rdd.NewHadoopRDD

import org.apache.spark.SparkConf

import org.apache.spark.storage.StorageLevel

import org.apache.spark.streaming.flume._

import org.apache.spark.streaming.Seconds

import org.apache.spark.streaming.StreamingContext

import org.apache.spark.util.IntParam

import org.apache.spark.sql.SQLContext

object FlumeStreaming {

   case class MySensorData(tableid: String, deviceid: String, thedate: String, thetime: String, temp: Short, status: String)

    def readSensorData(str: String): MySensorData = {

      val col = str.split(",")

      val thetableid = col(0)

      val thedeviceid = col(1)

      val thedate = col(2)

      val thetime = col(3)

      val thetemp = col(4)

      val thestatus = col(5)

      MySensorData(col(0), col(1), col(2), col(3), col(4).toShort, col(5))

    }

  def main(args: Array[String]) {

    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("FlumeStreaming")

    val sc = new SparkContext(sparkConf)

    val ssc = new StreamingContext(sc, Seconds(1))

       // the arguments are for host name and port number

    val flumeStream = FlumeUtils.createPollingStream(ssc,args(0),args(1).toInt)

    val sensorDStream = flumeStream.map (x => new String(x.event.getBody.array)).map(readSensorData)

    sensorDStream.foreachRDD {rdd =>

             val sqlContext = new SQLContext(sc)

             import sqlContext.implicits._

             val kuduContext = new KuduContext("kudumaster01:7051")

// convert the RDD into a DataFrame and insert it into the Kudu table

             val DataDF = rdd.toDF

             kuduContext.insertRows(DataDF, "impala::default.sensortable")

             DataDF.registerTempTable("currentDF")

             // Update the table based on the thresholds

             val WarningFilteredDF = sqlContext.sql("select * from currentDF where temp > 50 and temp <= 60")

             WarningFilteredDF.registerTempTable("warningtable")

             val UpdatedWarningDF = sqlContext.sql("select tableid,deviceid,thedate,thetime,temp,'WARNING' as status from warningtable")

             kuduContext.updateRows(UpdatedWarningDF, "impala::default.sensortable")

             val CriticalFilteredDF = sqlContext.sql("select * from currentDF where temp > 61")

             CriticalFilteredDF.registerTempTable("criticaltable")

             val UpdatedCriticalDF = sqlContext.sql("select tableid,deviceid,thedate,thetime,temp,'CRITICAL' as status from criticaltable")

             kuduContext.updateRows(UpdatedCriticalDF, "impala::default.sensortable")

     }

    ssc.start()

    ssc.awaitTermination()

  }

}

Listing 2-1

Spark Streaming and Kudu

Listing 2-2 shows the flume configuration file, with Kafka used as a flume channel.

agent1.sources  = source1

agent1.channels = channel1

agent1.sinks = spark

agent1.sources.source1.type = spooldir

agent1.sources.source1.spoolDir = /tmp/streaming

agent1.sources.source1.channels = channel1

agent1.channels.channel1.type = org.apache.flume.channel.kafka.KafkaChannel

agent1.channels.channel1.brokerList = kafkabroker01:9092, kafkabroker02:9092, kafkabroker03:9092

agent1.channels.channel1.zookeeperConnect = server03:2181

agent1.channels.channel1.topic = mytopic

agent1.sinks.spark.type = org.apache.spark.streaming.flume.sink.SparkSink

agent1.sinks.spark.hostname = 127.0.0.1

agent1.sinks.spark.port =  9999

agent1.sinks.spark.channel = channel1

agent1.sinks.spark.batchSize=5

Listing 2-2

Flume configuration file

After compiling the package, submit the application to the cluster to execute it.

spark-submit

--class FlumeStreaming

--jars kudu-spark_2.10-0.10.0.jar

--master yarn-client

--driver-memory=512m

--executor-memory=512m

--executor-cores 4  

/mydir/spark/flume_streaming_kudu/target/scala-2.10/test-app_2.10-1.0.jar localhost 9999

The Flafka pipeline with Spark Streaming and Kudu should look like Figure 2-5.

Kudu Java Client API

Listing 2-3 provides an example using the Java client API.

Kudu Python Client API

The Python client API provides an easy way to interact with Kudu. The Python API is still in experimental stage and might change at any time. See Listing 2-4 for an example.

Kudu C++ Client API

Kudu also provides a C++ client API. See Listing 2-5 for an example.

More examples ^xviii can be found on Kudu’s official website ^xix and github repository. ^xx The sample code available online was contributed by the Kudu development team and served as reference for this chapter.

Backup via CTAS

The simplest way to back up a Kudu table is to use CREATE TABLE AS (CTAS) . You’re basically just creating another copy of the Kudu table in HDFS, preferably in Parquet format (or other compressed format), so you can copy the file to a remote location such as another cluster or S3.

You can create the table first so you can customize table options if needed, and then use INSERT INTO to insert data from the Kudu table.

Check the files with the HDFS command.

Export Results via impala-shell to Local Directory, NFS, or SAN Volume

Impala-shell can generate delimited files that you can then compress and copy to a remote server or NFS/SAN volume. Note that this method is not appropriate for large tables.

Export Results Using the Kudu Client API

The Kudu client API can be used to export data as well. See Listing 2-6 for an example.

Compile the java code and run it from the command line. Redirect the results to a file. This method is appropriate for small data sets.

Export Results with Spark

You can also back up data using Spark. This is more appropriate for large tables since you can control parallelism, the number of executors, executor cores, and executor memory.

Start by creating a Data Frame.

Save the data in CSV format.

Or you can save it as Parquet.

Using coalesce to limit the amount of files generated when writing to HDFS may cause performance issues. I discuss coalesce in more details in Chapter 5.

Replication with Spark and Kudu Data Source API

We can use Spark to copy data from one Kudu cluster to another.

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 another Kudu table in another cluster.

Let’s go ahead and insert the data to a table in another Kudu cluster. The destination table needs to be present in the other Kudu cluster.

Verify the data in the destination table.

The rows were successfully replicated.

Real-Time Replication with StreamSets

StreamSets is a powerful real-time and batch ingestion tool used mostly in real-time streaming and Internet of Things (IoT) use cases. You can use StreamSets to replicate data from JDBC sources such as Oracle, MySQL, SQL Server, or Kudu to another destination in real time or near real time. StreamSets offers two origins to facilitate replication through a JDBC connection: JDBC Query Consumer and JDBC Multitable Consumer. The JDBC Query Consumer origin uses a user-defined SQL query to read data from tables. See Figure 2-6.

The JDBC Multitable Consumer origin reads multiple tables in the same database. The JDBC Multitable Consumer origin is appropriate for database replication. StreamSets includes a Kudu destination; alternatively the JDBC Producer is a (slower) option and can be used to replicate data to other relational databases. Chapter 7 covers StreamSets in more detail.

Replicating Data Using ETL Tools Such as Talend, Pentaho, and CDAP

Talend (Figure 2-9) and Cask Data Platform (Figure 2-7) offer native support for Kudu. Both provide Kudu source and sinks and can be used to replicate data from one Kudu cluster to one or more Kudu clusters; another destination such as S3; or an RDBMS such as SQL Server, MySQL, or Oracle. Other tools such as Pentaho PDI (Figure 2-8) does not have native Kudu support. However, it can transfer data to Kudu via Impala, albeit slower. Chapter 7 covers batch and real-time ingestion tools in including StreamSets, Talend, Pentaho, and CDAP in detail.

Impyla

Cloudera built a Python package known as Impyla. ^xxiv Impyla simply communicates with Impala using standard ODBC/JDBC. One of the nice features of Impyla is its ability to easily convert query results into pandas DataFrame (not to be confused with Spark DataFrames). Here’s an example.

pyodbc

pyodbc is a popular open source Python package that you can use to access databases via ODBC/JDBC. ^xxv Here’s an example on how to use pyodbc. To learn more about pyodbc, visit its github page at github.com/mkleehammer/pyodbc.

SQLAlchemy

SQLAlchemy is an SQL toolkit for Python that features an object-relational mapper (ORM). To learn more about SQLAlchemy, visit its website at sqlalchemy.org. Here’s an example on how to use SQLAlchemy to connect to Impala.

Active-Active Dual Ingest with Kafka and Spark Streaming

In this option, all data is published to a Kafka cluster. Spark streaming is used to read data from the Kafka topics. Using Spark Streaming, you have the option to perform data transformation and cleansing before writing the data to Kudu. Figure 2-10 shows two Kudu destinations, but you could have more depending on your HA requirements.

Active-Active Kafka Replication with MirrorMaker

Another high availability option is to replicate Kafka using MirrorMaker . As shown in Figure 2-11, data ingested into the source Kafka cluster is replicated using MirrorMaker to the destination Kafka cluster. From there, the data is read by Spark Streaming and written to the Kudu destinations similar to Figure 2-10. If your goal is dual ingest, using MirrorMaker to replicate your Kudu cluster might be overkill. However, it provides better data protection since data is replicated on two or more Kafka and Kudu clusters.

Active-Active Dual Ingest with Kafka and StreamSets

This option is very similar to Figure 2-10 but uses StreamSets instead of Spark Streaming. StreamSets is easier to use and administer than Spark Streaming and provides built-in monitoring, alerting, and exception handling. It also offers an event framework that makes it easy to kick off a task based on events. I usually recommend StreamSets over Spark Streaming in most projects (Figure 2-12).

Active-Active Dual Ingest with StreamSets

Technically, you don’t need Kafka in order to have active-active dual ingest. StreamSets allows you to configure multiple destinations. Using the Stream Selector processor, it can even route data to different destinations based on certain conditions (Figure 2-13). Using Kafka provides added high availability and scalability and is still recommended in most cases. I cover StreamSets in Chapter 7 in more detail.

Cloudera Manager Kudu Service

Cloudera Manager is Cloudera’s cluster management tool, providing a single pane of glass for administering and managing Cloudera Enterprise clusters. With Cloudera Manager, you can perform common administration tasks such as starting and stopping the Kudu service, updating configuration, monitoring performance, and checking logs.

Kudu Master Web UI

Kudu Masters provide a web interface (available on port 8051) that provides information about the cluster. It displays information about tablet servers, heartbeat, hostnames, tables, and schemas. You can also view details on available logs, memory usage, and resource consumption.

Kudu Tablet Server Web UI

Each table server also provides a web interface (available on port 8050) that provides information about the tablet servers cluster. It displays more detailed information about each tablet hosted on the tablet servers, debugging and state information, resource consumption, and available logs.

Kudu Metrics

Kudu provides several metrics that you can use to monitor and troubleshoot your cluster. You can get a list of available Kudu metrics by executing $ kudu-tserver --dump_metrics_json or kudu-master --dump_metrics_json. Once you know the metric you want to check, you can collect the actual value via HTTP by visiting /metrics end-point. These metrics are also collected and aggregated by Cloudera Manager. For example:

Kudu Command-Line Tools

In addition to Cloudera Manager and the web user interfaces accessible provided by the master and tablet servers, Kudu includes command-line tools for common system administration tasks.