Spark：java api读取hdfs目录下多个文件

216 阅读 0 评论 143 点赞

我是靠谱客的博主还单身香烟，这篇文章主要介绍Spark：java api读取hdfs目录下多个文件，现在分享给大家，希望可以做个参考。

需求：

由于一个大文件，在spark中加载性能比较差。于是把一个大文件拆分为多个小文件后上传到hdfs，然而在spark2.2下如何加载某个目录下多个文件呢?

public class SparkJob {
    public static void main(String[] args) {
        String filePath = args[0];
        // initialize spark session
        String appName = "Streaming-MRO-Load-Multiple-CSV-Files-Test";
        SparkSession sparkSession = SparkHelper.getInstance().getAndConfigureSparkSession(appName);

        // reader multiple csv files.
        try {
            Dataset<Row> rows = sparkSession.read().option("delimiter", "|").option("header", false)
                    .csv(filePath).toDF(getNCellSchema());
            rows.show(10);
        } catch (Exception ex) {
            ex.printStackTrace();
        }

        try {
            Dataset<String> rows = sparkSession.read().textFile(filePath);
            rows.show(10);
        } catch (Exception ex) {
            ex.printStackTrace();
        }

        SparkHelper.getInstance().dispose();
    }

    private static Seq<String> getNCellSchema() {
        List<String> ncellColumns = "m_id,m_eid,m_int_id,.....";

        List<String> columns = new ArrayList<String>();
        for (String column : ncellColumns) {
            columns.add(column);
        }

        Seq<String> columnsSet = JavaConversions.asScalaBuffer(columns);

        return columnsSet;
    }
}