如何将 csv 文件转换为 rdd [英] How do I convert csv file to rdd
问题描述
我是新来的火花.我想对 CSV 记录中的特定数据执行一些操作.
I'm new to spark. I want to perform some operations on particular data in a CSV record.
我正在尝试读取 CSV 文件并将其转换为 RDD.我的进一步操作基于 CSV 文件中提供的标题.
I'm trying to read a CSV file and convert it to RDD. My further operations are based on the heading provided in CSV file.
(来自评论)到目前为止,这是我的代码:
(From comments) This is my code so far:
final JavaRDD<String> File = sc.textFile(Filename).cache();
final JavaRDD<String> lines = File.flatMap(new FlatMapFunction<String, String>() {
@Override public Iterable<String> call(String s) {
return Arrays.asList(EOL.split(s));
}
});
final String heading=lines.first().toString();
我可以得到这样的标头值.我想将此映射到 CSV 文件中的每条记录.
I can get the header values like this. I want to map this to each record in CSV file.
final String[] header=heading.split(" ");
我可以得到这样的标头值.我想将此映射到 CSV 文件中的每条记录.
I can get the header values like this. I want to map this to each record in CSV file.
在 java 中,我使用 CSVReader record.getColumnValue(Column header)
来获取特定值.我需要在这里做类似的事情.
In java I’m using CSVReader record.getColumnValue(Column header)
to get the particular value. I need to do something similar to that here.
推荐答案
一种简单的方法是保留标题.
A simplistic approach would be to have a way to preserve the header.
假设您有一个 file.csv,例如:
Let's say you have a file.csv like:
user, topic, hits
om, scala, 120
daniel, spark, 80
3754978, spark, 1
我们可以定义一个使用第一行解析版本的头类:
We can define a header class that uses a parsed version of the first row:
class SimpleCSVHeader(header:Array[String]) extends Serializable {
val index = header.zipWithIndex.toMap
def apply(array:Array[String], key:String):String = array(index(key))
}
我们可以使用该标头来进一步处理数据:
That we can use that header to address the data further down the road:
val csv = sc.textFile("file.csv") // original file
val data = csv.map(line => line.split(",").map(elem => elem.trim)) //lines in rows
val header = new SimpleCSVHeader(data.take(1)(0)) // we build our header with the first line
val rows = data.filter(line => header(line,"user") != "user") // filter the header out
val users = rows.map(row => header(row,"user")
val usersByHits = rows.map(row => header(row,"user") -> header(row,"hits").toInt)
...
请注意,header
只不过是一个简单的助记符到数组索引的映射.几乎所有这些都可以在数组中元素的序数位置完成,例如 user = row(0)
Note that the header
is not much more than a simple map of a mnemonic to the array index. Pretty much all this could be done on the ordinal place of the element in the array, like user = row(0)
PS:欢迎使用 Scala :-)
PS: Welcome to Scala :-)
这篇关于如何将 csv 文件转换为 rdd的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!