Let's walk through the most likely reasons your Scala Spark Streaming code is generating empty output files, along with actionable fixes for each issue:

1. textFileStream only processes newly added files (not existing ones) #

Spark Streaming's textFileStream monitors a directory specifically for files added after the streaming job starts. If you placed files in /home/xyz/folderone/ before launching your application, those files won't be picked up at all.

Fix:

• Start your Spark Streaming job first, then copy/move files into the monitored directory.
• If you need to process existing files, run a batch job first, or switch to FileStreamSource (available in newer Spark versions) with maxFilesPerTrigger to handle backlogs.

2. Overwrite mode is wiping out valid output #

You’re using .mode("overwrite") for your write operation. This means every batch—even empty ones—will delete all existing files in the output directory and write new (empty) files. If your job runs a batch where no new files are detected, it’ll overwrite previous valid output with nothing.

Fix:

• Use .mode("append") instead if you want to retain data from all batches.
• Or, add a check to only write the DataFrame if it’s not empty:

  if (!x2.isEmpty) {
    x2.toDF.write.format("text").mode("overwrite").save("file:///home/xyz/oparekta")
  }
  

3. Incorrect accumulator usage (and premature println) #

Your accumulator cnt has two critical issues:

• You’re calling println(s"value of count ${cnt.value}") before starting the streaming context. This line runs once during initialization—long before any data is processed—so it will always print 0. This doesn’t mean no data was handled; you’re just checking the value at the wrong time.
• Accumulators in Spark Streaming need checkpointing to guarantee consistent updates. Without it, failed batches might re-apply accumulator updates, leading to incorrect counts.

Fix:

• Move the accumulator log inside the foreachRDD block to see counts after each batch:

  x.foreachRDD{ rddx =>
    val batchCount = rddx.count()
    cnt.add(batchCount)
    println(s"Total records processed so far: ${cnt.value}")
    
    if (batchCount > 0) {
      rddx.toDF.write.format("text").mode("append").save("file:///home/xyz/oparekta")
    }
  }
  

• Add checkpointing to ensure accumulator consistency:

  ssc.checkpoint("/home/xyz/spark-checkpoint")
  

4. File permission or path issues #

Double-check that:

• The Spark process has read permissions on /home/xyz/folderone/ and write permissions on /home/xyz/oparekta/.
• Your file paths are free of typos (e.g., folderone vs folderOne can cause silent failures).
• If running in a cluster, use shared storage paths (like HDFS) instead of local file:/// paths so all workers can access the directories.

5. Lazy evaluation debug check #

Spark uses lazy evaluation, so while your toDF.write action should trigger processing, it’s worth verifying if the RDD actually contains data in each batch.

Fix:
Add a debug log inside foreachRDD to confirm batch size:

x.foreachRDD{ rddx =>
  val batchSize = rddx.count()
  println(s"Received $batchSize records in this batch")
  // rest of your code
}

内容的提问来源于stack exchange，提问作者suba