如何使用 Flink Java API 读取目录(本地文件系统/HDFS)下的文件名

Question

我是 Flink 的新手。实际上，我正在尝试通过flink java api读取文件和csv转换。

根据我们的要求。a( 需要传递文件夹作为输入参数，输出参数作为csv文件名b( 需要从本地文件系统/HDFS 读取文件c( 将相同的数据写入 csv

我的代码：

public class WriteToCSV {
    public static void main(String[] args) throws Exception {
        final ParameterTool params = ParameterTool.fromArgs(args);
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().setGlobalJobParameters(params);
        List<String> paths = new ArrayList<String>();
        File dir = new File("C://");
        for (File f : dir.listFiles()) {
              paths.add(f.getName());
        }
        DataSet<String> data = env.fromCollection(paths).rebalance();
        DataSet<Tuple2<String, Integer>> counts = 
                    // split up the lines in pairs (2-tuples) containing: (word,1)
                    data.flatMap(new MySplitter()).groupBy(0).sum(1);
        System.out.println(" data -:"+data);
        data.print();
        counts.writeAsCsv("C://new.csv");
    }
}

class MySplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
        // normalize and split the line into words
        String[] tokens = value.split("\W+");
        // emit the pairs
        for (String token : tokens) {
            if (token.length() > 0) {
                out.collect(new Tuple2<String, Integer>(token, 1));
            }
        }
    }
}

我能够获取文件名(data.print(((。 但是csv没有创建，服务器控制台中没有异常。

Answer 1

代码中没有向 csv 写入任何内容的原因是您在counts.writeAsCsv("C://new.csv");之后不会调用 env.execute()

要进一步改进代码，您可以使用 env.readTextFile(path) 它接受目录的路径并读取该目录中的所有文件，为每一行生成记录。