使用S3AFileSystem的Flink不會讀取S3中的子文件夾

Question

我們使用Flink 1.2.0和建議的S3AFileSystem配置。當源代碼是S3存儲桶中的單個文件夾時，一個簡單的流式作業可以按預期工作。

作業運行時沒有錯誤 - 但是不是產生輸出 - 當它的源文件夾本身包含子文件夾時。

爲了清楚起見，以下是S3存儲桶的一個模型。運行作業以指向s3a://bucket/folder/2017/04/25/01/可以正確讀取存儲桶中出現的所有三個對象和任何後續對象。將作業指向s3a://bucket/folder/2017/（或任何其他中間文件夾）會導致作業不會產生任何內容。

在絕望中，我們嘗試了[in | ex]包含尾隨/的排列組合。

. 
`-- folder 
    `-- 2017 
     `-- 04 
      |-- 25 
      | |-- 00 
      | | |-- a.txt 
      | | `-- b.txt 
      | `-- 01 
      |  |-- c.txt 
      |  |-- d.txt 
      |  `-- e.txt 
      `-- 26 

工作代碼：

def main(args: Array[String]) { 

    val parameters = ParameterTool.fromArgs(args) 
    val bucket = parameters.get("bucket") 
    val folder = parameters.get("folder") 

    val path = s"s3a://$bucket/$folder" 

    val env = StreamExecutionEnvironment.getExecutionEnvironment 

    val lines: DataStream[String] = env.readFile(
    inputFormat = new TextInputFormat(new Path(path)), 
    filePath = path, 
    watchType = FileProcessingMode.PROCESS_CONTINUOUSLY, 
    interval = Time.seconds(10).toMilliseconds) 

    lines.print() 
    env.execute("Flink Streaming Scala API Skeleton") 
} 

核心的site.xml是按每文檔配置：

<configuration> 
    <property> 
    <name>fs.s3a.impl</name> 
    <value>org.apache.hadoop.fs.s3a.S3AFileSystem</value> 
    </property> 
    <property> 
    <name>fs.s3a.buffer.dir</name> 
    <value>/tmp</value> 
    </property> 
</configuration> 

這裏已包括所有S3AFileSystem這裏列出的罐子：https://ci.apache.org/projects/flink/flink-docs-release-1.2/setup/aws.html#flink-for-hadoop-27

我們很難過。這似乎應該工作;互聯網上有很多面包屑，表明這個做了的工作。 [例如，http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Reading-files-from-an-S3-folder-td10281.html]

幫助我，同伴松鼠...你是我唯一的希望！

Answer 1

回答我自己的問題......在上面Steve Loughran的幫助下。

在弗林克，具有file-based data source to process continuously工作時，FileInputFormat不不enumerate nested files默認。

無論源是S3還是其他任何東西，情況都是如此。

您必須設置它像這樣：

def main(args: Array[String]) { 

    val parameters = ParameterTool.fromArgs(args) 
    val bucket = parameters.get("bucket") 
    val folder = parameters.get("folder") 

    val path = s"s3a://$bucket/$folder" 

    val env = StreamExecutionEnvironment.getExecutionEnvironment 

    val textInputFormat = new TextInputFormat(new Path(path)) 

    //this is important! 
    textInputFormat.setNestedFileEnumeration(true) 

    val lines: DataStream[String] = env.readFile(
    inputFormat = textInputFormat, 
    filePath = path, 
    watchType = FileProcessingMode.PROCESS_CONTINUOUSLY, 
    interval = Time.seconds(10).toMilliseconds) 

    lines.print() 
    env.execute("Flink Streaming Scala API Skeleton") 

}

Answer 2

什麼版本的Hadoop在這個下面？

如果在Hadoop 2.8中停止了，可能是迴歸，也許是我的錯。首先在FLINK下提供一個JIRA @issues.apache.org，然後，如果它在2.8.0中的新功能將它鏈接爲HADOOP-13208

這裏的代碼片段是一個很好的例子，可以用於迴歸測試，它是我曾爲Flink做過一些事情。

這個大listFiles()更改將文件的枚舉從遞歸樹形路徑移動到路徑下的所有子條目的一系列扁平列表：它非常適合其他任何事情（distcp，測試，配置單元，火花）和自16年12月以來一直在產品中出貨;如果是原因，我會有些驚訝，但不能否認責備。對不起