如何将csv文件转换为rdd
Posted
技术标签:
【中文标题】如何将csv文件转换为rdd【英文标题】:How do I convert csv file to rdd 【发布时间】:2014-06-19 05:35:27 【问题描述】:我是新来的火花。我想对 CSV 记录中的特定数据执行一些操作。
我正在尝试读取 CSV 文件并将其转换为 RDD。我的进一步操作基于 CSV 文件中提供的标题。
(来自 cmets) 到目前为止,这是我的代码:
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 文件中的每条记录。
final String[] header=heading.split(" ");
我可以像这样获取标题值。我想将此映射到 CSV 文件中的每条记录。
在 java 中,我使用 CSVReader record.getColumnValue(Column header)
来获取特定值。我需要在这里做类似的事情。
【问题讨论】:
到目前为止你尝试过什么?你被困在哪里了? “Zend me da codez”不受欢迎。是事先知道 csv 结构还是需要从 headers 中发现? 我需要从标题中发现结构。 【参考方案1】:一种简单的方法是保留标题。
假设您有一个类似的 file.csv:
user, topic, hits
om, scala, 120
daniel, spark, 80
3754978, spark, 1
我们可以定义一个使用第一行的解析版本的头类:
class SimpleCSVHeader(header:Array[String]) extends Serializable
val index = header.zipWithIndex.toMap
def apply(array:Array[String], key:String):String = array(index(key))
我们可以使用该标头来进一步处理数据:
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)
PS:欢迎使用 Scala :-)
【讨论】:
PS:我知道你是 *** 的新手。欢迎。将来,您将通过尽可能具体地提出问题来获得更好的帮助,展示您到目前为止所做的事情以及您遇到的问题。尝试显示当前问题的小代码示例。 这是非常危险的:这是一个有效的 csv 行:"a,b,c,,d,,,,,,"
拆分后它将是 ("a","b","c","","d")
,因此您将在最后一个非空列之后丢失所有空列!【参考方案2】:
您可以使用 spark-csv 库:https://github.com/databricks/spark-csv
这直接来自文档:
import org.apache.spark.sql.SQLContext
SQLContext sqlContext = new SQLContext(sc);
HashMap<String, String> options = new HashMap<String, String>();
options.put("header", "true");
options.put("path", "cars.csv");
DataFrame df = sqlContext.load("com.databricks.spark.csv", options);
【讨论】:
这有点不完美/不完整的答案:因为设置 spark-csv 包并非易事,而且 OP 是“新的”。 @javadba 我认为这是这里唯一的通用答案。此处的所有其他尝试都假定您可以天真地用逗号拆分 csv,并且仅在一些非常简单的情况下才是正确的。 @zero323 经过几次尝试,我的 spark-csv 包失败了。我编写了自己的 csv 解析器。这就是为什么我不是这个答案的忠实粉丝。显然其他人的运气更好。【参考方案3】:首先我必须说,如果将标题放在单独的文件中会简单得多——这是大数据中的惯例。
无论如何,丹尼尔的答案非常好,但是它效率低下和错误,所以我将发布我自己的。效率低下的是你不需要检查每条记录来查看它是否是标题,你只需要检查每个分区的第一条记录。错误在于,使用.split(",")
,当条目为空字符串并出现在记录的开头或结尾时,您可能会抛出异常或获取错误的列——更正您需要使用.split(",", -1)
。所以这里是完整的代码:
val header =
scala.io.Source.fromInputStream(
hadoop.fs.FileSystem.get(new java.net.URI(filename), sc.hadoopConfiguration)
.open(new hadoop.fs.Path(path)))
.getLines.head
val columnIndex = header.split(",").indexOf(columnName)
sc.textFile(path).mapPartitions(iterator =>
val head = iterator.next()
if (head == header) iterator else Iterator(head) ++ iterator
)
.map(_.split(",", -1)(columnIndex))
最后一点,如果您只想找出某些列,请考虑 Parquet。或者,如果您的行很宽,至少可以考虑实现一个惰性求值的拆分函数。
【讨论】:
【参考方案4】:我们可以使用新的 DataFrameRDD 来读写 CSV 数据。 DataFrameRDD 相对于 NormalRDD 的优势很少:
-
DataFrameRDD 比 NormalRDD 快一点,因为我们确定了架构,这有助于在运行时进行大量优化并为我们提供显着的性能提升。
即使列在 CSV 中发生变化,它也会自动获取正确的列,因为我们没有将读取数据时存在的列号硬编码为 textFile,然后将其拆分,然后使用列号来获取数据.
只需几行代码,您就可以直接读取 CSV 文件。
你需要拥有这个库:在 build.sbt 中添加它
libraryDependencies += "com.databricks" % "spark-csv_2.10" % "1.2.0"
Spark Scala 代码:
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val csvInPath = "/path/to/csv/abc.csv"
val df = sqlContext.read.format("com.databricks.spark.csv").option("header","true").load(csvInPath)
//format is for specifying the type of file you are reading
//header = true indicates that the first line is header in it
通过从中获取一些列来转换为普通 RDD
val rddData = df.map(x=>Row(x.getAs("colA")))
//Do other RDD operation on it
将 RDD 保存为 CSV 格式:
val aDf = sqlContext.createDataFrame(rddData,StructType(Array(StructField("colANew",StringType,true))))
aDF.write.format("com.databricks.spark.csv").option("header","true").save("/csvOutPath/aCSVOp")
由于标头设置为 true,我们将在所有输出文件中获取标头名称。
【讨论】:
【参考方案5】:这是另一个使用 Spark/Scala 到 convert a CSV to RDD 的示例。有关更详细的说明,请参阅此post。
def main(args: Array[String]): Unit =
val csv = sc.textFile("/path/to/your/file.csv")
// split / clean data
val headerAndRows = csv.map(line => line.split(",").map(_.trim))
// get header
val header = headerAndRows.first
// filter out header (eh. just check if the first val matches the first header name)
val data = headerAndRows.filter(_(0) != header(0))
// splits to map (header/value pairs)
val maps = data.map(splits => header.zip(splits).toMap)
// filter out the user "me"
val result = maps.filter(map => map("user") != "me")
// print result
result.foreach(println)
【讨论】:
【参考方案6】:我建议直接从驱动程序读取标题,而不是通过 Spark。造成这种情况的两个原因:1)它是单行。分布式方法没有任何优势。 2) 我们在驱动中需要这条线,而不是工作节点。
它是这样的:
// Ridiculous amount of code to read one line.
val uri = new java.net.URI(filename)
val conf = sc.hadoopConfiguration
val fs = hadoop.fs.FileSystem.get(uri, conf)
val path = new hadoop.fs.Path(filename)
val stream = fs.open(path)
val source = scala.io.Source.fromInputStream(stream)
val header = source.getLines.head
现在,当您制作 RDD 时,您可以丢弃标头。
val csvRDD = sc.textFile(filename).filter(_ != header)
然后我们可以从一列做一个RDD,例如:
val idx = header.split(",").indexOf(columnName)
val columnRDD = csvRDD.map(_.split(",")(idx))
【讨论】:
最终 JavaRDD另一种选择是使用mapPartitionsWithIndex
方法,因为您将获得分区索引号和该分区内所有行的列表。分区 0 和行 0 将是标题
val rows = sc.textFile(path)
.mapPartitionsWithIndex( (index: Int, rows: Iterator[String]) =>
val results = new ArrayBuffer[(String, Int)]
var first = true
while (rows.hasNext)
// check for first line
if (index == 0 && first)
first = false
rows.next // skip the first row
else
results += rows.next
results.toIterator
, true)
rows.flatMap row => row.split(",")
【讨论】:
【参考方案8】:这个怎么样?
val Delimeter = ","
val textFile = sc.textFile("data.csv").map(line => line.split(Delimeter))
【讨论】:
转换为文本文件后得到这样的输出 "id" "fname" "lname" "Address" "1" "xxxx" "yyyy" "xxxx" "-#-EOL-#-" " 2" "yyyy" "aaaa" "zzzz" "-#-EOL-#-" "3" "ssss" "ssss" "zzzz" "-#-EOL-#-" 所以重点是您不希望标头出现在 RDD 中,对吗?我根据这个假设添加了一个答案。 我需要一个类似于每个值的映射与标题形成这样的记录 <1 fname:xxx lname:yyyy address:zzzz>1> 【参考方案9】:对于 spark scala,我通常在无法使用 spark csv 包时使用...
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
val rawdata = sc.textFile("hdfs://example.host:8020/user/example/example.csv")
val header = rawdata.first()
val tbldata = rawdata.filter(_(0) != header(0))
【讨论】:
【参考方案10】:建议你试试
https://spark.apache.org/docs/latest/sql-programming-guide.html#rdds
JavaRDD<Person> people = sc.textFile("examples/src/main/resources/people.txt").map(
new Function<String, Person>()
public Person call(String line) throws Exception
String[] parts = line.split(",");
Person person = new Person();
person.setName(parts[0]);
person.setAge(Integer.parseInt(parts[1].trim()));
return person;
);
您必须在此示例中拥有一个具有文件头规范的类,并将您的数据与架构相关联并应用 mysql 中的标准......以获得所需的结果
【讨论】:
【参考方案11】:我认为您可以尝试将该 csv 加载到 RDD 中,然后从该 RDD 创建一个数据帧,这是从 rdd 创建数据帧的文档:http://spark.apache.org/docs/latest/sql-programming-guide.html#interoperating-with-rdds
【讨论】:
【参考方案12】:从 Spark 2.0 开始,CSV 可以直接读入 DataFrame
。
如果数据文件没有标题行,则为:
val df = spark.read.csv("file://path/to/data.csv")
这将加载数据,但为每一列提供通用名称,如 _c0
、_c1
等。
如果有标题,那么添加.option("header", "true")
将使用第一行来定义DataFrame
中的列:
val df = spark.read
.option("header", "true")
.csv("file://path/to/data.csv")
举个具体的例子,假设你有一个包含内容的文件:
user,topic,hits
om,scala,120
daniel,spark,80
3754978,spark,1
那么下面会得到按主题分组的总命中数:
import org.apache.spark.sql.functions._
import spark.implicits._
val rawData = spark.read
.option("header", "true")
.csv("file://path/to/data.csv")
// specifies the query, but does not execute it
val grouped = rawData.groupBy($"topic").agg(sum($"hits))
// runs the query, pulling the data to the master node
// can fail if the amount of data is too much to fit
// into the master node's memory!
val collected = grouped.collect
// runs the query, writing the result back out
// in this case, changing format to Parquet since that can
// be nicer to work with in Spark
grouped.write.parquet("hdfs://some/output/directory/")
// runs the query, writing the result back out
// in this case, in CSV format with a header and
// coalesced to a single file. This is easier for human
// consumption but usually much slower.
grouped.coalesce(1)
.write
.option("header", "true")
.csv("hdfs://some/output/directory/")
【讨论】:
以上是关于如何将csv文件转换为rdd的主要内容,如果未能解决你的问题,请参考以下文章
如何将具有多类的 LibSVM 文件转换为 RDD [labelPoint]