如何将Java生成的.ser文件导入Apache Spark Streaming应用？

Got it, let's break down how to solve this—reading your custom Java .ser files with Spark Streaming is totally feasible, and here are two practical approaches you can implement right away:

Approach 1: Use FileInputDStream with Custom Deserialization #

This is the most straightforward method: leverage Spark Streaming's built-in file watcher to detect new .ser files, then add a transformation to deserialize their content back into your Java objects.

Step 1: Set Up Your Spark Streaming Context #

First, initialize the core streaming environment:

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

SparkConf conf = new SparkConf()
    .setAppName("SerFileStreamProcessor")
    .setMaster("local[*]"); // Use "yarn" or cluster URL in production

JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(10));

Step 2: Stream and Deserialize .ser Files #

Watch your target directory for new .ser files, then process each file to deserialize its objects:

import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairInputDStream;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

// Monitor the directory for new .ser files
JavaPairInputDStream<String, String> fileStream = jssc.fileStream(
    "/path/to/your/ser-files-folder",
    path -> path.getName().endsWith(".ser"), // Filter only .ser files
    false, // Don't scan existing files on startup (set to true if needed)
    String.class,
    String.class,
    TextInputFormat.class
);

// Transform to deserialize each file's content
JavaDStream<YourCustomClass> deserializedStream = fileStream.flatMap((fileContent, context) -> {
    List<YourCustomClass> objects = new ArrayList<>();
    String filePath = fileContent._1();

    try (ObjectInputStream ois = new ObjectInputStream(new FileInputStream(filePath))) {
        // Adjust this loop based on how you wrote the .ser file:
        // - If each file has one object: skip the loop and read once
        // - If multiple objects: loop until EOF is hit
        while (true) {
            try {
                YourCustomClass obj = (YourCustomClass) ois.readObject();
                objects.add(obj);
            } catch (EOFException e) {
                break; // End of file reached
            }
        }
    } catch (IOException | ClassNotFoundException e) {
        // Handle errors properly in production (log, skip the file, etc.)
        System.err.println("Failed to process file: " + filePath);
        e.printStackTrace();
    }

    return objects.iterator();
});

// Now you can process the deserialized stream (e.g., print, aggregate, write to storage)
deserializedStream.print();

jssc.start();
jssc.awaitTermination();

Critical Notes for This Approach: #

• Replace YourCustomClass with the actual class of the objects in your .ser files. Make sure this class has the same serialVersionUID and is available on all Spark worker nodes (package it into your job JAR or include it via --jars when submitting).
• Adjust the deserialization loop to match how you wrote the .ser files—if each file contains a single object, you can remove the loop and just call ois.readObject() once.
• In production, replace printStackTrace() with a proper logging framework (like SLF4J) to track errors without cluttering logs.

Approach 2: Custom Input Stream (Advanced) #

If you need more control (e.g., reading from a non-file source, custom partitioning, or real-time polling logic), you can build a custom InputDStream:

• Extend InputDStream<YourCustomClass> in Java.
• Implement logic to poll for new .ser files, read their content, deserialize objects, and emit them as RDDs at each batch interval.
• Register your custom stream with the JavaStreamingContext using jssc.receiverStream().

This is more complex, so only use it if Approach 1 doesn't fit your specific use case.

Additional Pro Tips: #

• Classpath Consistency: Ensure all dependencies of your serialized class are present on every worker node to avoid ClassNotFoundException.
• Performance: For large .ser files, split them into smaller chunks before streaming, or use compression (like GZIP) to reduce I/O overhead.
• Testing: Start with a small test .ser file to verify deserialization works, then scale up to streaming batches.

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