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

The Spark Streaming
tMongoDBLookupInput component belongs to the Databases family.

The component in this framework is available in Talend Real Time Big Data Platform and in 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).

• The Spark cluster and the MongoDB database to be used have
  been properly installed and are running.

• The above-mentioned data has been loaded in the MongoDB
  collection to be used.

1. In the
   Integration
   perspective of the Studio, create an empty
   Spark Batch 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, tMongoDBConfiguration, tFixedFlowInput, tMongoDBOutput,
   tMongoDBLookupInput, tMap and tLogRow.
   The tFixedFlowInput components are used to
   load the data about movies into the data flow. In the real-world practice, you
   can use other components such as tFileInputDelimited instead to design a sophisticated process to
   prepare your data to be processed.
3. Connect tFixedFlowInput to tMap using the Row >
   Main link.
   This way, the main flow to tMap is created.
   The movie information is sent via this flow.
4. Connect tMongoDBLookupInput to tMap using the Row >
   Main link.
   This way, the lookup flow to tMap is created.
   The movie director information is sent via this flow.
5. Connect tMap to tMongoDBOutput using the Row > Main
   link and name this connection in the dialog box that is displayed. For example,
   name it to out1.
6. Do the same to connect tMap to tLogRow and name this connection to reject.
7. Leave tHDFSConfiguration and tMongoDBConfiguration 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 tMongoDBConfiguration to open its
   Component view.
   
   
2. From the DB Version list, select the version
   of the MongoDB database to be used.
3. In the Server field and the Port field, enter corresponding information of the
   MongoDB database.
4. In the Database field, enter the name of the
   database. This database must already exist.

1. Double-click the tFixedFlowIput component to
   open its Component view.
   

2. Click the […] button next to Edit schema to open the schema editor.
3. Click the [+] button to add the schema
   columns as shown in this image.
   

4. Click OK to validate these changes and accept
   the propagation prompted by the pop-up dialog box.
5. In the Input repetition interval field, enter
   the time interval at the end of which tFixedFlowInput sends the movie data another time. This allows you
   to generate a stream of data.
6. In the Mode area, select the Use Inline Content radio button and paste the following
   data into the Content field that is
   displayed.
   691;Dark City;1998;http://us.imdb.com/M/title-exact?imdb-title-118929;4
1654;Chairman of the Board;1998;http://us.imdb.com/Title?Chairman+of+the+Board+(1998);6
903;Afterglow;1997;http://us.imdb.com/M/title-exact?imdb-title-118566;3
255;My Best Friend's Wedding;1997;http://us.imdb.com/M/title-exact?My+Best+Friend%27s+Wedding+(1997);2
1538;All Over Me;1997;http://us.imdb.com/M/title-exact?All%20Over%20Me%20%281997%29;5
   
7. In the Field separator field, enter a
   semicolon (;).

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

2. Click the […] button next to Edit schema to open the schema editor.
3. Click the [+] button to add the schema
   columns as shown in this image.
   

4. In the Collection field, enter the name of
   the collection from which tMongoDBLookupInput
   extracts data.
5. In the Query field, enter the following
   query."{'person.id':" + row2.directorID +"}"
   
   In this statement, row2 represents the main
   flow to tMap and
   row2.directorID the directorID
   column of this flow. You need to adapt this row2 to the label of the main flow link in your Job.
   The whole statement means to select every record in which the id field within the person field has the same value as this directorID column.
   The example above shows how to use the schema of the main flow to construct
   the SQL statement to load only the matched records into the lookup flow. This
   approach ensures that no redundant records are stored in memory before being
   sent to tMap.
6. In the Mapping table, the id and the name
   columns have been automatically added. Enter, within double quotation marks,
   person in the Parent
   node path column for each row.
   This table defines how the hierarchical construct of the data from MongoDB
   should be interpreted in order to fit the schema of tMongoDBLookupInput.

1. Double-click tMap to open its
   Map Editor view.

Creating the output schema

Setting the mapping conditions

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

2. If this component does not have the same schema of the preceding
   component, a warning icon appears. In this situation, click the Sync columns button to retrieve the schema from the
   preceding one and once done, the warning icon disappears.
3. In the Collection field, enter the name of
   the collection to which you need to write data. If this collection does not
   exist, it will be automatically created at runtime.
4. From the Action on data list, select the
   operation to be performed on the data. In this example, select Insert, which creates documents in MongoDB whether these
   documents already exist or not and in either case, generates a new technical ID
   for each of the new documents.
5. Leave the Mapping table as is. This adds each
   record to the root of each document.

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

2. If this component does not have the same schema of the preceding
   component, a warning icon appears. In this situation, click the Sync columns button to retrieve the schema from the
   preceding one and once done, the warning icon disappears.
3. Select the Table radio button to present the
   result in a table.

Then you can press F6 to run this Job.

Once done, in the console of the Run view, you can
see the data rejected by inner join.

This data is displayed for several times because tFixedFlowInput has created a data stream by regularly sending out the same
records.

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.

The movie information now contains the names instead of the IDs of their directors and
the same records have been written several times in the collection but their technical
IDs (the _id field) are all distinct.