JVM间代码分发实现机制及Spark/Flink相关源码咨询

Great question! This is a foundational concept in distributed data processing—being able to send code across JVMs is what lets engines like Spark and Flink scale your computations across clusters. Let’s break down how this works, starting with Spark, then Flink, and highlight the key source code classes you’ll want to explore.

Spark: Code Distribution & Closure Serialization #

Core Implementation Principle #

When you run code like sc.parallelize(1 to 10).map(i => i+1).collect, here’s what happens under the hood:

1. Closure Capture: The anonymous function i => i+1 isn’t just standalone code—it’s a closure that captures any variables or dependencies from the driver’s scope (even if none are used here, the framework still handles this logic).
2. Closure Cleaning: Spark’s ClosureCleaner scans the closure to remove unnecessary references (like non-serializable driver-side objects you don’t need) and ensure only required dependencies are included. This prevents serialization failures and reduces payload size.
3. Serialization: The cleaned closure is serialized using Spark’s default JavaSerializer (or Kryo if you’ve configured it). Scala functions natively implement Serializable, so the framework can package the function bytecode and its captured state into a serialized blob.
4. Task Distribution: The driver wraps the serialized closure into a Task object, along with metadata about which data partition to process. These tasks are sent to worker nodes over the network.
5. Execution: Each worker deserializes the closure, runs it on the assigned data partition, and sends the results back to the driver.

Key Spark Source Code Classes #

• org.apache.spark.serializer.Serializer: The base abstract class for all Spark serializers. Both JavaSerializer (default) and KryoSerializer extend this.
• org.apache.spark.serializer.JavaSerializer: Uses Java’s built-in ObjectOutputStream/ObjectInputStream to serialize closures and data—this is what handles your Scala anonymous function by default.
• org.apache.spark.util.ClosureCleaner: Critical for cleaning closures to avoid serialization errors with non-serializable objects.
• org.apache.spark.scheduler.Task: Represents the unit of work sent to workers; contains the serialized closure and partition details.
• org.apache.spark.api.java.function.Function: The interface Scala anonymous functions (like your map lambda) implicitly implement, which extends Serializable.

Flink: Code Distribution & UDF Serialization #

Core Implementation Principle #

Flink follows a similar pattern but uses its own optimized serialization stack for better control and performance. For your equivalent Flink code (e.g., env.fromElements(1 to 10: _*).map(i => i+1).execute()):

1. UDF Preparation: Your anonymous map function implements Flink’s Function interface, which requires Serializable. Any captured variables are included as part of the function’s state.
2. Closure Cleaning: Flink’s ClosureCleaner removes unused, non-serializable references from the function closure, just like Spark’s version.
3. Job Graph Packaging: The JobManager packages the serialized UDFs, along with the entire job graph (which defines the computation flow), into a payload sent to TaskManagers (Flink’s worker nodes).
4. Deserialization & Execution: TaskManagers deserialize the UDFs, initialize the execution environment, and run the functions on their assigned data partitions.

Key Flink Source Code Classes #

• org.apache.flink.api.common.functions.Function: The base interface for all Flink user-defined functions (UDFs), which extends Serializable. Your anonymous lambda implements this.
• org.apache.flink.util.ClosureCleaner: Cleans closures to eliminate non-serializable dependencies and reduce payload size.
• org.apache.flink.runtime.jobgraph.JobGraph: Contains serialized UDFs and job metadata that’s distributed from the JobManager to TaskManagers.
• org.apache.flink.runtime.taskmanager.Task: The worker-side component that receives serialized tasks, deserializes UDFs, and executes the computation.
• org.apache.flink.api.common.typeutils.TypeSerializer: Flink’s core serialization interface (used for both data and UDFs) that provides optimized serialization for custom types.

Quick Note on Common Pitfalls #

A frequent issue here is trying to serialize non-serializable objects (like a driver-side database connection) in your closure. Both frameworks will throw a NotSerializableException in this case. To fix this, initialize non-serializable objects inside the function (on the worker side) or use broadcast variables to distribute shared resources.

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