Pipelines

To ingest data, Data Collector requires that you design a pipeline. A pipeline consists of multiple stages that you configure to define your data sources (origins), any data transformation or routing needed (processors), and where you want to write your data (destinations).

After designing your pipeline, you can start it, and it will immediately start ingesting data. Data Collector will sit on standby and quietly wait for data to arrive at all times until you stop the pipeline. You can monitor Data Collector by inspecting data as it is being ingested or by viewing real-time metrics about your pipelines.

Origins

In a StreamSets pipeline, data sources are called Origins. They’re configurable components that you can add to your canvas without any coding. StreamSets includes several origins to save you development time and efforts. Some of the available origins include MQTT Subscriber, Directory, File tail, HDFS, S3, MongoDB, Kafka, RabbitMQ, MySQL binary blog, and SQL Server and Oracle CDC client to name a few. Visit StreamSets.com for a complete list of supported origins.

Processors

Processors allows you to perform data transformation on your data. Some of the available processors are Field Hasher, Field Masker, Expression Evaluator, Record Deduplicator, JSON Parser, XML Parser, and JDBC Lookup to name a few. Some processors, such as the Stream Selector, let you easily route data based on conditions. In addition, you can use evaluators that can process data based on custom code. Supported languages and frameworks include JavaScript, Groovy, Jython, and Spark. Visit StreamSets.com for a complete list of supported processors.

Destinations

Destinations are the target destination of your pipeline. Available destinations include Kudu, S3, Azure Data Lake Store, Solr, Kafka, JDBC Producer, Elasticsearch, Cassandra, HBase, MQTT Publisher, and HDFS to name a few. Visit StreamSets.com for a complete list of supported destinations.

Executors

Executors allow you to run tasks such as a MapReduce job, Hive query, Shell script, or Spark application when it receives an event.

Data Collector Console

The console is Data Collector’s main user interface (see Figure 7-1). This is where all the action happens. This is where you design your pipeline, configure your stages, run and troubleshoot pipelines, and more.

StreamSets Data Collector can ingest data in real time or batch (see Figure 7-2). Real-time data sources or origins include MQTT, Kafka, and Kinesis to name a few. Batch data sources includes HDFS, S3, Oracle, SQL Server, MongoDB, and so on. Data eventually lands on one or more destinations. In between the data sources and destinations are processors that transform and process data in-stream.

There are several pre-built processors for some of the most common data transformations and enrichments. Examples of processors include column splitters and hashers, data type converters, XML parsers, JDBC lookups, and Stream processors to name a few. These processors can be configured without writing a single line of code. However, there are times when you want to write code, mainly, because none of the existing processors can handle the specific type of transformation that you require. StreamSets provide evaluators that support JavaScript, Java, Java EL, Spark, Jython, and Groovy. StreamSets connects all these stages (origins, processors, data sources) to form a pipeline. StreamSets executes the pipeline in-memory, providing maximum performance and scalability.

Batch-Oriented Data Ingestion

As mentioned earlier, StreamSets also supports batch workloads. As an origin and destination, I suggest Kudu for most batch-oriented jobs. Kudu provides most of the benefits of HDFS without the administration overheads. Performance benchmarks performed by the Kudu development team suggests that Kudu is still slightly slower than HDFS (Parquet) in some operations. If the difference in performance matters to your application (or if you’re processing unstructured data), then I suggest you use HDFS. For most applications, the difference in performance may not be important. In that case, I would suggest you stick with Kudu (see Figure 7-3). With StreamSets, batch pipelines are built in the same way as real-time pipelines. The only difference is when you use batch-oriented origins such as S3, HDFS, RDBMS, or file directory, StreamSets detects the type of origin and automatically reads the data in batch mode.

Internet of Things (IoT)

StreamSets is the Swiss army knife of data ingestion. Adding to its numerous features and capabilities is its support for IoT or Internet of Things ^viii (see Figure 7-4). StreamSets includes an MQTT Subscriber origin and MQTT Publisher destination that allows it to be used as an Internet of Things gateway. For reading data from SCADA networks and OPC historian, StreamSets includes an OPC UA Client origin. Finally, StreamSets supports CoAP (Constrained Application Protocol). ^ix CoAP is a protocol for low-power and low-bandwidth environments designed for machine-to-machine device communication. In Chapter 9, we implement a complete IoT data ingestion and visualization application using StreamSets, Kudu, and Zoomdata.

Deployment Options

StreamSets supports different deployment options, stand-alone or in-cluster modes. In stand-alone mode, StreamSets Data Collector is installed on one or more edge nodes. These edge nodes can be located in the same data center or in geographically disparate sites, as long as the network can support the latency and throughput requirements of your pipelines. In cluster mode, StreamSets pipelines are deployed and executed as Spark Streaming applications, utilizing YARN or Mesos as its cluster manager to take full advantage of the Hadoop cluster’s scalability (see Figure 7-5).

Consult the Data Collector User Guide for information on installing StreamSets from a Tar ball (Service Start), an RPM Package (Service Start), and Cloudera Manager (Cluster Mode).

Using StreamSets Data Collector

Let’s start by creating our first pipeline. Navigate to the StreamSets Data Collector URL, and you will be presented with a login page. The default username is “admin.” The default password is “admin.” Log in, then click “Create New Pipeline.” When prompted, enter a title and description for your new pipeline (see Figure 7-6).

You can now design your first pipeline.

Ingesting XML to Kudu

In this example we will ingest sensor data in XML format into a Kudu table. Log in to StreamSets. On the Pipelines page, click the “Create New Pipeline” button (see Figure 7-7).

In the Data Collector Console, Click the “Stage Library” icon located near the “Help” icon (see Figure 7-8). You will be presented with a list of available stages you can use. Choose “Origins” in the list of stages. We will use Directory origins in this example. With the Directory origins, files that are copied to a designated directory will be ingested by StreamSets. Drag the Directory origins to the right side of the canvas.

Next, we need to add a processor. Choose “Processors” in the stage library. We will use an XML Parser to parse our XML data and convert it into a format that can be inserted into Kudu. Scroll down until you see the XML Parser icon. Drag the XML Parser processor to the canvas, near the Directory origins. Connect the Directory origins to the XML Parser processor in the canvas as shown in Figure 7-9. You will notice a yellow Pipeline Creation Help Bar. You can use the help bar to assist you in choosing stages. For now, we’ll ignore the help bar.

We need a Destination to complete our pipeline. Choose “Destinations” in the stage library. We will use a Kudu as our destination. Drag the Kudu destination to the canvas, near the XML Parser processor. Connect the XML Parser to the Kudu destination in the canvas as shown in Figure 7-10.

Before we can start configuring our stages, we need to create the directory that will serve as the origin of our data source and the destination Kudu table. We’ll also prepare some test data.

Open impala-shell and create a table.

Log in to the Cloudera cluster and create a directory that will contain our sensor data.

Let’s create some test data that we will use later in the exercise.

Configure Pipeline

You can set configuration options for the entire pipeline by clicking on the name of the pipeline; in our case click on the “Sensor Data Kudu Pipeline” link near the upper-left side of the panel (see Figure 7-11).

In the “Error Records” tab, change “Error Records” to “Discard (Library: Basic)” (see Figure 7-12). For now, leave the rest of the default parameters.

Configure the Directory Origin

Click the Directory origin stage in the pipeline canvas. In the properties panel, navigate to the “Configurations” tab. Within the “Configurations” panel, you’ll see another set of tabs (see Figure 7-13). We’ll go through each tab.

The “General” allows users to specify a name and description for the origin. There’s an option to enable the origin to generate events. If you enable “Produce Events,” the directory origin produces event records every time the directory origin begins or ends reading a file. You can ignore this configuration item for now. “On Record Error” allows you to choose an action to take on records sent to error. The default option is “Send to Error.” The other options are “Discard” and “Stop Pipeline.” Leave the default option for now.

The “Files” tab contains various configuration options related to file ingestion. The most interesting option for us in our examples are “Files Directory,” which is the local directory where you want to place files you want to ingest. “File Name Pattern Mode” lets you select whether the file name pattern specified uses regex syntax or glob syntax. Leave it set to “Glob” for now. “File Name Pattern” is an expression that defines the pattern of the file names in the directory. Since we’re ingesting XML files, let’s set the pattern to “*.xml”. “Read Order” controls the order of how files are read and processed by Data Collector. There are two options, “Last Modified Timestamp” and “Lexicographically Ascending File Names.” When lexicographically ascending file names are used, files are read in lexicographically ascending order based on file names. ^x It’s important to note that this ordering will order numbers 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15 as 1, 10, 11, 12, 13, 14, 15, 2, 3, 4, 5, 6, 7, 8, 9.If you must read files in lexicographically ascending order, consider adding leading zeros to the file names such as: file_001.xml, file_002.xml, file_003.xml, file_004.xml, file_010.xml, file_011.xml, file_012.xml.

When the last modified timestamp is used, files are read using the timestamp. Files with the same timestamp are ordered based on file names. For this exercise, let’s set the read order to “Last Modified Timestamp” (see Figure 7-14).

The “Post Processing” tab provides configuration options for files after they’ve been read and processed. “Error Directory” can be configured as the destination for files that can’t be read. “File Post Processing” provides options to take after a file has been processed. You can archive or delete a file after processing, or you can do nothing. If you choose “Archive,” additional options will be presented, such as the “Archive Directory” and “Archive Retention Time (mins)” (see Figure 7-15). Let’s set “File Post Processing” to nothing for this example.

The “Data Format” tab allows you to set configuration parameters related to the format of your data (see Figure 7-16). The most interesting parameters for us are “Data Format.” Note that even though we’re ingesting XML data, we’ll set this parameter to “Text.” The XML data that we’re ingesting is not strictly valid XML since it does not include a root element or XML prolog. So we will have to process our XML data using custom delimiters. ^xi Let’s check “Use Customer Delimiters” and specify a customer delimiter; in our case let’s set it to “</sensordata>.” Finally, let’s check “Include Customer Delimeters.” We’ll leave the other parameters to its default values for now.

Configure the XML Parser Processor

Click the XML parser processor in the canvas and navigate to the “Configuration” section of the Properties Panel. The “General” tab will let you set a name and description for the XML Parser processor. Navigate to the “Parse” tab and set “Field to Parse” and “Target Field” to “/text” (see Figure 7-17). As mentioned earlier, the directory origins were configured to process text data. It writes records into a single text field named “text.” ^xii This is a convention used by SDC.

When the Directory origins processes our XML data, it will create two records, delimited by “</sensordata>.”

In the “Configuration” tab, navigate to Kudu and fill in the required parameters. The hostnames and port of the “Kudu Masters.” Note that because we created the “sensordata” table in Impala, it’s considered an internally managed table. Kudu tables created in Impala follow the naming convention “impala::database.table_name.” Set the “Table Name” to “impala::default.sensordata”. You also need to set the field to column mapping as seen in Figure 7-18. The format of the SDC field looks like “/text/column[0]/value.” Finally, you set the “Default Operation” to “INSERT.” Note, that you have the option to perform other operations such as UPSERT, UPDATE, and DELETE.

Validate and Preview Pipeline

Now that you’re done designing your pipeline, you need to check and make sure all the configurations are valid. Click the validate icon on the upper-right corner of the canvas to validate the pipeline. Once you’ve completed the validation, you can preview your pipeline to test it without actually running it. Previewing your pipeline is an easy way to debug and step through your pipeline, allowing you to inspect your data at each stage. ^xiii Click the preview icon. A “Preview Configuration” window will appear (see Figure 7-19). Accept the default values and click the “Run Preview” button.

You’re now in preview mode. Click the Directory origin. In the preview stage pane, you can see a list of test records (see Figure 7-20). Collapsing each record shows the actual data. Note that the third record is empty. This is going to cause errors in the XML Parser processor.

Click the XML Parser processor. Here you can see how the input data was processed and what the resulting output data looks like. As you can see, the XML data was processed correctly but as expected, there was an error processing the empty record (see Figure 7-21).

As expected, the Kudu destination’s input data is the XML Parser’s output (see Figure 7-22).

You can also preview multiple stages. Click “multiple” on top of the preview stage panel. The “Preview Multiple Stages” panel allows you to inspect the flow of the records from one stage to another (see Figure 7-23).

Start the Pipeline

Now you’re ready to start the pipeline. Click the start icon, and then copy our test data (file001.xml) to /sparkdata.

After a couple of seconds, you should see some runtime statistics on the monitoring panel such as record count, record throughput, and batch throughput to mention a few (see Figure 7-24).

Click on each stage in the canvas to see runtime statistics specific for that stage. Let’s click on the Directory origin located in the canvas (see Figure 7-25).

Clicking on the XML Parser processor gives you a different set of metrics (see Figure 7-26).

The Kudu destination has a different set of metrics as well. According to the screenshot in Figure 7-27, two rows were inserted to the sensordata Kudu table.

Let’s verify if the rows were indeed successfully inserted in Kudu. Start impala-shell and query the sensordata table.

The rows were successfully added to the table. Try adding more data (see Figure 7-28). You’ll see the charts and other metrics get updated after adding more data.

Congratulations! You’ve successfully built and run your first StreamSets data pipeline.

Stream Selector

There are times when you have a requirement to route data to certain streams based on conditions. For example, you can insert data to certain tables based on “Country.” There is a default stream that handles records that do not match any of the user-defined conditions. ^xiv

In this pipeline, all records where “Country” matches USA goes to Stream 1, Canada goes to Stream 2, and Brazil goes to Stream 3. Everything else goes to Stream 4. You’ll be able to specify these conditions using the Stream Selector.

Let’s design another pipeline. Instead of starting from scratch, let’s make a copy of the pipeline we just created and modify it to use a Stream Selector. In this example, we’ll write to three different versions of the XML data to three different Kudu tables. Here’s what our new XML data looks like. The Stream Selector will examine the value of firmwareversion.

We’ll create two more Kudu tables: sensordata2 and sensordata3. We’ll also add a firmwareversion column in the existing sensordata table. We’ll have a total of three sensordata tables, each with a different set of columns.

Drag a Stream Selector processor and two more Kudu destinations onto the canvas. Connect the XML Parser to the Stream Connector. You need to define three conditions: XML data with firmwareversion of 1 (or any value other than 2 or 3) goes to default stream (Kudu 1), XML data with firmwareversion 2 goes to stream 2 (Kudu 2), and finally XML data with firmwareversion 3 goes to stream 3 (Kudu 3).

Add two more conditions to the Stream Selector processor with these conditions.

Your new pipeline should look like Figure 7-29.

For each Kudu destination, make sure you map the correct SDC field to the corresponding column names (Figure 7-30).

Note that the second and third Kudu destinations will have additional columns (Figure 7-31). Don’t forget to add the firmwareversion to the first Kudu destination (Figure 7-32).

Validate and preview the pipeline you just created (Figure 7-33). Confirm that each Kudu destination is receiving the correct XML data (Figures 7-34 and 7-35).

Before you start the pipeline, confirm that the destination Kudu tables are empty from the impala-shell.

Start the pipeline, then copy file002.xml to /sparkdata. After a few seconds, you’ll see some updates on the charts in the monitoring panel. Click the Stream Selector processor. As you can see in the Record Count chart, Input has three records, while the three Output has one record each.

Check each Kudu destination to see if the records were successfully ingested.

Check the Kudu tables to confirm that the records were successfully ingested.

Expression Evaluator

The Expression Evaluator allows users to perform data transformations and calculations and writes the results to new or existing fields. Users can also add or change field attributes and record header attributes. ^xv I often use the Expression Evaluator as a rules engine.

Let’s work on a new example. We’ll use an Expression Evaluator to perform two data transformations. We’ll convert the temperature from Celsius to Fahrenheit and save it to the existing temperature field. For the second transformation, we’ll update a new status field with the values “NORMAL,” “WARNING,” or “CRITICAL” depending on the values of the pressure field. If pressure is over 40, the status field gets updated with “WARNING,” and if pressure is over 50 it gets updated with “CRITICAL.” Let’s get started.

Add another “status” column to the Kudu sensordata table.

We’ll use the following XML data for this example.

Make a copy of the original pipeline and add an Expression Evaluator processor in between the XML Parser processor and Kudu destination. Connect the stages. Your canvas should look like Figure 7-40.

Click the Expression Evaluator processor (Figure 7-38) and navigate to the “Expressions” tab. We’ll add two entries in the “Field Expressions” section. For the first entry, in the “Output Field” enter the following value:

In the “Field Expression”, enter the expression:

This is the formula to convert temperature from Celsius to Fahrenheit. Specifying an existing field in the “Output Field” will overwrite the value in the field.

For the second entry, we’ll use an if-then-else expression. Enter a new field in the “Output Field”:

Enter the expression in “Field Expression”:

Your screen should look like Figure 7-41.

Map “/text/status” to the new status field in the Kudu destination.

Validate and preview the pipeline. Click the Expression Evaluator stage and confirm the values of the input and output records. Note that the value of temperature in the output Record 4 and Record 5 has been converted to Fahrenheit. The value of status on both output Record 4 and Record 5 were updated with “WARNING” and “CRITICAL” respectively, based on the value of the pressure field.

Confirm the values in the Kudu destination.

Run the pipeline and copy the XML file to /sparkdata.

It looks like all five records were successfully inserted to the Kudu sensordata table.

Confirm that the data were successfully inserted to Kudu. As you can see the status field was successfully updated based on pressure. Also, the temperatures are now in Fahrenheit.

Using the JavaScript Evaluator

StreamSets Data Collector includes a JavaScript evaluator that you can use for more complex processing and transformation. It uses JavaScript code to process one record or one batch at a time. ^xvi As you can imagine, processing one record at a time will most likely be slower compared to processing by batch of data at a time. Processing by batch is recommended in production environments.

Although we’re using JavaScript in this example, other languages are supported by Data Collector. In addition to JavaScript, you can use Java, Jython (Python), Groovy, and Spark Evaluators. These evaluators enable users to perform more advanced complex stream processing, taking advantage of the full capabilities of each programming languages.

Let’s start with a simple example. We’ll create two additional fields, action and uniqueid. We’ll generate the value of action based on the value of the status field. Next, we’ll generate a universally unique identifier (UUID) using a custom JavaScript function. We’ll save this UUID in our uniqueid field.

We don’t have to start from scratch. We’ll make a copy of the previous pipeline and then drag a JavaScript Evaluator processor into the canvas and place in between the Expression Evaluator processor and the Kudu destination. Reconnect the stages. Your canvas should look like Figure 7-46.

Now let’s configure the JavaScript Evaluator to use our custom JavaScript code in our pipeline. In the Properties panel, on the JavaScript tab, make sure that “Record Processing Mode” is set to “Batch by Batch.” “Script” will contain your customer JavaScript code. As mentioned earlier, we’ll create a function to generate a UUID. ^xvii The JavaScript Evaluator passes the batch to the script as an array. To access individual fields, use the format: records[arrayindex].value.text.columname. To refer to the value of the status field use “records[i].value.text.status.” We also created a new field, “record[i].value.text.action” using this format. See Listing 7-1.

Your screen should look like Figure 7-47.

Add the action and uniqueid columns to the Kudu sensordata columns.

Map the SDC field with their corresponding Kudu table columns.

Validate and preview your pipeline. Click the JavaScript Evaluator and confirm the values of the input and output records. Note that the action and unique fields have been populated.

Confirm the values in the Kudu destination.

Run the pipeline and copy the XML file to /sparkdata.

It looks like all five records were successfully processed by the JavaScript Evaluator.

All five records were also successfully inserted to the Kudu sensordata table.

Confirm that the data was successfully inserted to Kudu. As you can see, the action and uniqueid fields were successfully populated.

Congratulations! You’ve successfully used a JavaScript evaluator in your pipeline.

Ingesting into Multiple Kudu Clusters

Sometimes you’ll be required to ingest data into two or more active-active clusters for high availability and scalability reasons. With Data Collector this is easily accomplished by simply adding Kudu destinations to the canvas and connecting them to the processor. In this example, we will simultaneously ingest XML data into two Kudu clusters, kuducluster01 and kuducluster02.

Drag a second Kudu destination to the canvas. You should now have two Kudu destinations. Make sure they’re both connected to the XML processor.

Click the first Kudu destination. Make sure the correct host name and ports are specified in the Kudu Masters field. Make sure the SDC fields also map to the Kudu table fields.

Next, click the second Kudu destination. Perform the same task as you did on the first Kudu destination.

Validate and preview the pipeline. When you’re done, start the pipeline and start adding data. Monitor the XML Parser. Note the number of input and output data.

Monitor the first Kudu destination. Note that it processed 1,582, 554 records.

Check the second Kudu destination and note that it also ingested the same amount of records.

Verify in Impala. Perform a SELECT COUNT on the sensordata table on the first Kudu cluster.

Do the same SELECT COUNT on the second Kudu cluster.

You ingested data into two Kudu clusters simultaneously. You can ingest to different combinations of platforms such as HBase, Cassandra, Solr, Kafka, S3, MongoDB, and so on. Consult the Data Collector User Guide for more details.

REST API

Data Collector has an easy-to-use web-based GUI for designing, running, and administering pipelines. However, you can also use the built-in REST API if you’d like to programmatically interact with Data Collector, for automation purposes, for example. ^xviii The REST API lets you access all aspects of Data Collector, from starting and stopping pipelines, returning configuration information and monitoring pipeline metrics.

You can access the REST API by clicking the Help icon and then selecting “RESTful API.”

You’ll be presented with a group of available operations. Expand the manager list by clicking on “manager.” You can explore different operations under the “manager” group such as returning pipeline status, starting and stopping pipelines, and so on.

Expand “Return All Pipeline Status.” Choose the response content type “application/json” and click the “try it out” button.

The “response body” will contain details about your pipelines in JSON format similar to Listing 7-2. Depending on the number of pipelines and type of error messages returned, the response body could be quite large.

You can make REST API requests by using curl, a utility that can download data using standard protocols. The username and password and Custom HTTP header attribute (X-Requested-By) are required.

Event Framework

StreamSets has an Event Framework that allows users to kick off tasks in response to triggers or events that happen in the pipeline. StreamSets uses dataflow triggers to execute tasks such as send e-mails, execute a shell script, starting a JDBC query, or starting a Spark job after an event, for example, after the pipeline successfully completes a JDBC query.

Dataflow Performance Manager

One of the most powerful feature of StreamSets is StreamSets Dataflow Performance Manager. StreamSets Dataflow Performance Manager (DPM) is a management console that lets you manage complex data flows, providing a unified view of all running pipelines in your environment. DPM is extremely helpful in monitoring and troubleshooting pipelines. You’ll appreciate its value more when you’re tasked to monitor hundreds or even thousands of pipelines. Covering DPM is beyond the scope of this book. For more information, consult the Data Collector User Guide.

I’ve only covered a small subset of StreamSets features and capabilities. For more information about StreamSets, consult the StreamSets Data Collector User Guide online. In Chapter 9, I use StreamSets, Zoomdata , and Kudu to create a complete Internet of Things (IoT) application that ingests and visualizes sensor data in real time.

Data Ingestion with Kudu

The first thing we need to do is prepare test data for our example.

To access CDAP, direct your browser to the hostname of the server where the application is installed, using port 11011. The first time you access CDAP, you’re greeted with a welcome page.

Close the window and click the button “Start by Adding Entities to CDAP.” You’re presented with five choices. Let’s create a Pipeline.

A canvas will appear. This is where you’ll design, deploy, and administer your data pipelines. On the right side of the canvas, you can choose different sources, transformations, and sinks that you can use to design your pipeline.

You may notice that the Kudu source and sink are missing. We need to add the Kudu source and sink from the Cask Market. Click the “Cask Market” link near the upper-right corner of the application window.

Scroll down the list of items in the Cask Market until you find the Kudu source and sink.

Click the Kudu source and sink icon and then click “Deploy.”

Click “Finish.”

Click “Create a Pipeline.” Notice the Kudu icon is now available as a data source and sink.

Our example CDAP pipeline will ingest a CSV file into a Kudu table. But before inserting the data, we will hash the name with a popular hashing algorithm. Let’s drag a “File” source to the canvas.

Double-click the file source icon. Enter the file properties such as the filename and the path of the file in HDFS.

CDAP makes it easy to access documentation about a particular sink, source, or transformation.

Next, drag a CSVParser transformation to the canvas and connect it to the File source.

Enter the CSVParser properties. Make sure the Output Schema has the correct columns.

The documentation for the CSV Parser transformation is readily available if you need assistance.

Drag a Hasher transformation to the canvas and connect it with the CSVParser.

Configure the Hasher by choosing the hashing algorithm and specifying the field you want to hash. For this example, we’ll choose MD5 as our hashing algorithm.

Drag a Kudu sink and connect it to the Hasher.

Configure the Kudu sink. Note that CDAP uses the Kudu native APIs to insert data to the tables instead of going through Impala. CDAP will also create the destination table so you just need to specify the table name, instead of using the “impala::database_name.table_name” format. In this example, we’ll use the table name “users_table.” Later, we’ll create an external Impala table on top of this Kudu table.

The Kudu sink documentation is available just in case you need help with some of the options.

Your canvas should look like Figure 7-77. You can now deploy the pipeline. You can preview and validate the pipeline first to make sure there are no errors.

Finally, the Canvas will show you the number of records transferred and processed from the File source all the way to the Kudu sink. You’ll see an indicator near the upper-left corner of the canvas whether the job as successful or not. See Figure 7-78.

Check the logs and confirm that the job succeeded as shown in Figure 7-79.

Confirm that the rows were successfully inserted to the Kudu table. First, we have to create an external table in Impala on top of the Kudu table that CDAP created. Make sure the name field is hashed.

You’ve successfully ingested data into a Kudu table, using a Field Hasher to perform in-stream transformation to your data.

Pentaho Data Integration

Pentaho offers a complete line of products for data integration, big data processing, and business analytics. We’ll focus on Pentaho Data Integration (PDI) in this chapter. There is a community version called Pentaho Community Edition (CE), which includes Kettle, the open source version of Pentaho Data Integration. There is an Enterprise version, which includes Native YARN integration, analyzer and dashboard enhancements, advanced security, and high availability features. ^xix

Pentaho PDI has an intuitive user interface with ready-built and easy-to-use components to help you develop data ingestion pipelines. Similar to most ETL tools in the market, Pentaho PDI allows you to connect to different types of data sources ranging from popular RDBMS such as Oracle, SQL Server, MySQL, and Teradata to Big Data and NoSQL platforms such as HDFS, HBase, Cassandra, and MongoDB. PDI includes an integrated enterprise orchestration and scheduling capabilities for coordinating and managing workflows. You can effortlessly switch between Pentaho’s native execution engine and Apache Spark to scale your pipelines to handle large data volumes.

Ingest CSV into HDFS and Kudu

Unlike other data integration tools described in this chapter, Pentaho does not include native support for Kudu yet. In order for us to insert data into Kudu, we will need to use Pentaho’s generic Table Output component using Impala’s JDBC driver. Using Table Output directly may not be fast enough depending on the size of your data and data ingestion requirements. One way to improve performance is to stage data to HDFS first, then using Table Output to ingest data from HDFS to Kudu. In some cases, this may be faster than directly ingesting into Kudu.

The first thing we need to do is prepare test data for our example.

Start Pentaho PDI. We’ll use the Community Edition (Kettle) for this example. Navigate to where you installed the binaries and execute ./spoon.sh. See Figure 7-80.

The Spoon graphical user interface is shown in Figure 7-81. This is where you design and build your jobs and transformation.

On the left side of the window, you can find a list of all the supported input, output and transformation steps available for ETL development. Expand “Input” and drag “CSV file input” step to the canvas as shown in Figure 7-82.

Double-click the icon. Enter the configuration details such as the file name, delimiters, and so on. See Figure 7-83.

Click “Get Fields” to let Pentaho infer the field type and size of the fields. See Figure 7-84.

Expand “Big Data” and drag “Hadoop File Output” to the canvas as shown in Figure 7-85.

Connect “CSV file input” to “Hadoop File Output,” as shown in Figure 7-86.

Double-click the “Hadoop File Output” component to configure the destination. Enter all required information such as the name of the cluster, the address of the HDFS namenode, and so on. See Figure 7-87.

Enter additional parameters to the Hadoop File Output stage. In our example, we’ll rename the file based on our own Date time format. See Figure 7-88.

Specify details about the fields such as data type, format, length, precision and so on. See Figure 7-89.

You can preview and sanity check your data. See Figure 7-90.

One of the most helpful features of PDI is the ability to monitor execution metrics of your job. Try doing a test run to make sure that the data is getting transferred from your source to destination. See Figure 7-91.

Navigate to “metrics” in the Execution Results panel as shown in Figure 7-92.

Looking closer, it shows you how long each step in the job takes. In this example, the entire job took 412 ms to execute. Initializing the transformation took 122 ms, part of the 122 ms, initializing the CSV file input step took 3 ms while initializing the Hadoop File Output step took 96 ms. The actual execution of the CSV file input step took 15 ms, while the execution of the Hadoop File Output took 251 ms. See Figure 7-93.

Navigate to the “Logging” tab to inspect the application log. As you can see the job successfully executed as shown in Figure 7-94.

Confirm that the file was indeed copied to HDFS.

Let’s configure the final destination. Drag a Table Output step to the canvas. Double-click the icon and start configuring the step. Let’s configure the Impala driver. Download the Impala JDBC driver from Cloudera.com and copy it to <install directory>/data-integration/lib. See Figure 7-95.

Configure the JDBC driver. Enter the hostname or IP address of the Impala server, the TCP/UP port and the database name as shown in Figure 7-96.

Test the database connection as shown in Figure 7-97.

Create the destination table.

Enter the destination table and other relevant configuration options. See Figure 7-98.

Click “Get fields” and make sure the source and destination fields match as shown in Figure 7-99.

Preview and sanity check the data as shown in Figure 7-100.

Execute the job. Monitor the logs and make sure the job successfully executes. See Figure 7-101.

The job ran successfully.

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

Data Ingestion to Kudu with Transformation

Let’s start with another example. This time we’ll use a string replace transformation to replace the string “Yonkers” with “Berkeley.”

Prepare the data.

Enter the source directory and regular expression (Wildcard) to search for files. See Figure 7-102.

Preview and sanity check the data. See Figure 7-103.

Drag a “Replace string” transformation step to the canvas. Double-click the icon and configure it by specifying the string to search and the string to replace it with. As we mentioned earlier, we’ll replace the string “Yonkers” with “Berkeley” in the city field. You can ignore the other options for now. See Figure 7-104.

Run the job as shown in Figure 7-105.

Check the table and make sure “Yonkers” is replaced with “Berkeley.” Compare the value of the city in row id 5 with the original source text file.

SQL Server to Kudu

In this example, we’ll show how to ingest data from an RDBMS (SQL Server 2016) to Kudu.

Make sure you have data in your source SQL Server 2016 database. We’ll use the same table we used in other examples. Inside a salesdb database, there’s a table called users with the following rows. Use SQL Server Management Studio to confirm. See Figure 7-106.

In Spoon, drag a Table Input step to the canvas as shown in Figure 7-107.

Configure the Table Input step. See Figure 7-108.

Make sure to install the SQL Server JDBC driver by copying sqljdbc41.jar (JDK 1.7) or sqljdbc42.jar (JDK 1.8) to <install directory>/data-integration/lib. Download the JDBC driver from Microsoft.com if you haven’t already. Choose the MS SQL Server (native) connection type as shown in Figure 7-109.

Configure the JDBC setting (Figure 7-110).

Test the connection to the source SQL Server 2016 database (Figure 7-111).

Now that you’ve configured the connection to your source database, specify a SQL query as your data source. See Figure 7-112.

Preview your data to ensure your SQL query is valid (Figure 7-113).

Drag a Table Output step to your canvas and connect it to the Table Input step (Figure 7-114).

Configure your destination. Enter your target schema and target table. If you haven’t done it already, configure your Impala JDBC connection as well as in Figure 7-115.

Run the job (Figure 7-116).

Inspect the logs for errors.

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

Congratulations! You’ve successfully ingested data from SQL Server 2016 to Kudu.

Talend Open Studio

Talend is one of the leading software companies that specializes in big data integration. Talend offers free open source data ingestion tools called Open Studio for Big Data and Open Studio for Data Integration. Both tools provide modern graphical user interface, YARN support, HDFS, HBase, Hive and Kudu support, connectors to Oracle, SQL Server and Teradata, and fully open source under Apache License v2. ^xx The main difference between Open Studio for Data Integration and Open Studio for Big Data is that Data Integration can only generate native Java code, while Big Data can generate both native Java, Spark and MapReduce code.

The commercial version gives you access to live Talend support, guaranteed response times, upgrades, and product patches. The open source version only provides community support. ^xxi If you are processing terabytes or petabytes of data, I suggest you use the Big Data edition. For traditional ETL-type workloads such as moving data from an RDBMS to Kudu, with some light data transformation in between, the Data Integration edition is sufficient. We will use Talend Open Studio for Data Integration in this chapter.

Ingesting CSV Files to Kudu

Let’s start with a familiar example of ingesting CSV files to Kudu.

Prepare the test data.

Create the destination Kudu table in Impala.

Start Talend Open Studio for Data Integration as shown in Figure 7-117.

You can select to create a new project or open an existing one. Let’s start a new project (Figure 7-118).

You’re presented with a GUI similar to Figure 7-119.

Let’s create a new job (Figure 7-120).

Specify a name for the job. You can specify other properties such as purpose, description, author, and so on. Click “Finish” when you’re done (Figure 7-121).

You’ll see a canvas similar to Figure 7-122. This is where you’ll design and run jobs.

Similar to previous data ingestion tools discussed in earlier in this chapter, you have to specify a source, transformation steps, and destination when designing data ingestion pipelines. On the right-hand side of the canvas you will see a list of inputs that you can use as your data source. Because we’re ingesting a CSV file, drag and drop a tFileInputDelimited source into the canvas. Configure the source by specifying the file name of the CSV file. See Figure 7-123.

Next, specify an output by dragging and dropping a Kudu output into the canvas. Configure the Kudu output by specifying the table name, connection information, and so on (Figure 7-124).

Don’t forget to connect the input and output icons (Figure 7-125).

Run the job as shown in Figure 7-126.

The logs will show information about the job while it’s running. You’ll see an exit code when the job completes. An exit code of zero indicates that the job successfully completed.

Confirm that the rows were successfully inserted into the Kudu table.

SQL Server to Kudu

For our second example, let’s ingest data from SQL Server to Kudu using Talend Open Studio. Create the Kudu table DimGeography in Impala if you haven’t done it already.

Create a new job (Figure 7-127).

Configure the component by making sure the correct data types are specified and so on (Figure 7-129).

Drag and drop a Kudu output component into the canvas. Connect it to the tMSSQLinput component as shown in Figure 7-130.

Configure the Kudu output and make sure the data types matches with the SQL Server table as shown in Figure 7-131.

Sync the schemas if they do not match (Figure 7-132).

Run the job as shown in Figure 7-133.

Verify if the job ran successfully by comparing source and destination row counts (Figure 7-134).

Data Transformation

Let’s now use some of Talend’s built-in features for data transformation. I’ll use the tReplace component to replace a value in the specified input column. We’ll replace the value “United Kingdom” with “UK.” I’ll also use the tFilterRow component to filter results to only include records where city is equal to “London” or “Berkshire.”

Drag and drop the tReplace and tFilterRow components into the canvas, in between the input and the output as shown in Figure 7-136.

Configure the tReplace component by selecting the field that contains the value that you want to replace (Figure 7-135).

Configure the tReplace component by specifying the value you want to replace and the replacement value. In this case we’ll replace “United Kingdom” with “UK” (Figure 7-136).

Configure the tFilterRow component. We’ll return only the rows that equals “London” and “Berkshire” in the city field (Figure 7-137).

Don’t forget to connect all the components as shown in Figure 7-138.

Run the job as shown in Figure 7-139.

Inspect the data in the Kudu table to ensure that the job successfully executed. As you can see from the results, only records where city is equal to London and Berkshire were returned. United Kingdom was also replaced with UK.

Informatica

Informatica is the largest software development company in the world that specializes in data integration. Founded in 1993, the company is based in Redwood City, California. Informatica also develops software for master data management, data quality, b2b data exchange, data virtualization, and more. ^xxii Informatica PowerCenter Big Data Edition is the company’s flagship product for big data integration. Like the other ETL tools described in this chapter, Informatica PowerCenter Big Data Edition features an easy-to-use visual development environment. Informatica PowerCenter Big Data Edition has tight integration with the leading traditional and big data platforms making it easy for you schedule, manage and monitor processes, and workflows across your enterprise. ^xxiii As of this writing, Informatica PowerCenter doesn’t have native Kudu support; however you may be able to ingest data into Kudu using Informatica PowerCenter via Impala and JDBC/ODBC.

Microsoft SQL Server Integration Services

This list would not be complete if I didn’t mention the top three largest enterprise software development companies, Microsoft, IBM, and Oracle. SQL Server Integration Services (SSIS) includes features that supports Hadoop and HDFS data integration. SSIS provides the Hadoop Connection Manager and the following Control Flow Tasks: Hadoop File System Task, Hadoop Hive Task, and Hadoop Pig Task. SSIS supports the following data source and destination: HDFS File Source and HDFS File Destination. As with all the ETL tools described in this chapter, SSIS also sports a GUI development environment. ^xxiv As of this writing, SSIS doesn’t have native Kudu support; however you may be able to ingest data into Kudu via Impala and JDBC/ODBC.

Oracle Data Integrator for Big Data

Oracle Data Integrator for Big Data provides advanced data integration capabilities for big data platforms. It supports a diverse set of workloads, including Spark, Spark Streaming, and Pig transformations and connects to various big data sources such as Kafka and Cassandra. For Orchestration of ODI jobs, users have a choice between using ODI agents or Oozie as orchestration engines. ^xxv As of this writing, Oracle Data Integrator doesn’t have native Kudu support; however you may be able to ingest data into Kudu using Oracle Data Integrator via Impala and JDBC/ODBC.

IBM InfoSphere DataStage

IBM InfoSphere DataStage is IBM’s data integration tool that comes with IBM InfoSphere Information Server. It supports extract, transform, and load (ETL) and extract, load, and transform (ELT) patterns. ^xxvi It provides some big data support such as accessing and processing files on HDFS and moving tables from Hive to an RDBMS. ^xxvii The IBM InfoSphere Information Server operations console can be used to monitor and administer DataStage jobs, including monitoring their job logs and resource usage. The operations console also aids in troubleshooting issues when runtime issues occur. ^xxviii As of this writing, DataStage doesn’t have native Kudu support; however you may be able to ingest data into Kudu using DataStage via Impala and JDBC/ODBC.

Syncsort

Syncsort is another leading software development company specializing in big data integration. They are well known for their extensive support for mainframes. By 2013, Syncsort had positioned itself as a “Big Iron to Big Data” data integration company. ^xxix

Syncsort DMX-h, Syncsort’s big data integration platform, has a batch and streaming interfaces that can collect different types of data from multiple sources such as Kafka, HBase, RDBMS, S3, and Mainframes. ^xxx Like the other ETL solutions described in this chapter, Syncsort features an easy-to-use GUI for ETL development. ^xxxi As of this writing, DMX-h doesn’t have native Kudu support; however you may be able to ingest data into Kudu using DMX-h via Impala and JDBC/ODBC.

Apache NIFI

Apache NIFI is an open source real-time data ingestion tool mostly used in the Hortonworks environment, although it can be used to ingest data to and from other platforms such as Cloudera and MapR. It’s similar to StreamSets in many aspects. One of the main limitations of NIFI is its lack of YARN support. NIFI runs as an independent JVM process or multiple JVM processes if configured in cluster mode. ^xxxii NIFI does not have native Kudu support at the time of this writing, although the open source community is working on this. Check out NIFI-3973 for more details. ^xxxiii

Kudu and Spark

Kudu provides a Spark API that you can use to ingest data into Kudu tables. In the following example, we’ll join a table stored in a SQL Server database with another table stored in an Oracle database and insert the joined data into a Kudu table. Chapter 6 provides a more thorough discussion of Kudu and Spark.

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 the dataframes using the following 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.

Flume, Kafka, and Spark Streaming

Using Flume, Kafka, and Spark Streaming for real-time data ingestion and event processing is a common architectural pattern.Apache Flume

Allows users to collect, transform, and move large amounts of streaming and event data into HDFS, HBase, and Kafka to name a few. ^xxxiv It’s an integrated component of most Hadoop distributions. Flume’s architectures can be divided into sources, channels, and sinks. Sources are your data sources, channels provide an intermediate buffer between sources and sinks and provides reliability, and sinks represent your destination. Flume has a simple architecture as you can see in Figure 7-140.

Flume is configured using a simple configuration file. The example configuration file lets you generate events from a script and then logs them to the console.

# Name the components

agent1.sources = source1

agent1.sinks = sinks1

agent1.channels = channel1

# Configure the source

agent1.sources.source1.type = exec

agent1.sources.source1.command = /tmp/myscript.sh

agent1.sources.source1.channels = channel1

# Configure the channel

agent1.channels.channel1.type = memory

agent1.channels.channel1.capacity = 10000

agent1.channels.channel1.transactionCapacity = 1000

# Configure the sink

agent1.sinks.sinks1.type = logger

agent1.sinks.sinks1.channel = channel1

To learn more about Flume, Using Flume (O’Reilly, 2014) by Hari Shreedharan is the definitive guide.

Apache Kafka

Kafka is a fast, scalable, and reliable distributed publish-subscribe messaging system. Kafka is now a standard component of architectures that requires real-time data ingestion and streaming. Although not required, Kafka is frequently used with Apache Flume, Storm, Spark Streaming, and StreamSets.

Kafka runs as a cluster on one or more brokers and has built-in replication and partitioning as shown in Figure 7-141.

Kafka records are published as topics. Each topic is partitioned and continuously appended to. Each record in the partition is assigned a unique offset that identifies each record. ^xxxv Think of the topic as a table and the offset as a primary key (Figure 7-142).

To learn more about Kafka, Kafka: The Definitive Guide (O’Reilly, 2017) by Shapiro, Narkhede, and Paling, is the best resource available.

Flafka

When Kafka was first released in 2011, users had to write Java code in order to ingest data into and read data out of Kafka. Unfortunately, this requirement slowed the adoption of Kafka because not everyone can write code in Java. There has to be an easier way for applications to interact with Kafka. As discussed earlier, these tasks are easily accomplished with Flume (Figure 7-143).

The engineers at Cloudera and others in the open source community recognized the benefits of integrating Flume with Kafka, so they developed Flume and Kafka integration frequently referred to as Flafka. Starting in CDH 5.2, Flume can act as consumer and producer for Kafka. With further development included in Flume 1.6 and CDH 5.3, the ability for Kafka to act as a Flume channel has been added. ^xxxvi

Spark Streaming

Historically, Kafka did not provide any mechanism for stream processing. Kafka Streams was recently released to provide basic stream processing capabilities. Flume also has Interceptors that can be used for light-weight stream processing. However, Flume Interceptors have become out of vogue lately due to the popularity of Apache Spark. Spark Streaming is more powerful than Kafka Streams and Flume Interceptors due to Spark Streaming’s integration with Spark SQL, Spark MLlib, and other features of Spark (Figure 7-144). Unlike Interceptors, Spark Streaming doesn’t require Flume and can integrate with other data sources such as MQTT, Kafka, and Kinesis to name a few. If you have an unsupported data source that you want to integrate with Spark Streaming, you can implement a custom receiver that can handle that particular data source. Consult Chapter 2 for an example on how read and write to Kudu with Spark Streaming.

Sqoop

Sqoop is not technically compatible with Kudu. You cannot use Sqoop to transfer data from an RDBMS to Kudu and vice versa. However, Sqoop may be the only tool some users may have for ingesting data from an RDBMS.

What you can do is to use Sqoop to ingest data from an RDBMS into HDFS and then use Spark or Impala to read the data from HDFS and insert it into Kudu. Here’s a few examples of what you can do with Sqoop. Make sure you install and configure the correct drivers.

Get a list of available databases in SQL Server.

Copy a table from SQL Server to Hadoop.

Copy a table from Hive to SQL Server.

Then you can just simply execute an Impala insert into…select statement: