title: How to save all the output of spark sql query into a text file
body: <p>I am writing a simple consumer program using spark streaming. My code save some of the data in to the file but not ALL of the data. Can anyone help me how to fix this. I am not sure where I am losing the data. I get the data from kafka topic then I apply my schema from java Bean class. </p>
  
  <pre><code>public class ConsumerFile {
  public static void main(String[] args){
  Logger.getLogger("org").setLevel(Level.OFF);
  Logger.getLogger("akka").setLevel(Level.OFF);
  String topic = args[0];
  final String path=new String(args[2]);
  String broker = args[1];
  SparkConf sparkConf = new SparkConf().setAppName("SparkConsumer").setMaster("local[*]");;
      JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(2));
      HashSet&lt;String&gt; topicsSet = new HashSet&lt;String&gt;(Arrays.asList(topic.split(",")));
  
      HashMap&lt;String, String&gt; kafkaParams = new HashMap&lt;String, String&gt;();
  
      kafkaParams.put("metadata.broker.list", broker);
      JavaPairInputDStream&lt;String, String&gt; kafkaStream = KafkaUtils.createDirectStream(
      ssc, String.class, String.class,StringDecoder.class,StringDecoder.class,kafkaPrams,
  topicsSet
  );
  
  JavaDStream&lt;String&gt; words = kafkaStream.map(new Function&lt;Tuple2&lt;String, String&gt;, String&gt;() 
  
                                           {
                     public String call(Tuple2&lt;String, String&gt; message)
  
                                               {
                                                   return message._2();}});
  words.foreachRDD(
            new Function2&lt;JavaRDD&lt;String&gt;, Time, Void&gt;() {
         public Void call(JavaRDD&lt;String&gt; rdd, Time time) {
     SQLContext sqlContext = JavaSQLContextSingleton.getInstance(rdd.context());
                // Convert RDD[String] to RDD[case class] to DataFrame
                JavaRDD&lt;JavaRow&gt; rowRDD = rdd.map(new Function&lt;String, JavaRow&gt;() {
                  public JavaRow call(String line) throws Exception{
                  String[] fields = line.split(",");
                    JavaRow record = new JavaRow(fields[0], fields[1],fields[2]  );
  
                    return record;
  
                  }
  
                });
  
                DataFrame wordsDataFrame = sqlContext.createDataFrame(rowRDD, JavaRow.class);
                wordsDataFrame.registerTempTable("Data");
                DataFrame wDataFrame = sqlContext.sql(" select * from Data");  
                if(!wDataFrame.rdd().isEmpty()){
               wDataFrame.rdd().coalesce(1,true,null).saveAsTextFile(path); }
                return null;
              }} );
  ssc.start();
   ssc.awaitTermination();}
  </code></pre>
  
  <p>}</p>