title: Spark filter dataframe returns empty result
body: <p>I'm working in a project with Scala and Spark processing files that are stored in HDFS. Those files are landing in HDFS everyday in the morning. I have a job that reads that file from HDFS each day, process it and then writes the result in HDFS. After I convert the file into a Dataframe, this job executes a filter to get only the rows that contain a timestamp higher than the highest timestamp that was processed within the last file. This filter has an unknown behavior only some days. Some days works as expected and other days despite of the new file contains rows that match that filter, the filter result is empty. This happens all the times for the same file when it's executed in TEST environment but in my local works as expected using the same file with the same HDFS connection.</p>
  
  <p>I've tried to filter in different ways but none of then work in that environment for some specific files but all of then work fine in my LOCAL:
  1) Spark sql</p>
  
  <pre><code>val diff = fp.spark.sql("select * from curr " +
  s"where TO_DATE(CAST(UNIX_TIMESTAMP(substring(${updtDtCol}, 
  ${substrStart},${substrEnd}),'${dateFormat}') as TIMESTAMP))" +
  s" &gt; TO_DATE(CAST(UNIX_TIMESTAMP('${prevDate.substring(0,10)}' 
  ,'${dateFormat}') as TIMESTAMP))")
  </code></pre>
  
  <p>2) Spark filter functions</p>
  
  <pre><code>val diff = df.filter(date_format(unix_timestamp(substring(col(updtDtCol),0,10),dateFormat).cast("timestamp"),dateFormat).gt(date_format(unix_timestamp(substring(col("PrevDate"),0,10),dateFormat).cast("timestamp"),dateFormat)))
  </code></pre>
  
  <p>3) Adding extra column with the result of the filter and then filter by this new column</p>
  
  <pre><code>val test2 = df.withColumn("PrevDate", lit(prevDate.substring(0,10)))
        .withColumn("DatePre", date_format(unix_timestamp(substring(col("PrevDate"),0,10),dateFormat).cast("timestamp"),dateFormat))
        .withColumn("Result", date_format(unix_timestamp(substring(col(updtDtCol),0,10),dateFormat).cast("timestamp"),dateFormat).gt(date_format(unix_timestamp(substring(col("PrevDate"),0,10),dateFormat).cast("timestamp"),dateFormat)))
        .withColumn("x", when(date_format(unix_timestamp(substring(col(updtDtCol),0,10),dateFormat).cast("timestamp"),dateFormat).gt(date_format(unix_timestamp(substring(col("PrevDate"),0,10),dateFormat).cast("timestamp"),dateFormat)), lit(1)).otherwise(lit(0)))
  
  val diff = test2.filter("x == 1")
  </code></pre>
  
  <p>I think that the issue is not caused either by the filter itself or probably by the file but I would like to receive feedback about what should I check or if anybody has faced this before.</p>
  
  <p>Please let me know what information could be useful to post here in order to receive some feedback.</p>
  
  <p>A part of file example looks like the following:</p>
  
  <pre><code>|TIMESTAMP                 |Result|x|
  |2017-11-30-06.46.41.288395|true  |1|
  |2017-11-28-08.29.36.188395|false |0|
  </code></pre>
  
  <p>The TIMESTAMP values are compared with the previousDate (for instance: 2017-11-29) and I create a column called 'Result' with the result of that comparison that always works in both environment and also another column called 'x' with the same result.</p>
  
  <p>As I mentioned before, if I use the comparator function between both dates or the result in column 'Result' or 'x' to filter the dataframe, sometimes the result is an empty dataframe but in local using the same HDFS and file, the result contains data.</p>