Spark Streaming textFileStream 不支持通配符

Question

我设置了一个简单的测试来从 S3 流式传输文本文件，并在我尝试类似的东西时让它工作

val input = ssc.textFileStream("s3n://mybucket/2015/04/03/")

在存储桶中，我会将日志文件放入其中，一切都会正常工作。

但如果它们是子文件夹，它不会找到放入子文件夹的任何文件（是的，我知道 hdfs 实际上并不使用文件夹结构）

val input = ssc.textFileStream("s3n://mybucket/2015/04/")

所以，我尝试像以前使用标准 spark 应用程序那样简单地使用通配符

val input = ssc.textFileStream("s3n://mybucket/2015/04/*")

但是当我尝试这个时它会抛出一个错误

java.io.FileNotFoundException: File s3n://mybucket/2015/04/* does not exist.
at org.apache.hadoop.fs.s3native.NativeS3FileSystem.listStatus(NativeS3FileSystem.java:506)
at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:1483)
at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:1523)
at org.apache.spark.streaming.dstream.FileInputDStream.findNewFiles(FileInputDStream.scala:176)
at org.apache.spark.streaming.dstream.FileInputDStream.compute(FileInputDStream.scala:134)
at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:300)
at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:300)
at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:299)
at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:287)
at scala.Option.orElse(Option.scala:257)
.....

我知道您可以在为标准 Spark 应用程序读取 fileInput 时使用通配符，但似乎在进行流式输入时，它不会这样做，也不会自动处理子文件夹中的文件。我这里有什么遗漏吗？？

我最终需要的是一个 24/7 全天候运行的流式作业，它将监控按日期放置日志的 S3 存储桶

类似

s3n://mybucket/<YEAR>/<MONTH>/<DAY>/<LogfileName>

有没有办法把它放在最顶层的文件夹中，它会自动读取出现在任何文件夹中的文件（因为显然日期每天都会增加）？

编辑

因此，在深入研究http://spark.apache.org/docs/latest/streaming-programming-guide.html#basic-sources 的文档时，它指出不支持嵌套目录。

谁能解释一下为什么会这样？

另外，由于我的文件将根据它们的日期嵌套，在我的流应用程序中解决这个问题的好方法是什么？这有点复杂，因为日志需要几分钟才能写入 S3，因此当天写入的最后一个文件可能会写入前一天的文件夹中，即使我们距离新的一天还有几分钟。

Answer 1

通过扩展 FileInputDStream 可以创建一些“丑陋但有效的解决方案”。 写sc.textFileStream(d)相当于

new FileInputDStream[LongWritable, Text, TextInputFormat](streamingContext, d).map(_._2.toString)

您可以创建将扩展 FileInputDStream 的 CustomFileInputDStream。自定义类将从 FileInputDStream 类复制计算方法，并根据需要调整 findNewFiles 方法。

从以下位置更改 findNewFiles 方法：

 private def findNewFiles(currentTime: Long): Array[String] = {
    try {
      lastNewFileFindingTime = clock.getTimeMillis()

  // Calculate ignore threshold
  val modTimeIgnoreThreshold = math.max(
    initialModTimeIgnoreThreshold,   // initial threshold based on newFilesOnly setting
    currentTime - durationToRemember.milliseconds  // trailing end of the remember window
  )
  logDebug(s"Getting new files for time $currentTime, " +
    s"ignoring files older than $modTimeIgnoreThreshold")
  val filter = new PathFilter {
    def accept(path: Path): Boolean = isNewFile(path, currentTime, modTimeIgnoreThreshold)
  }
  val newFiles = fs.listStatus(directoryPath, filter).map(_.getPath.toString)
  val timeTaken = clock.getTimeMillis() - lastNewFileFindingTime
  logInfo("Finding new files took " + timeTaken + " ms")
  logDebug("# cached file times = " + fileToModTime.size)
  if (timeTaken > slideDuration.milliseconds) {
    logWarning(
      "Time taken to find new files exceeds the batch size. " +
        "Consider increasing the batch size or reducing the number of " +
        "files in the monitored directory."
    )
  }
  newFiles
} catch {
  case e: Exception =>
    logWarning("Error finding new files", e)
    reset()
    Array.empty
}

}

到：

  private def findNewFiles(currentTime: Long): Array[String] = {
    try {
      lastNewFileFindingTime = clock.getTimeMillis()

      // Calculate ignore threshold
      val modTimeIgnoreThreshold = math.max(
        initialModTimeIgnoreThreshold,   // initial threshold based on newFilesOnly setting
        currentTime - durationToRemember.milliseconds  // trailing end of the remember window
      )
      logDebug(s"Getting new files for time $currentTime, " +
        s"ignoring files older than $modTimeIgnoreThreshold")
      val filter = new PathFilter {
        def accept(path: Path): Boolean = isNewFile(path, currentTime, modTimeIgnoreThreshold)
      }
      val directories = fs.listStatus(directoryPath).filter(_.isDirectory)
      val newFiles = ArrayBuffer[FileStatus]()

      directories.foreach(directory => newFiles.append(fs.listStatus(directory.getPath, filter) : _*))

      val timeTaken = clock.getTimeMillis() - lastNewFileFindingTime
      logInfo("Finding new files took " + timeTaken + " ms")
      logDebug("# cached file times = " + fileToModTime.size)
      if (timeTaken > slideDuration.milliseconds) {
        logWarning(
          "Time taken to find new files exceeds the batch size. " +
            "Consider increasing the batch size or reducing the number of " +
            "files in the monitored directory."
        )
      }
      newFiles.map(_.getPath.toString).toArray
    } catch {
      case e: Exception =>
        logWarning("Error finding new files", e)
        reset()
        Array.empty
    }
  }

将检查所有一级子文件夹中的文件，您可以将其调整为使用批处理时间戳来访问相关的“子目录”。

如前所述，我创建了 CustomFileInputDStream 并通过调用激活它：

new CustomFileInputDStream[LongWritable, Text, TextInputFormat](streamingContext, d).map(_._2.toString)

这似乎符合我们的预期。

当我写出这样的解决方案时，我必须添加一些考虑因素：

• 您打破了 Spark 封装并创建了一个自定义类，随着时间的推移您将不得不单独支持该类。

• 我相信这样的解决方案是最后的手段。如果您的用例可以通过不同的方式实现，通常最好避免这样的解决方案。

• 如果您将在 S3 上有很多“子目录”并且要检查每个子目录，那么您将付出代价。

• 如果 Databricks 不支持嵌套文件只是因为可能的性能损失，这将是非常有趣的，也许还有更深层次的原因我没有想到。

Answer 2

我们有同样的问题。我们用逗号连接了子文件夹名称。

List<String> paths = new ArrayList<>();
SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/dd");

try {        
    Date start = sdf.parse("2015/02/01");
    Date end = sdf.parse("2015/04/01");

    Calendar calendar = Calendar.getInstance();
    calendar.setTime(start);        

    while (calendar.getTime().before(end)) {
        paths.add("s3n://mybucket/" + sdf.format(calendar.getTime()));
        calendar.add(Calendar.DATE, 1);
    }                
} catch (ParseException e) {
    e.printStackTrace();
}

String joinedPaths = StringUtils.join(",", paths.toArray(new String[paths.size()]));
val input = ssc.textFileStream(joinedPaths);

希望通过这种方式解决你的问题。