tKafkaInput – Docs for ESB 5.x

tKafkaInput

1. Create the Kafka topic to be used to categorize the messages. The
   following command is used for demonstration purposes only. If you need
   further information about the creation of a Kafka topic, see Apache’s
   documentation for
   Kafka.

   /usr/lib/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic activities --partitions 1 --replication-factor 1

   1	/usr/lib/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic activities --partitions 1 --replication-factor 1

   This command creates a topic named activities, using
   the Kafka brokers managed by the Zookeeper service on the localhost machine.

2. Publish the message you want to analyze to the activities topic you have just created. In this scenario,
   Kafka is used to perform this publication using, for example, the following
   command:

   echo 'Ryan|M|Pool Remy|M|Drink Remy|M|Hiking Irene|F|Drink Pierre|M|Movie Irene|F|Pool Thomas|M|Drink Ryan|M|Cooking Wang|F|Cooking Chen|M|Drink | /usr/lib/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic activities

   1 2 3 4 5 6 7 8 9 10

   echo 'Ryan|M|Pool Remy|M|Drink Remy|M|Hiking Irene|F|Drink Pierre|M|Movie Irene|F|Pool Thomas|M|Drink Ryan|M|Cooking Wang|F|Cooking Chen|M|Drink | /usr/lib/kafka/bin/kafka-console-producer.sh --broker-list  localhost:9092 --topic  activities

   This
   command publishes 10 simple messages

   Ryan|M|Pool Remy|M|Drink Remy|M|Hiking Irene|F|Drink Pierre|M|Movie Irene|F|Pool Thomas|M|Drink Ryan|M|Cooking Wang|F|Cooking Chen|M|Drink

   1 2 3 4 5 6 7 8 9 10

   Ryan|M|Pool Remy|M|Drink Remy|M|Hiking Irene|F|Drink Pierre|M|Movie Irene|F|Pool Thomas|M|Drink Ryan|M|Cooking Wang|F|Cooking Chen|M|Drink

   As explained previously, you can use your actual message producer system instead to
   perform the publication.

1. In the Integration perspective
   of the Studio, create an empty Storm Job from the Job
   Designs node in the Repository tree view.

   For further information about how to create a Storm Job, see Talend Big Data Getting Started Guide.

2. In the workspace, enter the name of the component to be used and select
   this component from the list that appears. In this scenario, the components
   are tKafkaInput, tJavaStorm, tAggregateRow,
   and tLogRow.

3. Connect them using Row > Main links.

1. Click Run to open its view and then click
   Storm configuration to set up the
   connection to the Storm system to be used.

   

2. In the Storm node roles area, indicate
   the locations of the Nimbus server and the DRPC endpoint of the Storm
   cluster to be used:

   • Local mode: select this check
     box to build the Storm environment within the Studio. In this
     situation, the Storm Job you are designing is run locally within
     the Studio and you do not need to define any specific Nimbus and
     DRPC endpoint addresses.

   • Nimbus host and Port: in these two fields, enter the
     location of the Storm Nimbus server and its port number,
     respectively.

   • DRPC endpoint and Port: in these two fields, enter the
     location of the Storm DRPC endpoint and its port number,
     respectively.

3. In the Topology name field, enter the
   name you want the Storm system to use for the Storm Job, or topology in
   terms of the Storm, you are designing. It is recommended to use the same
   name as you have named this Storm Job so that you can easily recognize this
   Job even within the Storm system.

4. Select the Kill existing topology check box to make the
   Storm system stop any topology that has the same name as the Job you are
   designing and want to run.

   If you clear this check box and that Job of the same name is already running in the Storm
   system, the current Job will fail when you send it to Storm to run.

5. Select the Submit topology check box to submit the
   current Job to the Storm system. This feature is used when you need to send
   a new topology to Storm or used together with the Kill
   existing topology feature when you need to update a running
   topology in the Storm system.

   If you clear this check box, when you run the current Job, the submission
   of this Job is ignored while the other configuration information is still
   taken into account, for example, killing an existing topology.

6. Select the Monitor topology after
   submission check box to monitor the current Storm Job in the
   console of the Run view.

   If you clear this check box, you cannot read the monitoring information in
   the console.

7. In the Stop monitoring after timeout
   reached field, enter, without the double quotation marks, the
   numeric value to indicate whether to stop the monitoring when a running
   topology reaches its timeout. The default value is -1, which means no timeout is applied.

8. Select the Kill topology on quitting Talend Job check
   box to allow the Studio to kill, if it is still running, the current Job
   from the Storm system when you stop this Job from the Studio.

   If you clear this check box, the topology for this Job continues to run in the Storm
   system even though you kill it within the Studio or eventually shut down the
   Studio.

9. If you need to use any other Storm properties specific to your situation,
   add them to the Storm configuration
   table.

1. Double-click tKafkaInput to open its
   Component view.

   

2. In the Zookeeper host field, enter the location of the
   Zookeeper service used to coordinate the Kafka cluster to be used.

3. In the Port field, enter the port number of this
   Zookeeper service.

4. In the Topic name field, enter the name of the topic in
   which you need to receive messages. In this scenario, the topic is activities.

1. Double-click tJavaStorm to open its
   Component view.

   

2. Click the […] button next to Edit schema to open the schema editor.

3. On the output side (right), click the [+] button three
   times to add three rows and in the Column
   column, rename them to firstname,
   gender and activity, respectively. These columns correspond to the
   information you can extract from the sample message.

   

4. Click OK to validate these changes and
   accept the propagation prompted by the pop-up dialog box.

5. In the Bolt code area, enter the main
   method of the bolt to be executed. In this scenario, the code is as
   follows:

   String[] tokens = input.get_str().split("\|"); collector.emit(new Values( tokens[0], tokens[1], tokens[2] ));

   1 2 3 4 5 6

   String[] tokens = input.get_str().split("\|"); collector.emit(new Values( tokens[0], tokens[1], tokens[2] ));

1. Double-click tAggregateRow to open its
   Component view. This component allows
   you to find out the most popular activity recorded in the received
   messages.

   

2. Click the […] button next to Edit schema to open the schema editor.

3. On the output side (right), click the [+]
   button three times to add three rows and in the Column column, rename them to activity, gender and
   popularity, respectively.

   

4. In the Type column of the popularity row of the output side, select
   Double.

5. Click OK to validate these changes and
   accept the propagation prompted by the pop-up dialog box.

6. In the Group by table, add two rows by
   clicking the [+] button twice and configure
   these two rows as follows to group the outputted data:

   • Output column: select the
     columns from the output schema to be used as the conditions to
     group the outputted data. In this example, they are activity and gender.

   • Input column position: select
     the columns from the input schema to send data to the output
     columns you have selected in the Output
     column column. In this scenario, they are
     activity and gender.

7. In the Operations table, add one row by
   clicking the [+] button once and configure
   this row as follows to calculate the popularity of each activity:

   • Output column: select the
     column from the output schema to carry the calculated results.
     In this scenario, it is popularity.

   • Function: select the function
     to be used to process the incoming data. In this scenario,
     select count. It counts the
     frequency of each activity in the received messages.

   • Input column position: select
     the column from the input schema to provide the data to be
     processed. In this scenario, it is activity.