在Flink中，你可以通过以下方法为join操作设置并行度：
- 使用ExecutionEnvironment的setParallelism方法。首先，你需要创建一个ExecutionEnvironment实例，然后调用setParallelism方法来设置并行度。例如：
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(parallelism);
- 使用StreamExecutionEnvironment的createInputFormat方法。这个方法允许你根据输入格式来设置并行度。例如：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.createInputFormat(MyInputFormat.class, MyPOJO.class).setParallelism(parallelism);
- 使用DataStream的assignTimestampsAndWatermarks方法和transform方法。这两个方法都允许你在转换操作中设置并行度。例如：
DataStream<MyPOJO> dataStream = env.fromElements(...);
dataStream.assignTimestampsAndWatermarks(new MyTimestampAssigner()).setParallelism(parallelism);
dataStream.transform("Join", TypeInformation.of(MyPOJO.class), new MyJoinFunction()).setParallelism(parallelism);
注意：这些方法设置的并行度只对当前操作有效，不会影响到其他操作。如果你希望在整个Flink作业中使用相同的并行度，可以在创建ExecutionEnvironment时设置全局并行度。
这个问题可能是由于Flink的savepoint机制和Doris的写入机制之间的冲突导致的。Flink的savepoint会暂停所有的task，等待所有正在处理的数据都被写入到目标系统（如Doris），然后再进行checkpoint。但是，如果Doris的写入机制不支持这种暂停-恢复的操作，那么在Flink尝试从savepoint恢复的时候，可能会出现错误。
解决这个问题的一种可能的方法是，在Flink的sink connector中实现自定义的sink功能，这个自定义的sink可以在Flink尝试进行savepoint的时候，将Doris的写入操作也暂停，然后在Flink从savepoint恢复的时候，再将Doris的写入操作也恢复。
另一种可能的解决方案是，修改Flink的savepoint机制，使其在尝试进行savepoint的时候，不要求所有的数据都必须被写入到目标系统，而是只要求所有的数据都已经到达了Flink的sink connector，这样即使Doris的写入操作没有被暂停，Flink也可以进行savepoint。
这两种解决方案都需要对Flink和Doris的内部机制有深入的理解，可能需要一定的开发工作。
