MapReduce Java程序读取CSV文件报错，求排查指导

Hey there! Let's work through the issues with your StubMapper's getMapFromCSV function. Since you didn't include the full error message or complete code for that function, I'll cover common pitfalls that cause problems in this scenario, then share a corrected implementation.

Common Issues to Watch For #

• Incorrect Column Indexing: Java uses 0-based indexing, so if you want the 1st and 6th CSV columns, you need to access indices 0 and 5 (not 1 and 6). This is a super common source of ArrayIndexOutOfBoundsException.
• Naive CSV Splitting: Using String.split(",") can break if your CSV has fields containing commas (e.g., "Doe, John"). For production code, use a proper CSV parser like OpenCSV, but for simple cases, you can add basic handling.
• Distributed File Access: In a MapReduce cluster, using FileReader with a local path won't work—files need to be distributed via DistributedCache or accessed through HDFS paths. I'll include a note on this below.
• HashMap Initialization: Loading the CSV every time map() runs is inefficient; load it once in the setup() method instead.

Corrected Mapper Implementation #

Here's a revised version of your mapper with a fixed getMapFromCSV function:

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class StubMapper extends Mapper<Object, Text, Text, Text> {
    private HashMap<String, String> userIdToCheckOutMap = new HashMap<>();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        super.setup(context);
        Configuration conf = context.getConfiguration();
        // Path to your CSV file (use HDFS path for cluster, local path for testing)
        String csvPath = conf.get("csv.input.path");
        userIdToCheckOutMap = getMapFromCSV(csvPath, conf);
    }

    @Override
    protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
        // Use the loaded HashMap in your map logic here
        // Example: if value contains a userId, look up its CheckOutDateTime
        String userId = value.toString().trim();
        String checkOutTime = userIdToCheckOutMap.get(userId);
        if (checkOutTime != null) {
            context.write(new Text(userId), new Text(checkOutTime));
        }
    }

    private HashMap<String, String> getMapFromCSV(String csvPath, Configuration conf) throws IOException {
        HashMap<String, String> map = new HashMap<>();
        FileSystem fs = FileSystem.get(conf);
        
        try (BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(new Path(csvPath))))) {
            String line;
            // Skip header if your CSV has one
            br.readLine();
            
            while ((line = br.readLine()) != null) {
                // Basic CSV split that handles commas inside quoted fields
                String[] columns = line.split(",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)");
                if (columns.length >= 6) { // Ensure we have at least 6 columns to avoid index errors
                    String userId = columns[0].trim();
                    String checkOutDateTime = columns[5].trim();
                    map.put(userId, checkOutDateTime);
                } else {
                    // Log invalid lines instead of failing hard
                    System.err.println("Skipping invalid line (not enough columns): " + line);
                }
            }
        }
        return map;
    }
}

Key Fixes in This Code #

• 0-Based Indexing: Uses columns[0] (1st column) and columns[5] (6th column) to match your requirement.
• HDFS-Compatible File Access: Uses Hadoop's FileSystem to read files, which works both locally and in a distributed cluster.
• Robust CSV Splitting: The regex in split() handles commas inside quoted fields, preventing broken column splits.
• Efficient Loading: Loads the CSV once in setup() instead of reloading it for every map task.
• Graceful Error Handling: Skips invalid lines and logs them, so your job doesn't crash immediately on bad data.

Next Steps #

If you're still getting errors, please share the full stack trace of the exception you're seeing. Common remaining issues might include:

• FileNotFoundException: Ensure the CSV path is correct (use hdfs:// prefix for cluster paths, or a local path for testing).
• ArrayIndexOutOfBoundsException: Your CSV lines might have fewer than 6 columns—double-check your data or adjust the column count check.

内容的提问来源于stack exchange，提问作者Gideok Seong