您可以使用具有函数的自定义读取器
spark_reader
在版本中介绍
1.3.0
.
API reference
.
让我们看一个例子。假设您有2个文件:
sample1.csv
# file 1 skip line 1
# file 1 skip line 2
header1,header2,header3
row1col1,row1col2,1
row2col1,row2col2,1
row3col1,row3col2,1
sample2.csv
包含:
# file 2 skip line 1
# file 2 skip line 2
header1,header2,header3
row1col1,row1col2,2
row2col1,row2col2,2
row3col1,row3col2,2
以下代码从本地文件系统读取文件,但同样可以应用于HDFS源。
library(sparklyr)
library(dplyr)
sc <- spark_connect(master = "local")
paths <- paste("file:///",
list.files(getwd(), pattern = "sample\\d", full.names = TRUE),
sep = "")
paths
路径必须是绝对的,在我的示例中:
"file:///C:/Users/erodriguez/Documents/sample1.csv" ...
然后用数据类型定义模式。这个
custom_csv_reader
是获取URI并返回数据帧的读取器函数。读取器任务将分布在Spark worker节点上。注意
read.csv
这个调用有一个参数
skip = 2
删除前两行。
schema <- list(name1 = "character", name2 = "character", file = "integer")
custom_csv_reader <- function(path) {
read.csv(path, skip = 2, header = TRUE, stringsAsFactors = FALSE)
}
data <- spark_read(sc, path = paths, reader = custom_csv_reader, columns = schema)
data
结果:
# Source: spark<?> [?? x 3]
name1 name2 file
<chr> <chr> <int>
1 row1col1 row1col2 1
2 row2col1 row2col2 1
3 row3col1 row3col2 1
4 row1col1 row1col2 2
5 row2col1 row2col2 2
6 row3col1 row3col2 2