java - 从通过读取csv创建的spark数据集中删除第一行

Question

This question already has answers here:

How do I skip a header from CSV files in Spark?
                            
                                （12个答案）
                            
                    
                2年前关闭。
        

    

我有一个csv（employee.csv）文件，如下所示：

20180011,20180011123,007,07
Employee_ID,Name,Country
1,Maddy,IND
2,Sun,US

现在，我正在使用spark读取csv文件，如下所示：

Dataset<Row> dataset = spark.read().format("csv")
                        .option("header", "false")
                        .load("./employee.csv");

现在，我需要删除CSV文件的第一行 20180011,20180011123,007,07，并将其加载到带有CSV头文件的数据集中

Employee_ID,Name,Country
1,Maddy,IND
2,Sun,US

有人可以帮我吗？

最佳答案

此代码首先过滤垃圾行，然后提取标头，然后转换为Dataframe。

    val ss = SparkSession.builder().appName("local").master("local[*]").getOrCreate()

    val path = "C:\\Users\\user1\\data.txt"

    val data = ss.sparkContext.textFile(path)
    val junk = data.first()

    val fdata = data.filter(x => x != junk) // removes the first line

    val header = fdata.filter(x => x.split(",")(1) == "Name").collect().mkString // filtering the header line.

    import ss.implicits._
    val df = fdata
        .filter(x => x.split(",")(1) != "Name") // filtering all except header line
        .map(x => x.split(","))
        .map(t => (t(0), t(1), t(2))) //splitting data to tuples
        .toDF(header.split(","): _*) // applying header string as column names
    df.show()

输出：

+-----------+-----+-------+
|Employee_ID| Name|Country|
+-----------+-----+-------+
|          1|Maddy|    IND|
|          2|  Sun|     US|
+-----------+-----+-------+

关于java - 从通过读取csv创建的spark数据集中删除第一行，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/52069828/

上一篇：java - 在不同监听器之间切换

下一篇：java - Spring Boot批量读取/验证不同格式的多个csv文件

Answer 1

此代码首先过滤垃圾行，然后提取标头，然后转换为Dataframe。

    val ss = SparkSession.builder().appName("local").master("local[*]").getOrCreate()

    val path = "C:\\Users\\user1\\data.txt"

    val data = ss.sparkContext.textFile(path)
    val junk = data.first()

    val fdata = data.filter(x => x != junk) // removes the first line

    val header = fdata.filter(x => x.split(",")(1) == "Name").collect().mkString // filtering the header line.

    import ss.implicits._
    val df = fdata
        .filter(x => x.split(",")(1) != "Name") // filtering all except header line
        .map(x => x.split(","))
        .map(t => (t(0), t(1), t(2))) //splitting data to tuples
        .toDF(header.split(","): _*) // applying header string as column names
    df.show()

输出：

+-----------+-----+-------+
|Employee_ID| Name|Country|
+-----------+-----+-------+
|          1|Maddy|    IND|
|          2|  Sun|     US|
+-----------+-----+-------+