StreamSets tutorial: Stream real-time data

IBM watsonx.data integration works with Data Collector engines when you build flows and run the flows as jobs. Because the engine runs jobs in your corporate network, you maintain all ownership and control of your data.

Create a StreamSets environment to configure Data Collector engines for your project. You can then run engines based on this configuration.

export SSET_API_KEY=<cloud_api_key>

Create a StreamSets flow to define how to transform source data before writing it to target systems.

1. On the Assets tab of your project, click, New Asset > Create a real-time streaming data flow.

2. Specify the a name for the tutorial, such as StreamSets tutorial.

3. Select the environment that you created, and click Create.
   The engine health column for the environment should show that at least one engine associated with the environment is running. If an engine is not running, see Running the engine command.

4. Click the Settings icon , and then click Run.
   The Runtime Environment tab shows all of the StreamSets environments in the project, and the environment selected for the flow. You can use this tab to change the environment associated with the flow, like when you want to configure a flow to run on a production engine instead of a development engine.

5. Click the Error Records tab.
   On the Error Records tab, you can specify what to do with error records. By default, the flow discards error records. You can write error records to file or to a system such as Amazon S3, Google, and Kafka when you have the associated stage library installed. For more information, see Error record handling.

6. Click Cancel to exit the flow settings.

A flow used in production includes a source stage that connects to the source system, such as a Kafka Multitopic Consumer to read data from Kafka topics, to provide data for the flow.

In this tutorial, you will use one of the Data Collector development stages to provide source data. Development stages provide a simple, temporary way to include sample data in the flow so you can focus on flow processing. You will also use preview to see how the data appears in the flow.

1. On the canvas, expand the Sources list and select Dev Raw Data Source.

2. In the canvas, click the Dev Raw Data Source stage. Then, delete the default data in the Raw Data property.

3. Copy the following data, paste it into the Raw Data property, and then click Save.
   By default, Dev Raw Data Source accepts JSON objects, like this sample data. When needed, you can configure it to process other data formats, such as delimited, log, or Parquet data.

{
    "ID": 100288,
    "NAME": "Gayler Crosetti",
    "INCOME": 44956,
    "DEBT": 8060,
    "ZIP_CODE": 95401,
    "EMAIL_ADDRESS": "gcrosetti2a@1und1.de",
    "PHONE_NUMBER": "516-164-2822",
    "SSN": "857-36-6854",
    "MARITAL_STATUS": "M"
  }
  {
    "ID": 100310,
    "NAME": "Zuzana Dory",
    "INCOME": 43972,
    "DEBT": 340,
    "ZIP_CODE": 92629,
    "EMAIL_ADDRESS": "zdoryb4@bing.com",
    "PHONE_NUMBER": "805-143-7916",
    "SSN": "400-24-1226",
    "MARITAL_STATUS": "D"
  }
  {
    "ID": 100338,
    "NAME": "Joellyn Pilgram",
    "INCOME": 45813,
    "DEBT": 7385,
    "ZIP_CODE": 90001,
    "EMAIL_ADDRESS": "jpilgram8z@wunderground.com",
    "PHONE_NUMBER": "952-963-8176",
    "SSN": "521-13-0320",
    "MARITAL_STATUS": "S"
  }
  {
    "ID": 100381,
    "NAME": "Rebe Sicely",
    "INCOME": 53115,
    "DEBT": 674,
    "ZIP_CODE": 95661,
    "EMAIL_ADDRESS": "rsicely79@dell.com",
    "PHONE_NUMBER": "480-949-5817",
    "SSN": "842-77-3741",
    "MARITAL_STATUS": "W"
  }
  {
    "ID": 100389,
    "NAME": "Griff Ping",
    "INCOME": 59036,
    "DEBT": 8915,
    "ZIP_CODE": 91316,
    "EMAIL_ADDRESS": "gpingqq@shop-pro.jp",
    "PHONE_NUMBER": "432-175-5363",
    "SSN": "586-24-6396",
    "MARITAL_STATUS": "S"
  }
  {
    "ID": 100390,
    "NAME": "Essy Kollach",
    "INCOME": 53559,
    "DEBT": 2646,
    "ZIP_CODE": 94580,
    "EMAIL_ADDRESS": "ekollachgg@wunderground.com",
    "PHONE_NUMBER": "847-682-1704",
    "SSN": "845-87-8528",
    "MARITAL_STATUS": "M"
  }
  {
    "ID": 100423,
    "NAME": "Osmond Dunn",
    "INCOME": 49007,
    "DEBT": 13360,
    "ZIP_CODE": 90602,
    "EMAIL_ADDRESS": "odunndz@foxnews.com",
    "PHONE_NUMBER": "805-622-4620",
    "SSN": "789-62-0613",
    "MARITAL_STATUS": "M"
  }

4. To see how the data looks in the flow, click the Preview icon , and then click Preview.
   Preview displays the source data as the output for the Dev Raw Data Source. You can view the field name, field type and value, and record header attributes.
   You can run preview when all stages in the canvas are connected and all required stage properties are defined.

Data Collector includes a range of processor stages that process data on a record-by-record basis.
With this task, you add the following processors to the flow:

• Field Remover to remove unnecessary data

• Field Masker to mask sensitive data to protect Golden Bank customers

• Expression Evaluator to convert marital status codes to the expected formats

1. If you are continuing from Task 3, you are still previewing the source data. If you are returning from a break, click the Preview icon, and then click Preview to continue developing the flow in preview mode.

2. To add a processor, expand the Processors list and select the Field Remover, then double-click the stage in the canvas.
   Notice how the preview panel displays the source data as the input data for the processor. You will use this processor to remove contact information since the lenders and analysts do not make direct contact with customers.

   1. In the configuration panel, leave the Action property set to Remove Listed Fields.

   2. Click in the Fields property and select the following fields for removal:

      • NAME
      • EMAIL_ADDRESS
      • PHONE_NUMBER

      You can scroll to find the field names or simply start typing a field name.
      Fields are available for selection because you already ran preview. Fields are preceded by a slash character ( / ) because these are first-level fields. To specify these fields before running preview, you would enter /NAME, /EMAIL_ADDRESS, and /PHONE_NUMBER.

   3. Click Save.
      

3. Click Rerun preview.
   Notice how preview highlights the fields that have been removed from the input data for each record. We want to continue reviewing how the data, so let’s stay in preview while configuring the rest of the flow.

4. In the Processor list, select the Field Masker, then double-click the stage in the canvas.
   You will use this processor to partially mask the SSN field. The lenders and analysts use the last four digits of the Social Security number for tracking, but the rest should be masked to protect customer data.

   1. On the Mask tab, in the Field mask configurations table, click the Edit icon for the first row in the table.

   2. In the Fields to Mask property, select the SSN field.

   3. For the Mask Type property, select Custom.

   4. In the Custom mask field, enter the following mask: xxx-xx-####. The hash symbol reveals the characters in the specified location.

   5. Click OK to save the mask configuration, then click Save to save the processor configuration.

5. Click Rerun preview.
   Notice how the output records mask all but the last four digits of all Social Security numbers.

6. In the Processor list, select the Expression Evaluator, then double-click the stage in the canvas.
   You will use this processor to change marital status codes, such as changing S to Single.
   With this processor, you use the IBM StreamSets expression language to define expressions that modify record data. The IBM StreamSets expression language is based on the JSP 2.0 expression language. For more information, see Expression language.
   When you configure this processor, you specify the output field and the expression that generates the data for that field. If you specify a field that does not exist, the processor creates the field. If you specify an existing field, the processor overwrites the existing field.

   1. In the Field expressions table, click the Edit icon for the first row in the table.

   2. For the Output field property, select MARITAL_STATUS.
      This writes the results to the original field. When you want to keep the original data, you specify a new field name that the processor creates.

   3. For the Field expression property, delete the default value and paste the following expression:

      ${str:replace(str:replace(str:replace(str:replace(record:value('/MARITAL_STATUS'), 'M', 'Married'), 'S', 'Single'), 'D', 'Divorced'), 'W', 'Widowed')}

   4. Click OK, then click Save.

7. Click Rerun preview.
   Notice how the marital status codes have been replaced in the output column for the Expression Evaluator processor.

A flow used in production includes one or more target stages to write data to your target systems, such as a Apache Kafka Producer target to stream changed data to Apache Kafka.

To avoid writing to one of your target systems, you will use another development stage to complete the flow. Then, you can run the flow and view runtime statistics.

1. If you are still previewing the flow, click Exit Preview.

2. In the Stages list, type Tra and select the Trash target.
   There are no configuration options for the Trash target. The target simply discards all data.

3. Click the Save icon to save the flow.

4. To run the flow, click Run.
   A message indicates that the flow is valid, then changes to indicate that the job is running.
   After the job starts running, a real time summary displays for the selected stage. Real time summaries are available while the job runs.

5. Select other stages in the flow to view real time summaries for each stage, then click the canvas to view flow statistics.
   Notice how quickly the flow processes the data. Though you only provided a few sample records, Dev Raw Data Source passes the data continuously until you stop the flow.

6. To stop the flow, click Stop.

As a data engineer at Golden Bank, you have successfully removed unnecessary data, masked sensitive data, and converted data to the format expected by lenders and analysts. Well done!