title: change data capture in spark
body: <p>I have got a requirement to do , but I am confused how to do it. 
  I have two dataframes. so first time i got the below data file1</p>
  
  <p>file1
  prodid, lastupdatedate, indicator</p>
  
  <pre><code>00001,,A
  00002,01-25-1981,A
  00003,01-26-1982,A
  00004,12-20-1985,A
  </code></pre>
  
  <p>the output should be</p>
  
  <pre><code>0001,1900-01-01, 2400-01-01, A
  0002,1981-01-25, 2400-01-01, A
  0003,1982-01-26, 2400-01-01, A
  0004,1985-12-20, 2400-10-01, A
  </code></pre>
  
  <p>Second time i got another one file2</p>
  
  <p>prodid, lastupdatedate, indicator</p>
  
  <pre><code>00002,01-25-2018,U
  00004,01-25-2018,U
  00006,01-25-2018,A
  00008,01-25-2018,A
  </code></pre>
  
  <p>I want the end result like </p>
  
  <pre><code>00001,1900-01-01,2400-01-01,A
  00002,1981-01-25,2018-01-25,I
  00002,2018-01-25,2400-01-01,A
  00003,1982-01-26,2400-01-01,A
  00004,1985-12-20,2018-01-25,I
  00004,2018-01-25,2400-01-01,A
  00006,2018-01-25,2400-01-01,A
  00008,2018-01-25,2400-01-01,A
  </code></pre>
  
  <p>so whatever the updates are there in the second file that date should come in the second column and the default date (2400-01-01) should come in the third column and the relavant indicator. The default indicator is A</p>
  
  <p>I have started like this</p>
  
  <p><div class="snippet" data-lang="js" data-hide="false" data-console="true" data-babel="false">
  <div class="snippet-code">
  <pre class="snippet-code-html lang-html prettyprint-override"><code>val spark=SparkSession.builder()
      .master("local")
      .appName("creating data frame for csv")
      .getOrCreate()
     
      import spark.implicits._ 
      val df = spark.read
    .option("header", "true")
    .option("inferSchema", "true")
    .csv("d:/prod.txt")
    
      val df1 = spark.read
    .option("header", "true")
    .option("inferSchema", "true")
    .csv("d:/prod1.txt")
    
  
  val newdf = df.na.fill("01-01-1900",Seq("lastupdatedate"))
  
  if((df1("indicator")=='U') &amp;&amp; (df1("prodid")== newdf("prodid"))){
      val df3 = df1.except(newdf)
      }</code></pre>
  </div>
  </div>
  </p>