import org.<a style='color:#f60; text-decoration:underline;' href="https://www.php.cn/zt/15972.html" target="_blank">apache</a>.spark.SparkConf;
import org.apache.spark.SparkContext;
import java.util.Arrays;
import java.util.List;

public class SparkSort {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("SparkSort");
        SparkContext sc = new SparkContext(conf);

        // 从 HDFS 加载数据
        List<Integer> numbers = sc.textFile("/hdfs/numbers/part*")
                                .flatMap(line -> Arrays.asList(line.split(",")))
                                .map(Integer::parseInt)
                                .collect();

        // 在集群中排序数据
        List<Integer> sortedNumbers = numbers.stream()
                                             .sorted()
                                             .toList();

        // 将结果写入 HDFS
        sc.parallelize(sortedNumbers)
          .saveAsTextFile("/hdfs/sorted_numbers");
    }
}

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;

public class FlinkSort {

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

        // 从 HDFS 加载数据
        DataSet<Integer> numbers = env.readTextFile("/hdfs/numbers/part*")
                                       .map(new MapFunction<String, Integer>() {
                                           @Override
                                           public Integer map(String line) throws Exception {
                                               return Integer.parseInt(line);
                                           }
                                       });

        // 在集群中排序数据
        DataSet<Integer> sortedNumbers = numbers.sort();

        // 将结果写入 HDFS
        sortedNumbers.writeAsText("/hdfs/sorted_numbers");

        // 执行作业
        env.execute();
    }
}