• Make sure you pick S2S not RAW in Cloud Connection to NiFi

• Make sure No Spaces Before or After Destination ID, URL, Names, Topics, Brokers, Etc…​

• Make sure No Spaces Anywhere

• Everything is Case-Sensitive. It’s IoT, not iot.

• Use User=Admin for CDSW and HUE

• You must have HDFS User Created via HUE, Go there First

• Check all your connections and spellings

• Check /opt/cloudera/cem/minifi/logs/minifi-app.log if you can’t find an issue

1. Delete the agent manifest manually using the EFM API:

2. Verify each class has the same agent manifest ID:

   http://hostname:10080/efm/api/agent-classes
   [{"name":"iot1","agentManifests":["agent-manifest-id"]},{"name":"iot4","agentManifests":["agent-manifest-id"]}]

3. Confirm the manifest doesn’t have the NAR you installed

   http://hostname:10080/efm/api/agent-manifests?class=iot4
   [{"identifier":"agent-manifest-id","agentType":"minifi-java","version":"1","buildInfo":{"timestamp":1556628651811,"compiler":"JDK 8"},"bundles":[{"group":"default","artifact":"system","version":"unversioned","componentManifest":{"controllerServices":[],"processors":

4. Call the API endpoint:

   http://hostname:10080/efm/swagger/

5. Hit the DELETE - Delete the agent manifest specified by id button, and in the id field, enter `agent-manifest-id

Before we start building our flow, let’s create a Process Group to help organizing the flows in the NiFi canvas and also to enable flow version control.

1. Open the NiFi Web UI, create a new Process Group and name it something like Process Sensor Data.

   

2. We want to be able to version control the flows we will add to the Process Group. In order to do that, we first need to connect NiFi to the NiFi Registry. On the NiFi global menu, click on "Controller Services", navigate to the "Registry Clients" tab and add a Registry client with the following URL:

   Name: NiFi Registry
   URL:  http://edge2ai-1.dim.local:18080

   
   

3. On the NiFi Registry Web UI, add another bucket for storing the Sensor flow we’re about to build'. Call it SensorFlows:

   

4. Back on the NiFi Web UI, to enable version control for the Process Group, right-click on it and select Version > Start version control and enter the details below. Once you complete, a will appear on the Process Group, indicating that version control is now enabled for it.

   Registry:  NiFi Registry
   Bucket:    SensorFlows
   Flow Name: SensorProcessGroup

5. Let’s also enable processors in this Process Group to use schemas stored in Schema Registry. Right-click on the Process Group, select Configure and navigate to the Controller Services tab. Click the + icon and add a HortonworksSchemaRegistry service. After the service is added, click on the service’s cog icon (), go to the Properties tab and configure it with the following Schema Registry URL and click Apply.

   URL: http://edge2ai-1.dim.local:7788/api/v1

   

6. Click on the lightning bolt icon () to enable the HortonworksSchemaRegistry Controller Service.

7. Still on the Controller Services screen, let’s add two additional services to handle the reading and writing of JSON records. Click on the button and add the following two services:

   • JsonTreeReader, with the following properties:

     Schema Access Strategy: Use 'Schema Name' Property
     Schema Registry:        HortonworksSchemaRegistry
     Schema Name:            ${schema.name} -> already set by default!

   • JsonRecordSetWriter, with the following properties:

     Schema Write Strategy:  HWX Schema Reference Attributes
     Schema Access Strategy: Inherit Record Schema
     Schema Registry:        HortonworksSchemaRegistry

8. Enable the JsonTreeReader and the JsonRecordSetWriter Controller Services you just created, by clicking on their respective lightning bolt icons ().

   

1. Double-click on the newly created process group to expand it.

2. Inside the process group, add a new Input Port and name it "Sensor Data"

3. We need to tell NiFi which schema should be used to read and write the Sensor data. For this we’ll use an UpdateAttribute processor to add an attribute to the FlowFile indicating the schema name.

   Add an UpdateAttribute processor by dragging the processor icon to the canvas:

   

4. Double-click the UpdateAttribute processor and configure it as follows:

   1. In the SETTINGS tab:

      Name: Set Schema Name

   2. In the PROPERTIES tab:

      • Click on the button and add the following property:

        Property Name:  schema.name
        Property Value: SensorReading

   3. Click Apply

5. Connect the Sensor Data input port to the Set Schema Name processor.

6. Add a PublishKafkaRecord_2.0 processor and configure it as follows:

   SETTINGS tab:

   Name:                                  Publish to Kafka topic: iot

   PROPERTIES tab:

   Kafka Brokers:                         edge2ai-1.dim.local:9092
   Topic Name:                            iot
   Record Reader:                         JsonTreeReader
   Record Writer:                         JsonRecordSetWriter
   Use Transactions:                      false
   Attributes to Send as Headers (Regex): schema.*

   Note	Make sure you use the PublishKafkaRecord_2.0 processor and not the PublishKafka_2.0 one

7. While still in the PROPERTIES tab of the PublishKafkaRecord_2.0 processor, click on the button and add the following property:

   Property Name:  client.id
   Property Value: nifi-sensor-data

   Later, this will help us clearly identify who is producing data into the Kafka topic.

8. Connect the Set Schema Name processor to the Publish to Kafka topic: iot processor.

9. Add a new Funnel to the canvas and connect the PublishKafkaRecord processor to it. When the "Create connection" dialog appears, select "failure" and click Add.

   

10. Double-click on the Publish to Kafka topic: iot processor, go to the SETTINGS tab, check the "success" relationship in the AUTOMATICALLY TERMINATED RELATIONSHIPS section. Click Apply.

    

11. Start the input port and the two processors. Your canvas should now look like the one below:

    

12. The only thing that remains to be configured now is to finally connect the "from Gateway" Input Port to the flow in the "Processor Sensor Data" group. To do that, first go back to the root canvas by clicking on the NiFi Flow link on the status bar.

    

13. Connect the Input Port to the Process Sensor Data Process Group by dragging the destination of the current connection from the funnel to the Process Group. When prompted, ensure the "To input" fields is set to the Sensor data Input Port.

    
    

14. Refresh the screen (Ctrl+R on Linux/Windows; Cmd+R on Mac) and you should see that the records that were queued on the "from Gateway" Input Port disappeared. They flowed into the Process Sensor Data flow. If you expand the Process Group you should see that those records were processed by the PublishKafkaRecord processor and there should be no records queued on the "failure" output queue.

    

    At this point, the messages are already in the Kafka topic. You can add more processors as needed to process, split, duplicate or re-route your FlowFiles to all other destinations and processors.

15. To complete this Lab, let’s commit and version the work we’ve just done. Go back to the NiFi root canvas, clicking on the "Nifi Flow" breadcrumb. Right-click on the Process Sensor Data Process Group and select Version > Commit local changes. Enter a descriptive comment and save.

When the sensor data was sent to Kafka using the PublishKafkaRecord processor, we chose to attach the schema information in the header of Kafka messages. Now, instead of hard-coding which schema we should use to read the message, we can leverage that metadata to dynamically load the correct schema for each message.

To do this, though, we need to configure a different JsonTreeReader that will use the schema properties in the header, instead of the ${schema.name} attribute, as we did before.

We’ll also add a new RestLookupService controller service to perform the calls to the CDSW model API endpoint.

1. If you’re not in the Process Sensor Data process group, double-click on it to expand it. On the Operate panel (left-hand side), click on the cog icon () to access the Process Sensor Data process group’s configuration page.

   

2. Click on the plus button (), add a new JsonTreeReader, configure it as shown below and click Apply when you’re done:

   On the SETTINGS tab:

   Name: JsonTreeReader - With schema identifier

   On the PROPERTIES tab:

   Schema Access Strategy: HWX Schema Reference Attributes
   Schema Registry:        HortonworksSchemaRegistry

3. Click on the lightning bolt icon () to enable the JsonTreeReader - With schema identifier controller service.

4. Click again on the plus button (), add a RestLookupService controller service, configure it as shown below and click Apply when you’re done:

   On the PROPERTIES tab:

   URL:           http://cdsw.<YOUR_CLUSTER_PUBLIC_IP>.nip.io/api/altus-ds-1/models/call-model
   Record Reader: JsonTreeReader
   Record Path:   /response

   Note	<YOUR_CLUSTER_PUBLIC_IP> above must be replaced with your cluster’s public IP, not DNS name. The final URL should look something like this: http://cdsw.12.34.56.78.nip.io/api/altus-ds-1/models/call-model

5. Click on the lightning bolt icon () to enable the RestLookupService controller service.

   

6. Close the Process Sensor Data Configuration page.

We’ll now create the flow to read the sensor data from Kafka, execute a model prediction for each of them and write the results to Kudu. At the end of this section you flow should look like the one below:

1. Open CDSW Web UI and log in as admin, if you haven’t yet done so.

2. Navigate to the CDSW Admin page to fine tune the environment:

   1. In the Engines tab, add in Engines Profiles a new engine (docker image) with 2 vCPUs and 4 GB RAM, while deleting the default engine.

   2. Check if the following variable already exists under Environmental Variables. If not, add it:

      HADOOP_CONF_DIR=/etc/hadoop/conf/

      

1. Return to the main page and click on New Project, using this GitHub project as the source: https://github.com/tspannhw/edge2ai-workshop

   

2. Now that your project has been created, click on Open Workbench and start a Python3 session:

   

3. Once the Engine is ready, run the following command to install some required libraries:

   !pip3 install --upgrade pip scikit-learn

4. The project comes with a historical dataset. Copy this dataset into HDFS:

   !hdfs dfs -put -f data/historical_iot.txt /user/$HADOOP_USER_NAME

   

5. You’re now ready to run the Experiment to train the model on your historical data.

6. You can stop the Engine at this point.

1. Open the project you created in the previous lab and examine the file in the Workbench. This PySpark program uses the pickle.load mechanism to deploy models. The model is loaded from the iot_modelf.pkl file, which was saved in the previous lab from the experiment with the best predictive model.

   There program also contains the predict definition, which is the function that calls the model, passing the features as parameters, and will return a result variable.

2. Before deploying the model, try it out in the Workbench: launch a Python3 engine and run the code in file cdsw.iot_model.py. Then call the predict() method from the prompt:

   predict({"feature": "0, 65, 0, 137, 21.95, 83, 19.42, 111, 9.4, 6, 3.43, 4"})

   

3. The functions returns successfully, so we know we can now deploy the model. You can now stop the engine.

1. From the main page of your project, select the Models button. Select New Model and specify the following configuration:

   Name:          IoT Prediction Model
   Description:   IoT Prediction Model
   File:          cdsw.iot_model.py
   Function:      predict
   Example Input: {"feature": "0, 65, 0, 137, 21.95, 83, 19.42, 111, 9.4, 6, 3.43, 4"}
   Kernel:        Python 3
   Engine:        2 vCPU / 4 GB Memory
   Replicas:      1

   

2. After all parameters are set, click on the Deploy Model button. Wait till the model is deployed. This can take several minutes.

1. When your model status change to Deployed, click on the model name link to go to the Model’s Overview page. From the that page, click on the Test button to check if the model is working.

2. The green circle with the success status indicates that our REST call to the model is working. The 1 in the response {"result": 1}, means that the machine from where these temperature readings were collected is unlikely to experience a failure.

   

3. Now, lets change the input parameters and call the predict function again. Put the following values in the Input field:

   {
     "feature": "0, 95, 0, 88, 26.62, 75, 21.05, 115, 8.65, 5, 3.32, 3"
   }

4. With these input parameters, the model returns 0, which means that the machine is likely to break.

• Make sure you pick S2S not RAW in Cloud Connection to NiFi

• Make sure No Spaces Before or After Destination ID, URL, Names, Topics, Brokers, Etc…​

• Make sure No Spaces Anywhere

• Everything is Case-Sensitive. It’s IoT, not iot.

• Use User=Admin for CDSW and HUE

• You must have HDFS User Created via HUE, Go there First

• Check all your connections and spellings

• Check /opt/cloudera/cem/minifi/logs/minifi-app.log if you can’t find an issue

1. Delete the agent manifest manually using the EFM API:

2. Verify each class has the same agent manifest ID:

   http://hostname:10080/efm/api/agent-classes
   [{"name":"iot1","agentManifests":["agent-manifest-id"]},{"name":"iot4","agentManifests":["agent-manifest-id"]}]

3. Confirm the manifest doesn’t have the NAR you installed

   http://hostname:10080/efm/api/agent-manifests?class=iot4
   [{"identifier":"agent-manifest-id","agentType":"minifi-java","version":"1","buildInfo":{"timestamp":1556628651811,"compiler":"JDK 8"},"bundles":[{"group":"default","artifact":"system","version":"unversioned","componentManifest":{"controllerServices":[],"processors":

4. Call the API endpoint:

   http://hostname:10080/efm/swagger/

5. Hit the DELETE - Delete the agent manifest specified by id button, and in the id field, enter agent-manifest-id

if you get transfer relationship not specified in java

just change the schema access strategy from inherited to from attribute in the json writter

this will change in NiFi 1.11

OR

workaround

add the new column in the schema

in HWX Schema Registry

if they add the new column in the Schema Registry it went through