如何在Scala Spark中并行加载HDFS上的海量Thrift二进制文件？

Here's how you can leverage Spark's distributed computing capabilities to process your thousands of HDFS binary Thrift files using your existing decode function:

Step-by-Step Implementation #

1. Initialize Spark Context #

First, set up your SparkSession (or SparkContext for older Spark versions):

import org.apache.spark.sql.SparkSession
import org.apache.hadoop.fs.{FileSystem, Path}
import java.nio.file.Files

val spark = SparkSession.builder()
  .appName("HDFSThriftDecoder")
  .getOrCreate()

2. List HDFS File Paths #

Retrieve the full list of target files from your HDFS directory, filtering out any directories to focus only on the binary files:

val hdfsDirPath = "hdfs://your-cluster/path/to/thrift/files"
val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration)

// Get all valid file paths (exclude directories)
val hdfsFilePaths = fs.listStatus(new Path(hdfsDirPath))
  .filter(!_.isDirectory)
  .map(_.getPath.toString)

// Convert to RDD for distributed processing - adjust partitions based on your cluster size
val filePathsRDD = spark.sparkContext.parallelize(hdfsFilePaths, numPartitions = 200)

3. Distributed Decoding with Local File Download #

For each file path in the RDD, we'll handle downloading the HDFS file to the worker node's local filesystem, run your decoder, clean up temporary files, and flatten the result into individual records:

// Your existing Thrift decoder function
def decode(localFilePath: String): Array[MyCustomType] = {
  // Implementation that loads local Thrift file and returns record array
}

val recordsRDD: RDD[MyCustomType] = filePathsRDD.flatMap { hdfsFilePath =>
  // Initialize Hadoop FileSystem instance on the worker node
  val conf = spark.sparkContext.hadoopConfiguration
  val fs = FileSystem.get(conf)
  val sourcePath = new Path(hdfsFilePath)

  // Create a temporary local file to store the downloaded content
  val tempLocalFile = Files.createTempFile("thrift_temp_", ".bin").toFile

  try {
    // Copy HDFS file to the temporary local file
    fs.copyToLocalFile(sourcePath, new Path(tempLocalFile.getAbsolutePath))
    
    // Run your decoder and return the records
    decode(tempLocalFile.getAbsolutePath)
  } finally {
    // Ensure temporary file is deleted even if decoding fails
    tempLocalFile.delete()
  }
}

4. Work with the Resulting RDD #

You can now use recordsRDD like any standard Spark RDD for transformations or actions:

// Example: Count total records
recordsRDD.count()

// Example: Filter and save results
recordsRDD.filter(_.someField > 100).saveAsTextFile("hdfs://your-output/path")

Key Considerations #

• Parallelism: Adjust the numPartitions value when creating filePathsRDD to match your cluster's capacity. A good starting point is 2-3 times the number of available cores across your cluster, or equal to the number of files (whichever is larger).
• Temporary File Management: Using Files.createTempFile ensures files are stored in the system's temp directory, and the finally block guarantees cleanup even if an error occurs during decoding.
• Large File Handling: Since your decode function loads the entire file into memory, ensure your worker nodes have enough heap space to handle the largest files. Adjust Spark's executor memory settings (--executor-memory) if needed.
• HDFS Permissions: Confirm your Spark workers have read access to the target HDFS directory. This is typically managed via cluster-wide Kerberos or HDFS ACLs.
• Optimization (If You Can Modify Decode): If you can update your decode function to accept an InputStream instead of a local file path, you could skip writing to local disk entirely by reading the HDFS file directly via fs.open(sourcePath). This would be more efficient for large files.

内容的提问来源于stack exchange，提问作者Morteza Shahriari Nia