You have been tasked with filtering a noisy stream of syslog events which are available in a Kafka topic. The goal is to identify critical events and write them to another Kafka topic. Visit the Cloudera YouTube channel for a video walkthrough of this use case.
- On the CDP Public Cloud Home Page, navigate to DataFlow
- Navigate to the ReadyFlow Gallery
- Explore the ReadyFlow Gallery
Info:
The ReadyFlow Gallery is where you can find out-of-box templates for common data movement use cases. You can directly create deployments from a ReadyFlow or create new drafts and modify the processing logic according to your needs before deploying.
- Search for the “Kafka filter to Kafka” ReadyFlow.
- Click on “Create New Draft” to open the ReadyFlow in the Designer
- Select the only available workspace and give your draft a name
- Click Create. You will be forwarded to the Designer
- Start a Test Session by either clicking on the start a test session link in the banner or going to Flow Options and selecting Start in the Test Session section.
- In the Test Session creation wizard, keep the defaults for NiFi Runtime Version (latest), Inbound Connections (unchecked) and Custom NAR Configuration (unchecked). Click Start Test Session. Notice how the status at the top now says “Initializing Test Session”.
Note: Test Session initialization should take about 5-10 minutes.
Info:
Test Sessions provision infrastructure on the fly and allow you to start and stop individual processors and send data through your flow. By running data through processors step-by-step and using the data viewer as needed, you’re able to validate your processing logic during development in an iterative way without having to treat your entire data flow as one deployable unit.
The flow consists of three processors and looks very promising for our use case. The first processor reads data from a Kafka topic, the second processor allows us to filter the events before the third processor writes the filtered events to another Kafka topic. All we have to do now to reach our goal is to customize its configuration to our use case.
-
a. Navigate to Flow Options → Parameters
-
b. Select all parameters that show No value set and provide the following values.
Note: The parameter values that need to be copied from the Trial Manager homepage are found by selecting Manage Trial in the upper right corner and then selecting Configurations. See screenshots below.
| Name | Value |
|---|---|
| CDP Workload User | srv_nifi-kafka-ingest |
| CDP Workload User Password | <Copy the value for 'nifi-kafka-ingest-password' from Trial Manager homepage> |
| Data Input Format | AVRO |
| Data Output Format | JSON |
| Kafka Broker Endpoint | <Comma-separated list of Kafka Broker addresses. Copy the value for 'kafka_broker' from Trial Manager homepage> |
| Kafka Consumer Group ID | cdf |
| Kafka Destination Topic | syslog_critical |
| Kafka Producer ID | cdf |
| Kafka Source Topic | syslog_avro |
| Schema Name | syslog |
| Schema Registry Hostname | <Hostname of Schema Registry service. Copy the value for 'schema_registry_host_name' from Trial Manager homepage> |
- c. Click Apply Changes to save the parameter values
-
a. Once your test session has started, navigate to Flow Options → Services
-
b. Select CDP_Schema_Registry service and click Enable Service and Referencing Components action
Note: If clicking Enable Service and Referencing Components does not cause CDP_Schema_Registry and other services to enable, perform the following steps:
-
Click Enable.
-
The CDP_Schema_Registry service should be enabled. Click Disable.
-
Click Enable Service and Referencing Components.
-
-
c. Make sure all services have been enabled
Navigate back to the Flow Designer canvas by clicking on the Back To Flow Designer link at the top of the screen.
3. Start the ConsumeFromKafka processor using the right click action menu or the Start button in the configuration drawer if it hasn’t been started automatically.
4. Stop the Filter Events processor if it is running using the right click action menu or the Stop button in the configuration drawer.
After starting the ConsumeFromKafka processor and stopping the remaining processors, you should see events starting to queue up in the success_ConsumeFromKafka-FilterEvents connection.
-
a. Right-click on the success_ConsumeFromKafka-FilterEvents connection and select List Queue
Info:
The List Queue interface shows you all flow files that are being queued in this connection. Click on a flow file to see its metadata in the form of attributes. In our use case, the attributes tell us a lot about the Kafka source from which we are consuming the data. Attributes change depending on the source you’re working with and can also be used to store additional metadata that you generate in your flow.
-
b. Select any flow file in the queue and click the book icon to open it in the Data Viewer
Info:
The Data Viewer displays the content of the selected flow file and shows you the events that we have received from Kafka. It automatically detects the data format - in this case JSON - and presents it in human readable format.
-
c. Scroll through the content and note how we are receiving syslog events with varying severity.
-
a. Return to the Flow Designer by closing the Data Viewer tab and clicking Back To Flow Designer in the List Queue screen.
-
b. Select the Filter Events processor on the canvas. We will be using a QueryRecord processor to filter out low severity events. The QueryRecord processor is very flexible and can run several filtering or routing rules at once.
-
c. In the configuration drawer, scroll down until you see the filtered_events property. We are going to use this property to filter out the events. Click on the menu at the end of the row and select Go To Parameter.
-
d. Once you are on the Parameters view, you can see that this parameter already has a SQL query assigned to it. However it currently does not do any filtering - rather it passes through all events.
-
e. Modify the parameter value to:
SELECT * FROM FLOWFILE WHERE severity <= 2
-
f. Click Apply Changes to update the parameter value. Return back to the Flow Designer.
-
g. Select the Filter Events processor on the canvas. Start the Filter Events processor using the right-click menu or the Start icon in the configuration drawer.
-
a. After starting the Filter Events processor, flow files will start queueing up in the filtered_events-FilterEvents-WriteToKafka connection
-
b. Right click the filtered_events-FilterEvents-WriteToKafka connection and select List Queue.
-
c. Select a few random flow files and open them in the Data Viewer to verify that only events with severity <=2 are present.
-
d. Navigate back to the Flow Designer canvas.
Now all that is left is to start the WriteToKafka processor to write our filtered high severity events to syslog_critical Kafka topic.
- a. Select the WriteToKafka processor and explore its properties in the configuration drawer.
- b. Note how we are plugging in many of our parameters to configure this processor. Values for properties like Kafka Brokers, Topic Name, Username, Password and the Record Writer have all been parameterized and use the values that we provided in the very beginning.
- c. Start the WriteToKafka processor using the right-click menu or the Start icon in the configuration drawer.
Congratulations! You have successfully customized this ReadyFlow and achieved your goal of sending critical alerts to a dedicated topic! Now that you are done with developing your flow, it is time to deploy it in production!
-
a. Click the toggle next to Active Test Session to stop your Test Session
-
b. Click “End” in the dialog to confirm. The Test Session is now stopping and allocated resources are being released
-
a. Open the “Flow Options” menu at the top
-
b. Click “Publish” to make your modified flow available in the Catalog
-
c. Append your username to the Flow Name and provide a Flow Description. Click Publish.
-
d. You are now redirected to your published flow definition in the Catalog.
Info:
The Catalog is the central repository for all your deployable flow definitions. From here you can create auto-scaling deployments from any version or create new drafts and update your flow processing logic to create new versions of your flow.
Click Deploy, select the available environment from the drop down menu and click Continue to start the Deployment Wizard.
The Deployment Wizard guides you through a six step process to create a flow deployment. Throughout the six steps you will choose the NiFi configuration of your flow, provide parameters and define KPIs. At the end of the process, you are able to generate a CLI command to automate future deployments.
-
a. Provide a name such as Critical Syslogs - Prod to indicate the use case and that you are deploying a production flow. Click Next.
-
b. The NiFi Configuration screen allows you to customize the runtime that will execute your flow. You have the opportunity to pick from various released NiFi versions.
Change the NiFi version to a Previous Version and make sure Automatically start flow upon successful deployment is checked.
Click Next.
-
c. The Parameters step is where you provide values for all the parameters that you defined in your flow. In this example, you should recognize many of the prefilled values from the previous exercise - including the Filter Rule and our Kafka Source and Kafka Destination Topics.
To advance, you have to provide values for all parameters. Select the No Value option to only display parameters without default values.
You should now only see one parameter - the CDP Workload User Password parameter which is sensitive. Sensitive parameter values are removed when you publish a flow to the catalog to make sure passwords don’t leak.
Provide the CDP Workload User Password (the value for 'nifi-kafka-ingest password' from the Trial Manager homepage) and click Next to continue.
-
d. The Sizing & Scaling step lets you choose the resources that you want to allocate for this deployment. You can choose from several node configurations and turn on Auto-Scaling.
Let’s choose the Extra Small Node Size and turn on Auto-Scaling from 1-3 nodes.
Check the Flow Metrics Scaling option that appears after Auto-Scaling has been enabled. In addition to CPU based scaling, Flow Metrics Scaling uses back pressure prediction based on data in your flow to make intelligent scaling choices.
Click Next.
-
e. The Key Performance Indicators (KPI) step allows you to monitor flow performance. You can create KPIs for overall flow performance metrics or in-depth processor or connection metrics.
Add the following KPI
-
KPI Scope: Entire Flow
-
Metric to Track: Data Out
-
Alerts:
- Trigger alert when metric is less than: 1 MB/sec
- Alert will be triggered when metrics is outside the boundary(s) for: 1 Minute
Add the following KPI
-
KPI Scope: Processor
-
Processor Name: ConsumeFromKafka
-
Metric to Track: Bytes Received
-
Alerts:
- Trigger alert when metric is less than: 512 KBytes
- Alert will be triggered when metrics is outside the boundary(s) for: 30 seconds
Review the KPIs and click Next.
-
-
f. In the Review page, review your deployment details.
-
g. Click Deploy to initiate the flow deployment!
-
h. You are redirected to the Deployment Dashboard where you can monitor the progress of your deployment.
Note: Creating the deployment may take up to 15 minutes. Only the latest NiFi version is cached on the nodes. Any other version (like the previous version selected earlier for this deployment) has to be downloaded.
-
i. Congratulations! Your flow deployment has been created and is already processing Syslog events!
-
Notice how the dashboard shows you the data rates at which a deployment currently receives and sends data. The data is also visualized in a graph that shows the two metrics over time.
-
Change the Metrics Window setting at the top right. You can visualize as much as 24 hours.
-
Click on the Critical Syslogs - Prod deployment. The side panel opens and shows more detail about the deployment. On the KPIs tab it will show information about the KPIs that you created when deploying the flow.
Using the two KPIs Bytes Received and Data Out we can observe that our flow is filtering out data as expected since it reads more than it sends out.
-
Switch to the System Metrics tab where you can observe the current CPU utilization rate for the deployment. Our flow is not doing a lot of heavy transformation, so CPU utilization shouldn’t be too high.
-
Close the side panel by clicking anywhere on the Dashboard.
-
Notice how your Critical Syslogs - Prod deployment shows Concerning Health status. Hover over the warning icon and click View Details.
-
You will be redirected to the Alerts tab of the deployment. Here you get an overview of active and past alerts and events. Expand the Active Alert to learn more about its cause.
After expanding the alert, it is clear that it is caused by a KPI threshold breach for sending less than 1MB/s to external systems as defined earlier when you created the deployment.
-
Click on the Critical Syslogs - Prod deployment in the Dashboard. In the side panel, click Manage Deployment at the top right.
-
You are now being redirected to the Deployment Manager. The Deployment Manager allows you to reconfigure the deployment and modify KPIs, modify the number of NiFi nodes or turn auto-scaling on/off or update parameter values.
-
Explore NiFi UI for deployment. Click the Actions menu and click on View in NiFi.
-
You are being redirected to the NiFi cluster running the flow deployment. You can use this view for in-depth troubleshooting. Users can have read-only or read/write permissions to the flow deployment. Double click on the NiFi process group to see the deployment flow.
-
Go back to the Deployment Manager expand the Actions menu and click on Change NiFi Runtime Version
-
Looks like we’re running an older version of NiFi. Let’s upgrade to the latest! Select the latest version from the dropdown and click “Update” (if you chose the latest NiFi version when creating the deployment, you can use the Update action to downgrade to the previous NiFi version too).
You are redirected to the Dashboard and can monitor the upgrade process. Upgrading NiFi versions is as simple as that! One click!
Note: The NiFi version update may take up to 5-10 minutes.
Congratulations, you have completed the first use case! While the NiFi version of your deployment is being upgraded, feel free to proceed to the next use case!
Info:
The completed flow definition for this use case can be found in the Catalog under the name "Use Case 1 - Cloudera - Filter critical syslog events". If you run into issues during your flow development, it may be helpful to select this flow in the Catalog and select "Create New Draft" to open it in the Flow Designer to compare to your own.