星火流textFileStream不支持通配符

Question

我設置一個簡單的測試，流從S3的文本文件，當我試圖像

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

得到它的工作，並在桶我本來的日誌文件去那裏一切都會正常工作。

但是，如果他們是一個子文件夾，它不會發現，得到了投入的子文件夾的任何文件（是的，我知道HDFS實際上並未使用的文件夾結構）

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

所以，我想簡單地做通配符像我曾與一個標準的火花應用程序之前完成

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時可以使用通配符，但似乎在進行流輸入時，它不會這樣做，也不會自動處理子文件夾中的文件。有什麼我在這裏失蹤？

最終什麼，我需要的是全天候運行streaming作業將被監視具有按日期放在它的日誌的S3桶

因此，像

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

有什麼方式將其交給最頂層的文件夾，並自動讀取顯示在任何文件夾中的文件（因爲顯然日期會每天增加）？

編輯

所以在挖掘到的文檔在http://spark.apache.org/docs/latest/streaming-programming-guide.html#basic-sources它指出嵌套目錄，不支持。

誰能說出爲什麼會出現這種情況？

另外，由於我的文件將根據日期進行嵌套，因此在流式傳輸應用程序中解決此問題的好方法是什麼？這有點複雜，因爲日誌需要幾分鐘才能寫入S3，所以即使我們在新的一天進入了幾分鐘，也可以將前一天寫入的最後一個文件寫入前一天的文件夾。

Answer 1

我們有同樣的問題。我們用逗號加入了子文件夾名稱。

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); 

我希望以這種方式解決您的問題。

Answer 2

通過擴展FileInputDStream可以創建一些「醜陋但工作的解決方案」。 寫作sc.textFileStream(d)相當於

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

您可以創建CustomFileInputDStream，將延長FileInputDStream。自定義類將從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) 

這似乎表現我們的預期。

當我寫的解決方案這樣我必須補充幾點考慮：

• 您正在打破星火封裝和創造，你將不得不完全支持隨着時間的循環中的自定義類。

• 我相信像這樣的解決方案是最後的選擇。如果你的用例可以通過不同的方式實現，通常避免這樣的解決方案會更好。

• 如果你在S3上會有很多「子目錄」，並且會檢查它們中的每一個，這會花費你。

• 理解Databricks是否僅僅因爲可能的性能損失而不支持嵌套文件會非常有趣，也許還有更深層次的原因我沒有想過。