需求:
由于一个大文件,在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; } }
测试结果: