Apache Flink提供了一个非常方便的功能，即可以按键（分组）将数据转储到多个文件中。以下是一个解决该问题的示例代码：

首先，您需要导入必要的Flink库和类：

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.util.Collector;

然后，您可以编写一个自定义的Map函数，该函数将数据按键（分组）并将其写入多个文件中：

public static class KeyedDataWriter extends RichMapFunction, String> {
    private transient FileSystem fs;
    private transient int subtaskIndex;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        fs = FileSystem.get(getRuntimeContext().getTaskManagerInfo().getConfiguration());
        subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();
    }

    @Override
    public String map(Tuple2 value) throws Exception {
        // 按键（分组）将数据写入多个文件
        String outputPath = "file:///path/to/output/directory/" + subtaskIndex + ".txt";
        fs.create(new Path(outputPath), FileSystem.WriteMode.OVERWRITE).write(value.f0.getBytes());

        return value.f0;
    }
}

在上述代码中，我们使用open()方法来获取Flink的FileSystem实例以及当前任务的子任务索引。然后，在map()方法中，我们根据子任务索引为每个子任务生成一个文件路径，并使用FileSystem实例将数据写入文件。

最后，在主程序中，您可以使用keyBy()函数按键（分组）数据，并将其传递给自定义的Map函数进行处理：

public static void main(String[] args) throws Exception {
    final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

    // 从数据源读取数据
    DataSource> input = env.fromElements(
            new Tuple2<>("A", 1),
            new Tuple2<>("B", 2),
            new Tuple2<>("A", 3),
            new Tuple2<>("B", 4));

    // 按键（分组）数据并将其转储到多个文件中
    input.keyBy(0)
            .map(new KeyedDataWriter())
            .setParallelism(2) // 设置并行度为2
            .print();

    // 执行作业
    env.execute("Keyed Data Writer");
}

在上述代码中，我们使用keyBy(0)函数按键（分组）数据，并将其传递给自定义的KeyedDataWriter函数进行处理。然后，我们使用setParallelism(2)函数设置并行度为2，以便在两个并行任务中执行数据写入操作。

最后，我们使用print()函数将结果打印到控制台，并使用env.execute()函数执行作业。

请注意，上述代码中的文件路径是示例路径，您需要根据实际情况进行更改。此外，您还需要确保Flink集群上存在指定的输出目录，并且具有适当的访问权限。