我有两个脚本,一个是用R编写的,另一个是用pyspark
编写的短小脚本,后者使用了前者的输出。为了简化操作,我想将第二个脚本的功能复制到第一个脚本中。
第二个脚本非常简单--读取一堆csv文件并将它们作为分区parquet输出:
Original Answer: "最初的回答"
spark.read.csv(path_to_csv, header = True) \
.repartition(partition_column).write \
.partitionBy(partition_column).mode('overwrite') \
.parquet(path_to_parquet)
在R中,这应该同样简单,但我无法弄清楚如何匹配SparkR
中的partitionBy
功能。到目前为止,我已经得到了以下内容:
最初的回答:
library(SparkR); library(magrittr)
read.df(path_to_csv, 'csv', header = TRUE) %>%
repartition(col = .$partition_column) %>%
write.df(path_to_parquet, 'parquet', mode = 'overwrite')
这段代码成功地为
partition_column
的每个值写了一个parquet文件。问题在于生成的文件目录结构有误;与Python生成的目录结构不同。/path/to/parquet/
partition_column=key1/
file.parquet.gz
partition_column=key2/
file.parquet.gz
...
"最初的回答":R只产生
/path/to/parquet/
file_for_key1.parquet.gz
file_for_key2.parquet.gz
...
我有点不明白,SparkR
中的partitionBy
函数似乎只涉及窗口函数的上下文,并且我在手册中没有看到其他可能相关的内容。也许可以通过...
传递参数,但是我在文档或在线搜索中都没有看到任何示例。
最初的回答:
partitionBy
参数吗?(顺便说一下,我正在使用2.3.1) - undefinedcallJMethod = SparkR:::callJMethod
、setWriteOptions
和handledCallJMethod
,这样即使不需要更新软件包也能正常工作。 - undefined