val query = wordCounts.writeStream.trigger(ProcessingTime(5.seconds))

  .outputMode("complete")

  .foreach(new ForeachWriter[Row] {

      var fileWriter: FileWriter = _

      override def process(value: Row): Unit = {

        fileWriter.append(value.toSeq.mkString(","))

      }

      override def close(errorOrNull: Throwable): Unit = {

        fileWriter.close()

      }

      override def open(partitionId: Long, version: Long): Boolean = {

        FileUtils.forceMkdir(new File(s"/tmp/example/${partitionId}"))

        fileWriter = new FileWriter(new File(s"/tmp/example/${partitionId}/temp"))

        true

      }

    }).start()

import org.apache.spark.sql.streaming.ProcessingTime

val query =

  streamingSelectDF

    .writeStream

    .format("parquet")

    .option("path", "/mnt/sample/test-data")

    .option("checkpointLocation", "/mnt/sample/check")

    .partitionBy("zip", "day")

    .trigger(ProcessingTime("25 seconds"))

    .start()

        // Write new data to Parquet files

        // can be "orc", "json", "csv", etc.

        String hdfsFileFormat = SparkHelper.getInstance().getLTEBaseSaveHdfsFileFormat();

        String queryName = "save" + this.getTopicEncodeName(topicName) + "DataToHdfs";

        String saveHdfsPath = SparkHelper.getInstance().getLTEBaseSaveHdfsPath();

        // The file path which partitioned by scan_start_time (format:yyyyMMddHH0000)

        dsParsed.writeStream()

                .format(hdfsFileFormat)

                .option("path", saveHdfsPath + topicName + "/")

                .option("checkpointLocation", this.checkPointPath + queryName + "/")

                .outputMode("append")

                .partitionBy("scan_start_time")

                .trigger(Trigger.ProcessingTime(5, TimeUnit.MINUTES))

                .start();