Тихий логин с техническим пользователем

Как сказал Дин, textFileStream использует по умолчанию только использование новых файлов.

  def textFileStream(directory: String): DStream[String] = {
    fileStream[LongWritable, Text, TextInputFormat](directory).map(_._2.toString)
  }

Итак, все, что он делает, вызывает этот вариант fileStream

def fileStream[
    K: ClassTag,
    V: ClassTag,
    F <: NewInputFormat[K, V]: ClassTag
  ] (directory: String): InputDStream[(K, V)] = {
    new FileInputDStream[K, V, F](this, directory)
  }

И, посмотрев на класс FileInputDStream, мы увидим, что он действительно может искать существующие файлы, но по умолчанию только новый:

newFilesOnly: Boolean = true,

Итак, возвращаясь в код StreamingContext, мы может видеть, что есть и перегрузка, которую мы можем использовать, напрямую вызывая метод fileStream:

def fileStream[
 K: ClassTag,
 V: ClassTag,
 F <: NewInputFormat[K, V]: ClassTag] 
(directory: String, filter: Path => Boolean, newFilesOnly: Boolean):InputDStream[(K, V)] = {
  new FileInputDStream[K, V, F](this, directory, filter, newFilesOnly)
}

Итак, TL; DR; является

ssc.fileStream[LongWritable, Text, TextInputFormat]
    (directory, FileInputDStream.defaultFilter, false).map(_._2.toString)