将大量csv文件转换为parquet文件

yws3nbqq 于 2021-05-29 发布在 Spark

关注(0)|答案(2)|浏览(604)

我有大量的csv文件需要转换成Parquet文件，使用pyspark。一个csv是一个Parquet地板。
输入：csv文件：

000.csv
001.csv
002.csv
...

输出：qarquet文件：

000.parquet
001.parquet
002.parquet
...

我目前的解决方案是：

for each_csv in same_folder:
   df = spark.read.csv(each_csv, header = True)
   df.write.parquet(output_folder)

for循环很贵。有没有什么方法可以利用spark进行批处理？例如。
spark.read.csv（相同的文件夹/）.write.parquet（输出文件夹/）
根据quicksilver的回答，以下是我的Pypark版本：

spark = SparkSession.builder.master("local[*]").appName("csv_to_parquet").getOrCreate()

# Read csv files into a single data frame and add a column of input file names:

baseDf = spark.read.csv("input_folder/*.csv").withColumn("input_file_name", input_file_name())

# Convert file names into a list:

filePathInfo = baseDf.select("input_file_name").distinct().collect() 
filePathInfo_array = list(map(lambda row: row.input_file_name, filePathInfo))

# Write to parquet:

map(lambda csvFileName: baseDf.filter(col("input_file_name").endsWith(csvFileName)).write.mode('overwrite').parquet(f'output_folder/{csvFileName}'), filePathInfo_array)

apache-spark csv parquet

来源：https://stackoverflow.com/questions/62185753/convert-a-large-number-of-csv-files-to-parquet-files

2条答案

按热度按时间

s4chpxco1#

您可以使用globbing模式来选择文件，也可以提供文件列表。
如果文件夹中有两个文件 /tmp/file1_csv/file1.csv 以及 /tmp/file2_csv/file2.csv ，我可以使用以下

spark.read.option("header", "true").csv("/tmp/file*_csv/*.csv")

或者，如果您有奇怪的路径，也可以使用重载版本的 csv 方法。

val paths = "/dir1/,/dir2/,/dir3/"
val df = spark.read.option("header", "true").csv(paths.split(","): _*)

赞(0）回复(0）举报 2021-05-29

gjmwrych2#

您可以按照以下步骤来避免spark中的多个文件加载：，
使用源csv文件夹加载Dataframe
列 input_file_name 哪个记录源文件名
将文件名收集到一个列表中
遍历文件名列表
在文件名列表循环中，
按文件名筛选Dataframe
写入相应文件
scala中的sudo工作代码

import java.nio.file.Paths

import org.apache.spark.sql.{Encoders, SaveMode, SparkSession}
import org.apache.spark.sql.functions._

object ReadWriteToRespCsv {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder.master("local[*]").getOrCreate;

    val baseDf = spark.read.csv("src/main/resources/same_folder/*.csv")
      //Add a column `input_file_name` which records source file name
      .withColumn("input_file_name",input_file_name())

    //Collect the file names into a List
    val filePathInfo = baseDf.select("input_file_name").distinct()
      .map(row=>Paths.get(row.getString(0)).getFileName.toString)(Encoders.STRING).collect()

    //Iterate for file name list
    filePathInfo.foreach(csvFileName => {
      baseDf
        //Filter dataframe by file name
        .filter(col("input_file_name").endsWith(csvFileName) )
        .write
        .mode(SaveMode.Overwrite)
        //Write to respective file
        .parquet(s"src/main/resources/output_folder/${csvFileName}")
    })
  }

}

赞(0）回复(0）举报 2021-05-29

我来回答

将大量csv文件转换为parquet文件

2条答案

相关问题

热门标签

最新问答