EMR上的Flink-没有输出，无论是控制台还是文件

Question

object StreamingJob1 {
    def main(args: Array[String]) {
        val path = args(0)
        val file_input_format = new TextInputFormat(
            new org.apache.flink.core.fs.Path(path))
        file_input_format.setFilesFilter(FilePathFilter.createDefaultFilter())
        file_input_format.setNestedFileEnumeration(true)

        val env = StreamExecutionEnvironment.getExecutionEnvironment
        val myStream: DataStream[String] =
            env.readFile(file_input_format,
                path,
                FileProcessingMode.PROCESS_CONTINUOUSLY,
                1000L)
                .map(s => s.split(",").toString)

        myStream.print()
        // execute program
        env.execute("Flink Streaming Scala")
    }
}

我试图输出到一个Kinesis流，或一个S3文件。什么都没有记录。

    myStream.addSink(new BucketingSink[String](output_path))

我还尝试写到一个HDFS文件。在本例中，创建了一个文件，但大小为0。我确信输入文件已经使用一个简单的检查进行了处理：

myStream.map(s => {"abc".toInt})

它生成了一个异常。

我在这里漏掉了什么？

Answer 1

看起来stream.print（）在EMR上不起作用。

输出到文件：使用HDFS，有时（或大部分时间）我需要等待文件更新。

对Kinesis的输出：我的流名中有一个错别字。我不知道为什么我没有得到任何关于流不存在的异常。然而，在得到名字更正后，我得到了我期望的消息。