python - 如何在HDFS内将CSV转换为Parquet文件

Question

我是Big Data的新手，因此Hadoop和hdfs现在对我来说已经消失了，因此我寻求帮助。
现在我有4个csv格式的文件，它们位于HDFS集群中，我应该使用PARQUET将它们制作成4个Python格式的副本，但我不知道该如何制作。
希望您能帮助我解决这个难题。

Answer 1

我将您的示例放在Scala代码中，但是在Python中进行操作几乎是一样的。
我也发表了一些评论和一些解释

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession

object ReadCsv {
  val spark = SparkSession
    .builder()
    .appName("ReadCsv")
    .master("local[*]")
    .config("spark.sql.shuffle.partitions","4") //Change to a more reasonable default number of partitions for our data
    .config("spark.app.id","ReadCsv") // To silence Metrics warning
    .getOrCreate()

  val sqlContext = spark.sqlContext

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

    Logger.getRootLogger.setLevel(Level.ERROR)

    try {

      val df = sqlContext
        .read
        .csv("/path/directory_to_csv_files/") // Here we read the .csv files
        .cache()
      
      df.repartition(4) // we get four files
          .write
          .parquet("/path/directory_to_parquet_files/") // output format file.parquet.snappy by default
      // if we want parquet uncompressed before write we have to do:
      // sqlContext.setConf("spark.sql.parquet.compression.codec", "uncompressed")

      // To have the opportunity to view the web console of Spark: http://localhost:4040/
      println("Type whatever to the console to exit......")
      scala.io.StdIn.readLine()
    } finally {
      spark.stop()
      println("SparkSession stopped")
    }
  }
}