title: search replace using Apache spark java
body: <p><strong>Problem Statement:</strong></p>
  
  <p>We need to replace the Synonyms of the words in a row with its equivalent words (from large collection of synonym list ~40000+ Key Value pairs) on a large dataset(50000 rows).</p>
  
  <p><strong>Example:</strong><br>
      Input</p>
  
  <pre><code>Allen jeevi pramod Allen Armstrong
  sandesh Armstrong jeevi
  harsha Nischay DeWALT
  </code></pre>
  
  <p><strong>Synonym list (key value pair)</strong>
  //We have 40000 entries</p>
  
  <pre><code>Key         |   Value 
  ------------------------------------
  Allen       |   Apex Tool Group
  Armstrong   |   Columbus McKinnon
  DeWALT      |   StanleyBlack
  </code></pre>
  
  <p>Above Synonym list has to be used on the Input and the Output should be as shown in the below format.</p>
  
  <pre><code>Expected Output
  
  Apex Tool Group jeevi pramod Apex Tool Group Columbus McKinnon
  sandesh Columbus McKinnon jeevi
  harsha Nischay StanleyBlack
  </code></pre>
  
  <p>We have tried with 3 approaches all of them has its own limitations</p>
  
  <p><strong>Approach 1</strong></p>
  
  <p><em>Using UDF</em></p>
  
  <pre><code>    public void test () {
              List&lt;Row&gt; data = Arrays.asList(
              RowFactory.create(0, "Allen jeevi pramod Allen Armstrong"),
              RowFactory.create(1, "sandesh Armstrong jeevi"),
              RowFactory.create(2, "harsha Nischay DeWALT")
         );
  
          StructType schema = new StructType(new StructField[] {
              new StructField("label", DataTypes.IntegerType, false, Metadata.empty()),
              new StructField("sentence", DataTypes.StringType, false, Metadata.empty()) 
         });
          Dataset&lt;Row&gt; sentenceDataFrame = spark.createDataFrame(data, schema);
  
          List&lt;Row&gt; data2 = Arrays.asList(
              RowFactory.create("Allen", "Apex Tool Group"),
              RowFactory.create("Armstrong","Columbus McKinnon"),
              RowFactory.create("DeWALT","StanleyBlack")
          );
  
          StructType schema2 = new StructType(new StructField[] {
              new StructField("label2", DataTypes.StringType, false, Metadata.empty()),
              new StructField("sentence2", DataTypes.StringType, false, Metadata.empty()) 
          });
          Dataset&lt;Row&gt; sentenceDataFrame2 = spark.createDataFrame(data2, schema2);
  
          UDF2&lt;String, String, Boolean&gt; contains = new UDF2&lt;String, String, Boolean&gt;() 
          {
              private static final long serialVersionUID = -5239951370238629896L;
  
              @Override
              public Boolean call(String t1, String t2) throws Exception {
                  return t1.contains(t2);
              }
          };
          spark.udf().register("contains", contains, DataTypes.BooleanType);
  
          UDF3&lt;String, String, String, String&gt; replaceWithTerm = new UDF3&lt;String, 
          String, String, String&gt;() {
              private static final long serialVersionUID = -2882956931420910207L;
  
          @Override
          public String call(String t1, String t2, String t3) throws Exception {
              return t1.replaceAll(t2, t3);
          }
      };
  
      spark.udf().register("replaceWithTerm", replaceWithTerm, DataTypes.StringType);
  
  Dataset&lt;Row&gt; joined = sentenceDataFrame.join(sentenceDataFrame2, callUDF("contains", sentenceDataFrame.col("sentence"), sentenceDataFrame2.col("label2")))
                                         .withColumn("sentence_replaced", callUDF("replaceWithTerm", sentenceDataFrame.col("sentence"), sentenceDataFrame2.col("label2"), sentenceDataFrame2.col("sentence2")))
                                         .select(col("sentence_replaced"));
  
  joined.show(false);
  }
  </code></pre>
  
  <p>`</p>
  
  <pre><code>Input
      Allen jeevi pramod Allen Armstrong
      sandesh Armstrong jeevi
      harsha Nischay DeWALT
  
  Expected Output
      Apex Tool Group jeevi pramod Apex Tool Group Columbus McKinnon
      sandesh Columbus McKinnon jeevi
      harsha Nischay StanleyBlack
  
  Actual Output
      Apex Tool Group jeevi pramod Apex Tool Group Armstrong
      Allen jeevi pramod Allen Columbus McKinnon
      sandesh Columbus McKinnon jeevi
      harsha Nischay StanleyBlack
  </code></pre>
  
  <p><em>Issue with approach 1, if there are multiple synonym keys in the input dataset, that many rows are getting created as shown in the above example output. 
  Expected only one row with all the replacement</em></p>
  
  <p><strong>Approach 2.</strong>
  Using ImmutableMap with replace function: Here we kept key and values pair in hashmap within ImmutableMap function, we called replace function to replace all the things
   but if a row contains multiple keys then it ignores complete row without replacing single key…</p>
  
  <pre><code>try {
  
          JavaSparkContext sc = new JavaSparkContext(new SparkConf().setAppName("SparkJdbcDs").setMaster("local[*]"));
          SQLContext sqlContext = new SQLContext(sc);
          SparkSession spark = SparkSession.builder()
                  .appName("JavaTokenizerExample").getOrCreate();
  
          HashMap&lt;String, String&gt; options = new HashMap&lt;String, String&gt;();
          options.put("header", "true");
          Dataset&lt;Row&gt; dataFileContent = sqlContext.load("com.databricks.spark.csv", options);
          dataFileContent=dataFileContent.withColumn("ManufacturerSource", regexp_replace(col("ManufacturerSource"),"[^a-zA-Z0-9\\s+]",""));
          dataFileContent= dataFileContent.na().replace("ManufacturerSource",ImmutableMap.&lt;String, String&gt;builder()
              .put("Allen", "Apex Tool Group"),
              .put("Armstrong","Columbus McKinnon"),
              .put("DeWALT","StanleyBlack")
              //Here we have 40000 entries
              .build()
  
            );
            dataFileContent.show(10,false);
  
      } catch (Exception e) {
          e.printStackTrace();
      }
  </code></pre>
  
  <p>Here is the sample code and output:</p>
  
  <pre><code>Input
      Allen jeevi pramod Allen Armstrong
      sandesh Armstrong jeevi
      harsha Nischay DeWALT
  
  Expected Output
      Apex Tool Group jeevi pramod Apex Tool Group Columbus McKinnon
      sandesh Columbus McKinnon jeevi
      harsha Nischay StanleyBlack
  
  Actual Output
      Allen jeevi pramod Allen Armstrong
      sandesh Columbus McKinnon jeevi
      harsha Nischay StanleyBlack
  </code></pre>
  
  <p><strong>Approach 3</strong></p>
  
  <p><em>Using replace all within UDF</em></p>
  
  <pre><code>public static void main(String[] args) {
            JavaSparkContext sc = new JavaSparkContext(new SparkConf().setAppName("JoinFunctions").setMaster("local[*]"));
            SQLContext sqlContext = new SQLContext(sc);
            SparkSession spark = SparkSession.builder().appName("StringSimiliarityExample").getOrCreate();
  
  
              Dataset&lt;Row&gt; sourceFileContent = sqlContext.read()
                          .format("com.databricks.spark.csv")
                          .option("header", "true")
                          .load("source100.csv");
              sourceFileContent.show(false);
  
          StructType schema = new StructType(new StructField[] {
          new StructField("label", DataTypes.IntegerType, false,
                  Metadata.empty()),
          new StructField("sentence", DataTypes.StringType, false,
                  Metadata.empty()) });
          Dataset&lt;Row&gt; sentenceDataFrame = spark.createDataFrame(data, schema);
          UDF1 mode = new UDF1&lt;String, String&gt;() {
              public String call(final String types) throws Exception {
                  return types.replaceAll("Allen", "Apex Tool Group")
                  .replaceAll("Armstrong","Columbus McKinnon")
                  .replaceAll("DeWALT","StanleyBlack")
                  //40000 more entries.....
              }
          };
  
          sqlContext.udf().register("mode", mode, DataTypes.StringType);
  
          sentenceDataFrame.createOrReplaceTempView("people");
          Dataset&lt;Row&gt; newDF = sqlContext.sql("SELECT mode(sentence), label FROM people").withColumnRenamed("UDF(sentence)", "sentence");
          newDF.show(false);
      }
  </code></pre>
  
  <p><strong>Output</strong>
  Stackoverflow exception.</p>
  
  <p>Here, we are getting stackoverflow exception. Because it resembles recursive function call.</p>
  
  <p>Kindly, let us know if there are any other innovative approaches that can help to resolve this issue.</p>