如何在Apache Flink中为每个输入生成输出文件

Question

我正在使用Flink来处理我的流数据。

流媒体来自其他一些中间件，如Kafka、Pravega等。

说Pravega正在发送一些文字流，<代码>你好，世界，我的名字是 。

我需要的是三个过程步骤：

1. 将每个单词映射到我的自定义类对象MyJson。
2. 将对象MyJson映射到String。
3. 将字符串写入文件：一个字符串写入一个文件。

例如，对于流hello world my name is，我应该得到五个文件。

这是我的代码：

// init Pravega connector
PravegaDeserializationSchema<String> adapter = new PravegaDeserializationSchema<>(String.class, new JavaSerializer<>());
        FlinkPravegaReader<String> source = FlinkPravegaReader.<String>builder()
                .withPravegaConfig(pravegaConfig)
                .forStream(stream)
                .withDeserializationSchema(adapter)
                .build();
// map stream to MyJson
DataStream<MyJson> jsonStream = env.addSource(source).name("Pravega Stream")
            .map(new MapFunction<String, MyJson>() {
                @Override
                public MyJson map(String s) throws Exception {
                    MyJson myJson = JSON.parseObject(s, MyJson.class);
                    return myJson;
                }
            });
// map MyJson to String
DataStream<String> valueInJson = jsonStream
            .map(new MapFunction<MyJson, String>() {
                @Override
                public String map(MyJson myJson) throws Exception {
                    return myJson.toString();
                }
            });
// output
valueInJson.print();

此代码将所有结果输出到Flink日志文件。

我的问题是如何将一个单词写入一个输出文件？

Answer 1

我认为最简单的方法是使用定制水槽。

stream.addSink(new WordFileSink)

public static class WordFileSink implements SinkFunction<String> {

    @Override
    public void invoke(String value, Context context) {
        // generate a unique name for the new file and open it
        // write the word to the file
        // close the file
    }
}

请注意，此实现不一定只提供一次行为。您可能希望注意文件命名方案是唯一的和确定性的（而不是取决于处理时间），并为文件可能已经存在的情况做好准备。