Flink 数据流 CSV 写入器未将数据写入 CSV 文件

Question

我是 apache flink 的新手，正在尝试学习数据流。我正在从csv文件中读取有3列（姓名，主题和标记）的学生数据。我已经对标记应用了过滤器，并且只选择标记>40的记录。我正在尝试将此数据写入csv文件，但程序运行成功，csv文件仍为空。没有数据写入 csv 文件。

我

尝试使用不同的语法来编写csv文件，但没有一个对我有用。我正在通过日食在本地运行它。写入文本文件工作正常。

DataStream<String> text = env.readFile(format, params.get("input"), 
FileProcessingMode.PROCESS_CONTINUOUSLY,100);
DataStream<String> filtered = text.filter(new FilterFunction<String>(){
public boolean filter(String value) {
    String[] tokens = value.split(",");
    return Integer.parseInt(tokens[2]) >= 40;
}
});
filtered.writeAsText("testFilter",WriteMode.OVERWRITE);
DataStream<Tuple2<String, Integer>> tokenized = filtered
.map(new MapFunction<String, Tuple2<String, Integer>>(){
public Tuple2<String, Integer> map(String value) throws Exception {
    return new Tuple2("Test", Integer.valueOf(1));
}
});
tokenized.print(); 
tokenized.writeAsCsv("file:///home/Test/Desktop/output.csv", 
WriteMode.OVERWRITE, "/n", ",");
try {
env.execute();
} catch (Exception e1) {
e1.printStackTrace();
}
}
}

---

以下是我输入的 CSV 格式：

Name1,Subj1,30
Name1,Subj2,40
Name1,Subj3,40
Name1,Subj4,40

---

Tokenized.print（） 打印所有正确的记录。

Answer 1

我做了一些实验，发现这项工作工作得很好：

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class WriteCSV {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.fromElements(new Tuple2<>("abc", 1), new Tuple2<>("def", 2))
                .writeAsCsv("file:///tmp/test.csv", FileSystem.WriteMode.OVERWRITE, "n", ",");
        env.execute();
    }
}

如果我不将并行度设置为 1，则结果会有所不同。在这种情况下，test.csv 是一个包含四个文件的目录，每个文件由四个并行子任务中的一个写入。

我不确定您的情况出了什么问题，但也许您可以从此示例向后工作（假设它对您有用）。

Answer 2

您应该

在tokenized.writeAsCsv();之前删除tokenized.print();。

它将消耗数据 print(); .