-
Notifications
You must be signed in to change notification settings - Fork 28.3k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[SPARK-28209][CORE][SHUFFLE] Proposed new shuffle writer API #25007
Changes from 17 commits
1957e82
857552a
d13037f
8f5fb60
e17c7ea
3f0c131
f982df7
6891197
7b44ed2
df75f1f
a8558af
806d7bb
3167030
70f59db
3083d86
4c3d692
2421c92
982f207
594d1e2
66aae91
8b432f9
9f597dd
86c1829
a7885ae
9893c6c
cd897e7
9f17b9b
e53a001
56fa450
b8b7b8d
2d29404
06ea01a
7dceec9
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,31 @@ | ||
/* | ||
* Licensed to the Apache Software Foundation (ASF) under one or more | ||
* contributor license agreements. See the NOTICE file distributed with | ||
* this work for additional information regarding copyright ownership. | ||
* The ASF licenses this file to You under the Apache License, Version 2.0 | ||
* (the "License"); you may not use this file except in compliance with | ||
* the License. You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.spark.api.shuffle; | ||
|
||
import org.apache.spark.annotation.Experimental; | ||
|
||
/** | ||
* :: Experimental :: | ||
* An interface for launching Shuffle related components | ||
* | ||
* @since 3.0.0 | ||
*/ | ||
@Experimental | ||
public interface ShuffleDataIO { | ||
ShuffleExecutorComponents executor(); | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,33 @@ | ||
/* | ||
* Licensed to the Apache Software Foundation (ASF) under one or more | ||
* contributor license agreements. See the NOTICE file distributed with | ||
* this work for additional information regarding copyright ownership. | ||
* The ASF licenses this file to You under the Apache License, Version 2.0 | ||
* (the "License"); you may not use this file except in compliance with | ||
* the License. You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.spark.api.shuffle; | ||
|
||
import org.apache.spark.annotation.Experimental; | ||
|
||
/** | ||
* :: Experimental :: | ||
* An interface for building shuffle support for Executors | ||
* | ||
* @since 3.0.0 | ||
*/ | ||
@Experimental | ||
public interface ShuffleExecutorComponents { | ||
void initializeExecutor(String appId, String execId); | ||
|
||
ShuffleWriteSupport writes(); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. this should have a doc. At the very least, I'd mention that its called once per ShuffleMapTask |
||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,37 @@ | ||
/* | ||
* Licensed to the Apache Software Foundation (ASF) under one or more | ||
* contributor license agreements. See the NOTICE file distributed with | ||
* this work for additional information regarding copyright ownership. | ||
* The ASF licenses this file to You under the Apache License, Version 2.0 | ||
* (the "License"); you may not use this file except in compliance with | ||
* the License. You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.spark.api.shuffle; | ||
|
||
import java.io.IOException; | ||
|
||
import org.apache.spark.annotation.Experimental; | ||
|
||
/** | ||
* :: Experimental :: | ||
* An interface for creating and managing shuffle partition writers | ||
* | ||
* @since 3.0.0 | ||
*/ | ||
@Experimental | ||
public interface ShuffleMapOutputWriter { | ||
ShufflePartitionWriter getPartitionWriter(int partitionId) throws IOException; | ||
|
||
void commitAllPartitions() throws IOException; | ||
|
||
void abort(Throwable error) throws IOException; | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. these should have some more docs. Eg. at least saying one of these is created for the output of each ShuffleMapTask, and that the "partition" being referenced here is the reduce partition, so getPartitionedWriter will get called once per reduce partition |
||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,44 @@ | ||
/* | ||
* Licensed to the Apache Software Foundation (ASF) under one or more | ||
* contributor license agreements. See the NOTICE file distributed with | ||
* this work for additional information regarding copyright ownership. | ||
* The ASF licenses this file to You under the Apache License, Version 2.0 | ||
* (the "License"); you may not use this file except in compliance with | ||
* the License. You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.spark.api.shuffle; | ||
|
||
import java.io.IOException; | ||
import java.io.OutputStream; | ||
|
||
import org.apache.spark.annotation.Experimental; | ||
|
||
/** | ||
* :: Experimental :: | ||
* An interface for giving streams / channels for shuffle writes. | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. nit: should we omit "channel"? there's nothing else in the API referencing it |
||
* | ||
* @since 3.0.0 | ||
*/ | ||
@Experimental | ||
public interface ShufflePartitionWriter { | ||
|
||
/** | ||
* Opens and returns an underlying {@link OutputStream} that can write bytes to the underlying | ||
* data store. | ||
*/ | ||
OutputStream openStream() throws IOException; | ||
|
||
/** | ||
* Get the number of bytes written by this writer's stream returned by {@link #openStream()}. | ||
*/ | ||
long getNumBytesWritten(); | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,37 @@ | ||
/* | ||
* Licensed to the Apache Software Foundation (ASF) under one or more | ||
* contributor license agreements. See the NOTICE file distributed with | ||
* this work for additional information regarding copyright ownership. | ||
* The ASF licenses this file to You under the Apache License, Version 2.0 | ||
* (the "License"); you may not use this file except in compliance with | ||
* the License. You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.spark.api.shuffle; | ||
|
||
import java.io.IOException; | ||
|
||
import org.apache.spark.annotation.Experimental; | ||
|
||
/** | ||
* :: Experimental :: | ||
* An interface for deploying a shuffle map output writer | ||
* | ||
* @since 3.0.0 | ||
*/ | ||
@Experimental | ||
public interface ShuffleWriteSupport { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Since There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Found another reason to remove this There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It's important to keep interfaces minimal, and to keep each interface responsible for a single set of functionality. Since There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think there might be a world where we can coalesce There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Do you mean that There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I thought about this a bit more. One motivation to separation write APIs from read APIs is to pass in only the subsection of the plugin tree that is applicable in each case. So I only want to pass write-specific functionality to SortShuffleWriter, and only pass read-specific shuffle functionality to the reader side. But I hold this conviction pretty loosely. We can change this - let me know. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @mccheah Yeah seperating write from read is pretty natural. Here we think maybe remove the There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. That doesn't separate the concerns as I described. The only layer above I would like There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Can the There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. A comment left here @mccheah |
||
ShuffleMapOutputWriter createMapOutputWriter( | ||
int shuffleId, | ||
int mapId, | ||
mccheah marked this conversation as resolved.
Show resolved
Hide resolved
|
||
long mapTaskAttemptId, | ||
int numPartitions) throws IOException; | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,53 @@ | ||
/* | ||
* Licensed to the Apache Software Foundation (ASF) under one or more | ||
* contributor license agreements. See the NOTICE file distributed with | ||
* this work for additional information regarding copyright ownership. | ||
* The ASF licenses this file to You under the Apache License, Version 2.0 | ||
* (the "License"); you may not use this file except in compliance with | ||
* the License. You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.spark.api.shuffle; | ||
|
||
import java.io.IOException; | ||
|
||
import org.apache.spark.annotation.Experimental; | ||
|
||
/** | ||
* :: Experimental :: | ||
* Indicates that partition writers can transfer bytes directly from input byte channels to | ||
* output channels that stream data to the underlying shuffle partition storage medium. | ||
* <p> | ||
* This API is separated out for advanced users because it only needs to be used for | ||
* specific low-level optimizations. The idea is that the returned channel can transfer bytes | ||
* from the input file channel out to the backing storage system without copying data into | ||
* memory. | ||
* <p> | ||
* Most shuffle plugin implementations should use {@link ShufflePartitionWriter} instead. | ||
* | ||
* @since 3.0.0 | ||
*/ | ||
@Experimental | ||
public interface SupportsTransferTo extends ShufflePartitionWriter { | ||
|
||
/** | ||
* Opens and returns a {@link TransferrableWritableByteChannel} for transferring bytes from | ||
* input byte channels to the underlying shuffle data store. | ||
*/ | ||
TransferrableWritableByteChannel openTransferrableChannel() throws IOException; | ||
|
||
/** | ||
* Returns the number of bytes written either by this writer's output stream opened by | ||
* {@link #openStream()} or the byte channel opened by {@link #openTransferrableChannel()}. | ||
*/ | ||
@Override | ||
long getNumBytesWritten(); | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,54 @@ | ||
/* | ||
* Licensed to the Apache Software Foundation (ASF) under one or more | ||
* contributor license agreements. See the NOTICE file distributed with | ||
* this work for additional information regarding copyright ownership. | ||
* The ASF licenses this file to You under the Apache License, Version 2.0 | ||
* (the "License"); you may not use this file except in compliance with | ||
* the License. You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.spark.api.shuffle; | ||
|
||
import java.io.Closeable; | ||
import java.io.IOException; | ||
|
||
import java.nio.channels.FileChannel; | ||
import java.nio.channels.WritableByteChannel; | ||
import org.apache.spark.annotation.Experimental; | ||
|
||
/** | ||
* :: Experimental :: | ||
* Represents an output byte channel that can copy bytes from input file channels to some | ||
* arbitrary storage system. | ||
* <p> | ||
* This API is provided for advanced users who can transfer bytes from a file channel to | ||
* some output sink without copying data into memory. Most users should not need to use | ||
* this functionality; this is primarily provided for the built-in shuffle storage backends | ||
* that persist shuffle files on local disk. | ||
* <p> | ||
* For a simpler alternative, see {@link ShufflePartitionWriter}. | ||
* | ||
* @since 3.0.0 | ||
*/ | ||
@Experimental | ||
public interface TransferrableWritableByteChannel extends Closeable { | ||
|
||
/** | ||
* Copy all bytes from the source readable byte channel into this byte channel. | ||
* | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. though you mention "copy all", its probably worth repeating in this comment that this differs from FileChannel.transferTo(), in that this will block until all bytes have been transferred |
||
* @param source File to transfer bytes from. Do not call anything on this channel other than | ||
* {@link FileChannel#transferTo(long, long, WritableByteChannel)}. | ||
* @param transferStartPosition Start position of the input file to transfer from. | ||
* @param numBytesToTransfer Number of bytes to transfer from the given source. | ||
*/ | ||
void transferFrom(FileChannel source, long transferStartPosition, long numBytesToTransfer) | ||
throws IOException; | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Not sure if it is proper to add the interfaces to here
o.a.s.api
? Looks like most of the things under the api package are related to rdd functions. How about this packageo.a.s.shuffle.api
?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
+1