工具类——Spark写入csv文件到HDFS（Java代码）

package util;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.apache.spark.api.java.JavaRDD;

import java.io.IOException;
import java.util.Map;
import java.util.UUID;

public class HdfsUtil {
    static org.apache.hadoop.conf.Configuration conf = null;

    public static org.apache.hadoop.conf.Configuration Instance() {
        if (null == conf) {
            // 创建一个配置
            conf = new Configuration();
        }
        return conf;
    }
    public static void writeRddFile(JavaRDD<String> rdd,String output) {
        //创建hdfs文件，打开Hdfs输出流
        FileSystem fs = null;
        FSDataOutputStream out = null;
        FSDataInputStream in = null;
        Configuration hdfsconf = Instance();
        //临时保存文件
        String outputPath = output + UUID.randomUUID().toString().replace("-", "") + "Temp";
        System.out.println("outputPath" + outputPath);
        rdd.saveAsTextFile(outputPath);
        try {
            fs = FileSystem.get(hdfsconf);
            Path path = new Path(outputPath);
            for (FileStatus stat : fs.listStatus(path)) {
                Path outpath = new Path(output);
                if (!fs.exists(outpath)) {
                    out = fs.create(outpath);
                } else {
                    out = fs.append(outpath);
                }
                in = new FSDataInputStream(fs.open(stat.getPath()));
                IOUtils.copyBytes(in, out, hdfsconf);
            }
            fs.delete(path, true);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (in != null) {
                    in.close();
                }
                if (out != null) {
                    out.flush();
                    out.close();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}