These properties are used to configure tKafkaInput running in the Standard Job framework.

The Standard
tKafkaInput component belongs to the Internet family.

The component in this framework is available in all Talend products with Big Data
and in Talend Data Fabric.

Basic settings

Advanced settings

Usage

These properties are used to configure tKafkaInput running in the Spark Streaming Job framework.

The Spark Streaming
tKafkaInput component belongs to the Messaging family.

This component is available in Talend Real Time Big Data Platform and Talend Data Fabric.

Basic settings

Advanced settings

Usage

• Yarn mode (Yarn client or Yarn cluster):

  • When using Google Dataproc, specify a bucket in the
    Google Storage staging bucket
    field in the Spark configuration
    tab.

  • When using HDInsight, specify the blob to be used for Job
    deployment in the Windows Azure Storage
    configuration area in the Spark
    configuration tab.

  • When using Altus, specify the S3 bucket or the Azure
    Data Lake Storage for Job deployment in the Spark
    configuration tab.
  • When using Qubole, add a
    tS3Configuration to your Job to write
    your actual business data in the S3 system with Qubole. Without
    tS3Configuration, this business data is
    written in the Qubole HDFS system and destroyed once you shut
    down your cluster.

  • When using on-premise
    distributions, use the configuration component corresponding
    to the file system your cluster is using. Typically, this
    system is HDFS and so use tHDFSConfiguration.

• Standalone mode: use the
  configuration component corresponding to the file system your cluster is
  using, such as tHDFSConfiguration or
  tS3Configuration.

  If you are using Databricks without any configuration component present
  in your Job, your business data is written directly in DBFS (Databricks
  Filesystem).

1. In the
   Integration
   perspective of the Studio, create an empty
   Spark Streaming Job from the Job Designs node
   in the Repository tree view.
   For further information about how to create a Spark Streaming Job, see
   
   Talend Open Studio for 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
   tHDFSConfiguration, tKafkaInput, tWindow, tExtractJSONFields, tMap, tAggregateRow, tTop and tLogRow.
3. Connect tKafkaInput, tWindow, tExtractJSONFields and
   tMap using the Row >
   Main link.
4. Connect tMap to tAggregateRow using the Row > Main
   link and name this connection in the dialog box that is displayed. For example,
   name it to hashtag.
5. Connect tAggregateRow, tTop and tLogRow using the
   Row > Main link.
6. Leave the tHDFSConfiguration component alone
   without any connection.

1. Click Run to open its view and then click the
   Spark Configuration tab to display its view
   for configuring the Spark connection.
2. Select the Use local mode check box to test your Job locally.

   In the local mode, the Studio builds the Spark environment in itself on the fly in order to
   run the Job in. Each processor of the local machine is used as a Spark
   worker to perform the computations.

   In this mode, your local file system is used; therefore, deactivate the
   configuration components such as tS3Configuration or
   tHDFSConfiguration that provides connection
   information to a remote file system, if you have placed these components
   in your Job.

   You can launch
   your Job without any further configuration.

3. Clear the Use local mode check box to display the
   list of the available Hadoop distributions and from this list, select
   the distribution corresponding to your Spark cluster to be used.
   This distribution could be:

   • Databricks

   • Qubole

   • Amazon EMR

     For this distribution, Talend supports:

     • Yarn client

     • Yarn cluster

   • Cloudera

     For this distribution, Talend supports:

     • Standalone

     • Yarn client

     • Yarn cluster

   • Google Cloud
     Dataproc

     For this distribution, Talend supports:

     • Yarn client

   • Hortonworks

     For this distribution, Talend supports:

     • Yarn client

     • Yarn cluster

   • MapR

     For this distribution, Talend supports:

     • Standalone

     • Yarn client

     • Yarn cluster

   • Microsoft HD
     Insight

     For this distribution, Talend supports:

     • Yarn cluster

   • Cloudera Altus

     For this distribution, Talend supports:

     • Yarn cluster

       Your Altus cluster should run on the following Cloud
       providers:

       • Azure

         The support for Altus on Azure is a technical
         preview feature.

       • AWS

   As a Job relies on Avro to move data among its components, it is recommended to set your
   cluster to use Kryo to handle the Avro types. This not only helps avoid
   this Avro known issue but also
   brings inherent preformance gains. The Spark property to be set in your
   cluster is:
   

   spark.serializer org.apache.spark.serializer.KryoSerializer

   1	spark.serializer org.apache.spark.serializer.KryoSerializer

   If you cannot find the distribution corresponding to yours from this
   drop-down list, this means the distribution you want to connect to is not officially
   supported by
   Talend
   . In this situation, you can select Custom, then select the Spark
   version of the cluster to be connected and click the
   [+] button to display the dialog box in which you can
   alternatively:

   1. Select Import from existing
      version to import an officially supported distribution as base
      and then add other required jar files which the base distribution does not
      provide.

   2. Select Import from zip to
      import the configuration zip for the custom distribution to be used. This zip
      file should contain the libraries of the different Hadoop/Spark elements and the
      index file of these libraries.

      In
      Talend
      
      Exchange, members of
      Talend
      community have shared some ready-for-use configuration zip files
      which you can download from this Hadoop configuration
      list and directly use them in your connection accordingly. However, because of
      the ongoing evolution of the different Hadoop-related projects, you might not be
      able to find the configuration zip corresponding to your distribution from this
      list; then it is recommended to use the Import from
      existing version option to take an existing distribution as base
      to add the jars required by your distribution.

      Note that custom versions are not officially supported by
      
      Talend
      .
      Talend
      and its community provide you with the opportunity to connect to
      custom versions from the Studio but cannot guarantee that the configuration of
      whichever version you choose will be easy. As such, you should only attempt to
      set up such a connection if you have sufficient Hadoop and Spark experience to
      handle any issues on your own.

   For a step-by-step example about how to connect to a custom
   distribution and share this connection, see Hortonworks.

Configuring a Spark stream for your Apache Spark streaming Job

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

   Spark uses this component to connect to the HDFS system to which the jar
   files dependent on the Job are transferred.

2. If you have defined the HDFS connection metadata under the Hadoop
   cluster node in Repository, select
   Repository from the Property
   type drop-down list and then click the
   […] button to select the HDFS connection you have
   defined from the Repository content wizard.

   For further information about setting up a reusable
   HDFS connection, search for centralizing HDFS metadata on Talend Help Center
   (https://help.talend.com).

   If you complete this step, you can skip the following steps about configuring
   tHDFSConfiguration because all the required fields
   should have been filled automatically.

3. In the Version area, select
   the Hadoop distribution you need to connect to and its version.
4. In the NameNode URI field,
   enter the location of the machine hosting the NameNode service of the cluster.
   If you are using WebHDFS, the location should be
   webhdfs://masternode:portnumber; WebHDFS with SSL is not
   supported yet.
5. In the Username field, enter
   the authentication information used to connect to the HDFS system to be used.
   Note that the user name must be the same as you have put in the Spark configuration tab.

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

2. In the Broker list field, enter the locations
   of the brokers of the Kafka cluster to be used, separating these locations using
   comma (,). In this example, only one broker
   exists and its location is localhost:9092.
3. From the Starting offset drop-down list,
   select the starting point from which the messages of a topic are consumed. In
   this scenario, select From latest, meaning to
   start from the latest message that has been consumed by the same consumer group
   and of which the offset has been committed.
4. In the Topic name field, enter the name of
   the topic from which this Job consumes Twitter streams. In this scenario, the
   topic is twitter_live.
   This topic must exist in your Kafka system. For further information about how
   to create a Kafka topic, see the documentation from Apache Kafka or use the
   tKafkaCreateTopic component provided with the
   Studio. But note that tKafkaCreateTopic is not
   available to the Spark Jobs.
5. Select the Set number of records per second to read from
   each Kafka partition check box. This limits the size of each micro
   batch to be sent for processing.

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

   This component is used to apply a Spark window on the input RDD so that this Job always
   analyzes the Tweets of the last 20 seconds at the end of each 15 seconds. This
   creates, between every two window applications, the overlap of one micro batch,
   counting 5 seconds as defined in the Batch size
   field in the Spark configuration tab.
2. In the Window duration field, enter 20000, meaning 20
   seconds.
3. Select the Define the slide duration check
   box and in the field that is displayed, enter 15000, meaning 15 seconds.

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

   As you can read from https://dev.twitter.com/overview/api/entities-in-twitter-objects#hashtags, the raw Tweet data uses the JSON format.
2. Click Sync columns to retrieve the schema
   from its preceding component. This is actually the read-only schema of tKafkaInput, since tWindow does not impact the schema.
3. Click the […] button next to Edit
   schema to open the schema editor.
   

4. Rename the single column of the output schema to hashtag. This column is used to carry the hashtag field extracted from the Tweet JSON data.
5. Click OK to validate these changes.
6. From the Read by list, select JsonPath.
7. From the JSON field list, select the column of
   the input schema from which you need to extract fields. In this scenario, it is
   payload.
8. In the Loop Jsonpath query field, enter JSON path
   pointing to the element over which extraction is looped. According to the JSON
   structure of a Tweet as you can read from the documentation of Twitter, enter
   $.entities.hashtags to loop over the
   hashtags entity.
9. In the Mapping table, in which the hashtag column of the output schema has been filled in
   automatically, enter the element on which the extraction is performed. In this
   example, this is the text attribute of each
   hashtags entity. Therefore, enter text within double quotation marks in the Json query column.

1. Double-click tMap to open its
   Map editor.
   

2. In the table representing the output flow (on the right side), enter StringHandling.DOWNCASE(row2.hashtag) in the Expression column. This automatically creates the map
   between the hashtag column of the input
   schema and the hashtag column of the output
   schema.
   Note that row2 in this expression is the ID
   of the input link to tMap. It can be labeled
   differently in the Job you are designing.
3. Click Apply to validate these changes and
   click OK to close this editor.

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

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

3. On the output side, click the [+] button two
   times to add two rows to the output schema table and rename these new schema
   columns to hashtag and count, respectively.
4. Click OK to validate these changes and accept
   the propagation prompted by the pop-up dialog box.
5. In the Group by table, add one row by clicking
   the [+] button and select hashtag for both the Output
   column column and the Input column
   position column. This passes data from the hashtag column of the input schema to the hashtag column of the output schema.
6. In the Operations table, add one row by
   clicking the [+] button.
7. In the Output column column, select count, in the Function
   column, select count and in the Input column position column, select hashtag.

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

2. In the Number of line selected field, enter the
   number of rows to be output to the next component, counting down from the first
   row of the data sorted by tTop. In this
   example, it is 5, meaning the 5 most used hashtags in each 20 seconds.
3. In the Criteria table, add one row by clicking
   the [+] button.
4. In the Schema column column, select count, the column for which the data is sorted, in
   the sort num or alpha column, select num, which means the data to be sorted are numbers,
   and in the Order asc or desc column, select
   desc to arrange the data in descending
   order.

1. Ensure that your Twitter streaming program is still running and keep writing
   the received Tweets into the given topic.
2. Press F6 to run this Job.
   

Leave the Job running a while and then in the console of the Run view, you can read the Job is listing the 5 most used hashtags in
each batch of Tweets mentioning Paris. According to the configuration of the size of
each micro batch and the Spark window, each of these Tweet batches contains the last 20
seconds’ worth of Tweets received at the end of each 15-second interval.

Note that you can manage the level of the execution information to be outputted in this
console by selecting the log4jLevel check box in the
Advanced settings tab and then selecting the level of
the information you want to display.

For more information on the log4j logging levels, see the Apache documentation at http://logging.apache.org/log4j/1.2/apidocs/org/apache/log4j/Level.html.

These properties are used to configure tKafkaInput running in the Storm Job framework.

The Storm
tKafkaInput component belongs to the Messaging family.

This component is available in Talend Real Time Big Data Platform and Talend Data Fabric.

The Storm framework is deprecated from Talend 7.1 onwards. Use Talend Jobs for Apache Spark Streaming to accomplish your Streaming related tasks.

Basic settings

Usage

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
   
   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 Open Studio for 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:

   Option/Fields	Description
   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. 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:

   Column	Description
   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:

   Column	Description
   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.

Then you can run this Job.

The tLogRow component is used to present the
execution result of the Job.

Press F6 to run this Job

Once done, the Run view is opened automatically,
where you can check the execution result.

You can read that the activity Drink is the most
popular with 3 occurrences for the gender M
(Male) and 1 occurrence for the gender F (Female)
in the messages.

The Storm topology continues to run, waiting for messages to appear on the Kafka
message broker until you kill the Job. In this scenario, because the Kill topology on quitting Talend Job check box is
selected, the Storm topology will be stopped and removed from the cluster when this
Job is stopped.