使用 RDD 从 CSV 文件过滤数据答案

【问题标题】：Filtering Data using RDD from CSV file使用 RDD 从 CSV 文件过滤数据
【发布时间】：2018-10-12 20:47:29
【问题描述】：

我是 Spark 的新手，正在尝试一种方法来弄清楚如何在具有多个条件的 RDD 中使用过滤器并获取记录数。场景如下：

有一个 CSV 文件有多个列标题，如 ID、日期、描述、类型、年份等。

如果我想知道 YEAR=2018 和 TYPE=comedy 有多少条记录。

我不需要使用 DataSet/DataFrame 的解决方案，只需要使用带有地图和过滤器的 RDD。

任何帮助表示赞赏。

【问题讨论】：

请用您目前尝试过的代码更新问题。
您的 RDD、Row 或其他类型是什么？

标签： csv apache-spark

【解决方案1】：

你可以这样做：

input=sc.textFile("<<path>>");



input_filtered=input.filter(lambda row : return (row.split(",")[3]=="comedy" and row.split(",")[5]=="2018")  )


input_filetered.count()

另一种选择是将 foreach 与累加器一起使用。希望它有所帮助。

【讨论】：

我认为使用 Spark 内置的 csv 支持更容易。
是的。这是真的。但他期待数据集的 RDD 瞬间。如果我们使用 csv 中内置的 spark 将返回 DataSet 并且他可以在转换为 RDD 并进行操作的那一刻进行操作。我认为他正在编写一些较旧的代码@TerryDactyl。
是的，我正在寻找的是仅使用核心 RDD 进行转换和过滤的旧方法。我尝试了类似于 @Kuldip Puri Tejaswi 提出的方法。但计数为零。我会尝试再次更新。
它将起作用...假设列顺序保持不变。

【解决方案2】：

所以如果你有 RDD[Row] 并且每一列都是 StringType 那么

myRdd.filter(r => r.getAs("YEAR") == "2018" && r.getAs("TYPE") == "COMEDY")

这真的取决于您如何读取数据。

以下是如何将文件作为 RDD[Row] 读取。

val myRdd = spark.read.format("com.databricks.spark.csv").option("header","true").load("myFile.csv").rdd

【讨论】：

【解决方案3】：

我不确定您的期望如下。

JavaRDD<String> lines = sc.textFile("readcsvyear.csv");
            JavaRDD<String> filtered = lines.filter(line -> {
                String[] list = line.split(",");
                if(list.length >= 5 && list[4].equalsIgnoreCase("2018")) {
                    return true;
                }
                return false;
            });
            System.out.println(filtered);

            for(String line:filtered.collect()){
                System.out.println("filtered "+line);
            }

输入文件

1,Date,desc,type,2018
2,Date,desc,type,2018
3,Date,desc,type,2018
4,Date,desc,type,2018
5,Date,desc,type,2018
6,Date,desc,type,2017
7,Date,desc,type,2017
8,Date,desc,type,2017

输出

filtered 1,Date,desc,type,2018
filtered 2,Date,desc,type,2018
filtered 3,Date,desc,type,2018
filtered 4,Date,desc,type,2018
filtered 5,Date,desc,type,2018

【讨论】：