[SPARK-19365][Core]Optimize RequestMessage serialization #16706

zsxwing · 2017-01-26T01:57:05Z

What changes were proposed in this pull request?

Right now Netty PRC serializes RequestMessage using Java serialization, and the size of a single message (e.g., RequestMessage(..., "hello")`) is almost 1KB.

This PR optimizes it by serializing RequestMessage manually (eliminate unnecessary information from most messages, e.g., class names of RequestMessage, NettyRpcEndpointRef, ...), and reduces the above message size to 100+ bytes.

How was this patch tested?

Jenkins

I did a simple test to measure the improvement:

Before

$ bin/spark-shell --master local-cluster[1,4,1024]
...
scala> for (i <- 1 to 10) {
     |   val start = System.nanoTime
     |   val s = sc.parallelize(1 to 1000000, 10 * 1000).count()
     |   val end = System.nanoTime
     |   println(s"$i\t" + ((end - start)/1000/1000))
     | }
1       6830                                                                    
2       4353                                                                    
3       3322                                                                    
4       3107                                                                    
5       3235                                                                    
6       3139                                                                    
7       3156                                                                    
8       3166                                                                    
9       3091                                                                    
10      3029

After:

$ bin/spark-shell --master local-cluster[1,4,1024]
...
scala> for (i <- 1 to 10) {
     |   val start = System.nanoTime
     |   val s = sc.parallelize(1 to 1000000, 10 * 1000).count()
     |   val end = System.nanoTime
     |   println(s"$i\t" + ((end - start)/1000/1000))
     | }
1       6431                                                                    
2       3643                                                                    
3       2913                                                                    
4       2679                                                                    
5       2760                                                                    
6       2710                                                                    
7       2747                                                                    
8       2793                                                                    
9       2679                                                                    
10      2651

I also captured the TCP packets for this test. Before this patch, the total size of TCP packets is ~1.5GB. After it, it reduces to ~1.2GB.

zsxwing · 2017-01-26T01:58:12Z

core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala

 */
-private[netty] case class RequestMessage(
-    senderAddress: RpcAddress, receiver: NettyRpcEndpointRef, content: Any)
+private[netty] class RequestMessage(


Removed case to make RequestMessage non-serializable to avoid using Java serialization occasionally.

zsxwing · 2017-01-26T02:00:33Z

core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala


  @transient @volatile var client: TransportClient = _

-  private val _address = if (endpointAddress.rpcAddress != null) endpointAddress else null


Removed _address and _name to save some bytes.

zsxwing · 2017-01-26T02:02:16Z

core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala

+        out.writeUTF(senderAddress.host)
+        out.writeInt(senderAddress.port)
+      }
+      val receiverAddress = receiver.endpointAddress


Write receiver.endpointAddress rather than NettyRpcEndpointRef since we only need the address to recreate them.

SparkQA · 2017-01-26T06:08:37Z

Test build #72008 has finished for PR 16706 at commit b373c10.

This patch fails from timeout after a configured wait of `250m`.
This patch merges cleanly.
This patch adds no public classes.

srowen · 2017-01-26T10:00:35Z

core/src/main/scala/org/apache/spark/rpc/RpcEndpointAddress.scala

 * @param name Name of the endpoint.
 */
-private[spark] case class RpcEndpointAddress(val rpcAddress: RpcAddress, val name: String) {
+private[spark] case class RpcEndpointAddress(@Nullable rpcAddress: RpcAddress, name: String) {


Why Nullable here?

While I was working on this, I found I need to read codes to understand if this is nullable. Hence, I just added this annotation to improve the document.

The issue is that we don't use this otherwise in Spark and it's not actually a JDK class. It's probably not worth using it in a handful of places only.

+1 we could add a comment there instead of pulling in a new annotation

That's true. But since NettyRpcEnv already used it (added by @vanzin), I prefer to also use it in other places under the netty rpc package.

Okey. I removed Nullable.

srowen · 2017-01-26T10:02:44Z

core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala

-    endpointAddress: RpcEndpointAddress,
-    @transient @volatile private var nettyEnv: NettyRpcEnv)
-  extends RpcEndpointRef(conf) with Serializable with Logging {
+    val endpointAddress: RpcEndpointAddress,


private while we're at it?

srowen · 2017-01-26T10:03:46Z

core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala

+    val receiver: NettyRpcEndpointRef, val content: Any) {
+
+  /** Manually serialize [[RequestMessage]] to minimize the size of bytes. */
+  def serialize(nettyEnv: NettyRpcEnv): ByteBuffer = {


Wouldn't you just want to implement the standard Java serialization mechanism here?

It's different. If I just implement writeObject and call Java serialization APIs to write RequestMessage, it will write the full class name RequestMessage and a serialization id which are not needed by all RPC messages.

zsxwing · 2017-01-26T18:54:56Z

cc @vanzin

shivaram

Overall the change seems fine to me. However I wonder if a more general change is possible where we say do something like a 'closure cleaning' on the input or output to java serialization to reduce the class information written. But I guess we can discuss this in a separate JIRA

shivaram · 2017-01-26T19:28:42Z

core/src/main/scala/org/apache/spark/rpc/RpcEndpointAddress.scala

 * @param name Name of the endpoint.
 */
-private[spark] case class RpcEndpointAddress(val rpcAddress: RpcAddress, val name: String) {
+private[spark] case class RpcEndpointAddress(@Nullable rpcAddress: RpcAddress, name: String) {


+1 we could add a comment there instead of pulling in a new annotation

zsxwing · 2017-01-26T20:05:51Z

However I wonder if a more general change is possible where we say do something like a 'closure cleaning' on the input or output to java serialization to reduce the class information written.

Sounds like implementing Kryo :)

SparkQA · 2017-01-26T21:20:49Z

Test build #72036 has finished for PR 16706 at commit c4b8ff0.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

vanzin

Looks reasonable given the improvements. It'd be nice to have a targeted unit test for the new code in RequestMessage, though.

vanzin · 2017-01-27T00:03:37Z

core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala

-    senderAddress: RpcAddress, receiver: NettyRpcEndpointRef, content: Any)
+private[netty] class RequestMessage(
+    val senderAddress: RpcAddress,
+    val receiver: NettyRpcEndpointRef, val content: Any) {


nit: move content to next line

vanzin · 2017-01-27T00:04:08Z

core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala

+    val senderAddress: RpcAddress,
+    val receiver: NettyRpcEndpointRef, val content: Any) {
+
+  /** Manually serialize [[RequestMessage]] to minimize the size of bytes. */


nit: remove "of bytes".

vanzin · 2017-01-27T00:08:33Z

core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala

+      writeRpcAddress(out, senderAddress)
+      writeRpcAddress(out, receiver.address)
+      out.writeUTF(receiver.name)
+      val contentBytes = nettyEnv.serialize(content)


Hmmm... could you use JavaSerializerInstance.serializeStream here instead?

You avoid: extra object allocations in serialize, two copies of the serialized content in memory, and the extra copy operation below in out.write.

You could also use ObjectOutputStream directly (it implements DataOutput) but that makes it difficult to use Kryo later.

SparkQA · 2017-01-27T00:17:06Z

Test build #72047 has finished for PR 16706 at commit 195adc7.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-01-27T06:22:46Z

Test build #72065 has started for PR 16706 at commit 05e9a0c.

zsxwing · 2017-01-27T17:20:27Z

retest this please

zsxwing · 2017-01-27T18:26:50Z

retest this please

zsxwing · 2017-01-27T19:22:19Z

retest this please

SparkQA · 2017-01-27T22:26:06Z

Test build #72083 has finished for PR 16706 at commit 05e9a0c.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

vanzin · 2017-01-27T22:49:25Z

LGTM.

zsxwing · 2017-01-27T23:07:26Z

Thanks! Merging to master.

## What changes were proposed in this pull request? Right now Netty PRC serializes `RequestMessage` using Java serialization, and the size of a single message (e.g., RequestMessage(..., "hello")`) is almost 1KB. This PR optimizes it by serializing `RequestMessage` manually (eliminate unnecessary information from most messages, e.g., class names of `RequestMessage`, `NettyRpcEndpointRef`, ...), and reduces the above message size to 100+ bytes. ## How was this patch tested? Jenkins I did a simple test to measure the improvement: Before ``` $ bin/spark-shell --master local-cluster[1,4,1024] ... scala> for (i <- 1 to 10) { | val start = System.nanoTime | val s = sc.parallelize(1 to 1000000, 10 * 1000).count() | val end = System.nanoTime | println(s"$i\t" + ((end - start)/1000/1000)) | } 1 6830 2 4353 3 3322 4 3107 5 3235 6 3139 7 3156 8 3166 9 3091 10 3029 ``` After: ``` $ bin/spark-shell --master local-cluster[1,4,1024] ... scala> for (i <- 1 to 10) { | val start = System.nanoTime | val s = sc.parallelize(1 to 1000000, 10 * 1000).count() | val end = System.nanoTime | println(s"$i\t" + ((end - start)/1000/1000)) | } 1 6431 2 3643 3 2913 4 2679 5 2760 6 2710 7 2747 8 2793 9 2679 10 2651 ``` I also captured the TCP packets for this test. Before this patch, the total size of TCP packets is ~1.5GB. After it, it reduces to ~1.2GB. Author: Shixiong Zhu <[email protected]> Closes apache#16706 from zsxwing/rpc-opt.

Optimize RequestMessage serialization

b373c10

zsxwing commented Jan 26, 2017

View reviewed changes

srowen reviewed Jan 26, 2017

View reviewed changes

zsxwing added 2 commits January 26, 2017 10:40

Minor update

0872dfc

make endpointAddress private

c4b8ff0

shivaram reviewed Jan 26, 2017

View reviewed changes

Remove nullable

195adc7

vanzin reviewed Jan 27, 2017

View reviewed changes

Address

05e9a0c

asfgit closed this in 21aa8c3 Jan 27, 2017

zsxwing deleted the rpc-opt branch January 27, 2017 23:10


		@transient @volatile var client: TransportClient = _

		private val _address = if (endpointAddress.rpcAddress != null) endpointAddress else null

[SPARK-19365][Core]Optimize RequestMessage serialization #16706

[SPARK-19365][Core]Optimize RequestMessage serialization #16706

Uh oh!

Conversation

zsxwing commented Jan 26, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jan 26, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zsxwing commented Jan 26, 2017

Uh oh!

shivaram left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zsxwing commented Jan 26, 2017

Uh oh!

SparkQA commented Jan 26, 2017

Uh oh!

vanzin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jan 27, 2017

Uh oh!

SparkQA commented Jan 27, 2017

Uh oh!

zsxwing commented Jan 27, 2017

Uh oh!

zsxwing commented Jan 27, 2017

Uh oh!

zsxwing commented Jan 27, 2017

Uh oh!

SparkQA commented Jan 27, 2017

Uh oh!

vanzin commented Jan 27, 2017

Uh oh!

zsxwing commented Jan 27, 2017

Uh oh!

Uh oh!

zsxwing commented Jan 26, 2017 •

edited

Loading