Using Cloudera Flow Management To Ingest and Process RSS Feeds

[#1]

NiFi for XML / RSS / REST Feed Ingest

I want to retrieve the status from various Cloud providers and services, including Cloudera, AWS, Azure, and Google. I have found many of the available status APIs will return XML/RSS. We love that format for Apache NiFi, so let's do that.

Note: If you are doing development in a non-production environment, try the new NiFi 1.13.1. If you need to run your flows in production on-premise, private cloud, or in the public cloud, then use Cloudera Flow Management.

I have separated the processing module "Status" from the input, so I can pass in the input anyway I want. When I move this to a K8 environment, this will become a parameter that I pass in. Stay tuned to Cloudera releases.

The flow is pretty simple to process RSS status data. We call the status URL and in the next step easily convert RSS into JSON for easier processing. I split these records and grab just the fields I like.

I can easily add additional fields from my metadata for unique id, timestamp, company name, and service name. PutKudu will store my JSON records as Kudu fields at high speed. If something goes wrong, we will try again. Sometimes, the internet is down! But, without this app, how will we know???

We can run a QueryRecord processor to query live fields from the status messages and I will send Spark-related ones to my Slack channel. I can add as many ANSI SQL92 Calcite queries as I wish. It's easy.

We were easily able to insert all the status messages to our 'cloudstatus' table. Now, we can query it and use it in reports, dashboards, and visual applications.

I don't want to have to go to external sites to get the status alerts, so I will post key ones to a Slack channel.

I want to store my status reads in a table for fast analytics and permanent storage. So, I will store it in a Kudu table with Impala on top for fast queries.

CREATE TABLE cloudstatus (
	`uuid` STRING,
	`ts` TIMESTAMP,
	`companyname` STRING,
	`servicename` STRING,
	`title` STRING,
	`description` STRING,
	`pubdate` STRING,
	`link` STRING,
	`guid` STRING,
	PRIMARY KEY (`uuid`,`ts` ) )
	PARTITION BY HASH PARTITIONS 4
	STORED AS KUDU TBLPROPERTIES ('kudu.num_tablet_replicas' = '1');

My source code is available here.

In the next step, I can write some real-time dashboard with Cloudera Visual Apps, add fast queries on Kafka with Flink SQL, or write some machine learning in Cloudera Machine Learning to finish the application. Join my next live video broadcast to suggest what we do with this data next.

Thanks for reading!