Skip to content
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
96 commits
Select commit Hold shift + click to select a range
81d52c5
WIP on UnsafeSorter
JoshRosen Apr 29, 2015
abf7bfe
Add basic test case.
JoshRosen Apr 29, 2015
57a4ea0
Make initialSize configurable in UnsafeSorter
JoshRosen Apr 30, 2015
e900152
Add test for empty iterator in UnsafeSorter
JoshRosen May 1, 2015
767d3ca
Fix invalid range in UnsafeSorter.
JoshRosen May 1, 2015
3db12de
Minor simplification and sanity checks in UnsafeSorter
JoshRosen May 1, 2015
4d2f5e1
WIP
JoshRosen May 1, 2015
8e3ec20
Begin code cleanup.
JoshRosen May 1, 2015
253f13e
More cleanup
JoshRosen May 1, 2015
9c6cf58
Refactor to use DiskBlockObjectWriter.
JoshRosen May 1, 2015
e267cee
Fix compilation of UnsafeSorterSuite
JoshRosen May 1, 2015
e2d96ca
Expand serializer API and use new function to help control when new U…
JoshRosen May 1, 2015
d3cc310
Flag that SparkSqlSerializer2 supports relocation
JoshRosen May 1, 2015
87e721b
Renaming and comments
JoshRosen May 1, 2015
0748458
Port UnsafeShuffleWriter to Java.
JoshRosen May 2, 2015
026b497
Re-use a buffer in UnsafeShuffleWriter
JoshRosen May 2, 2015
1433b42
Store record length as int instead of long.
JoshRosen May 2, 2015
240864c
Remove PrefixComputer and require prefix to be specified as part of i…
JoshRosen May 2, 2015
bfc12d3
Add tests for serializer relocation property.
JoshRosen May 3, 2015
b8a09fe
Back out accidental log4j.properties change
JoshRosen May 3, 2015
c2fca17
Small refactoring of SerializerPropertiesSuite to enable test re-use:
JoshRosen May 3, 2015
f17fa8f
Add missing newline
JoshRosen May 3, 2015
8958584
Fix bug in calculating free space in current page.
JoshRosen May 3, 2015
595923a
Remove some unused variables.
JoshRosen May 3, 2015
5e100b2
Super-messy WIP on external sort
JoshRosen May 4, 2015
2776aca
First passing test for ExternalSorter.
JoshRosen May 4, 2015
f156a8f
Hacky metrics integration; refactor some interfaces.
JoshRosen May 4, 2015
3490512
Misc. cleanup
JoshRosen May 4, 2015
3aeaff7
More refactoring and cleanup; begin cleaning iterator interfaces
JoshRosen May 4, 2015
7ee918e
Re-order imports in tests
JoshRosen May 5, 2015
69232fd
Enable compressible address encoding for off-heap mode.
JoshRosen May 5, 2015
57f1ec0
WIP towards packed record pointers for use in optimized shuffle sort.
JoshRosen May 5, 2015
f480fb2
WIP in mega-refactoring towards shuffle-specific sort.
JoshRosen May 5, 2015
133c8c9
WIP towards testing UnsafeShuffleWriter.
JoshRosen May 5, 2015
4f70141
Fix merging; now passes UnsafeShuffleSuite tests.
JoshRosen May 5, 2015
aaea17b
Add comments to UnsafeShuffleSpillWriter.
JoshRosen May 6, 2015
b674412
Merge remote-tracking branch 'origin/master' into unsafe-sort
JoshRosen May 6, 2015
11feeb6
Update TODOs related to shuffle write metrics.
JoshRosen May 7, 2015
8a6fe52
Rename UnsafeShuffleSpillWriter to UnsafeShuffleExternalSorter
JoshRosen May 7, 2015
cfe0ec4
Address a number of minor review comments:
JoshRosen May 7, 2015
e67f1ea
Remove upper type bound in ShuffleWriter interface.
JoshRosen May 7, 2015
5e8cf75
More minor cleanup
JoshRosen May 7, 2015
1ce1300
More minor cleanup
JoshRosen May 7, 2015
b95e642
Refactor and document logic that decides when to spill.
JoshRosen May 7, 2015
9883e30
Merge remote-tracking branch 'origin/master' into unsafe-sort
JoshRosen May 8, 2015
722849b
Add workaround for transferTo() bug in merging code; refactor tests.
JoshRosen May 8, 2015
7cd013b
Begin refactoring to enable proper tests for spilling.
JoshRosen May 9, 2015
9b7ebed
More defensive programming RE: cleaning up spill files and memory aft…
JoshRosen May 9, 2015
e8718dd
Merge remote-tracking branch 'origin/master' into unsafe-sort
JoshRosen May 9, 2015
1929a74
Update to reflect upstream ShuffleBlockManager -> ShuffleBlockResolve…
JoshRosen May 9, 2015
01afc74
Actually read data in UnsafeShuffleWriterSuite
JoshRosen May 10, 2015
8f5061a
Strengthen assertion to check partitioning
JoshRosen May 10, 2015
67d25ba
Update Exchange operator's copying logic to account for new shuffle m…
JoshRosen May 10, 2015
fd4bb9e
Use own ByteBufferOutputStream rather than Kryo's
JoshRosen May 10, 2015
9d1ee7c
Fix MiMa excludes for ShuffleWriter change
JoshRosen May 10, 2015
fcd9a3c
Add notes + tests for maximum record / page sizes.
JoshRosen May 10, 2015
27b18b0
That for inserting records AT the max record size.
JoshRosen May 10, 2015
4a01c45
Remove unnecessary log message
JoshRosen May 10, 2015
f780fb1
Add test demonstrating which compression codecs support concatenation.
JoshRosen May 11, 2015
b57c17f
Disable some overly-verbose logs that rendered DEBUG useless.
JoshRosen May 11, 2015
1ef56c7
Revise compression codec support in merger; test cross product of con…
JoshRosen May 11, 2015
b3b1924
Properly implement close() and flush() in DummySerializerInstance.
JoshRosen May 11, 2015
0d4d199
Bump up shuffle.memoryFraction to make tests pass.
JoshRosen May 11, 2015
ec6d626
Add notes on maximum # of supported shuffle partitions.
JoshRosen May 11, 2015
ae538dc
Document UnsafeShuffleManager.
JoshRosen May 11, 2015
ea4f85f
Roll back an unnecessary change in Spillable.
JoshRosen May 11, 2015
1e3ad52
Delete unused ByteBufferOutputStream class.
JoshRosen May 11, 2015
39434f9
Avoid integer multiplication overflow in getMemoryUsage (thanks FindB…
JoshRosen May 11, 2015
e1855e5
Fix a handful of misc. IntelliJ inspections
JoshRosen May 11, 2015
7c953f9
Add test that covers UnsafeShuffleSortDataFormat.swap().
JoshRosen May 11, 2015
8531286
Add tests that automatically trigger spills.
JoshRosen May 11, 2015
69d5899
Remove some unnecessary override vals
JoshRosen May 11, 2015
d4e6d89
Update to bit shifting constants
JoshRosen May 11, 2015
4f0b770
Attempt to implement proper shuffle write metrics.
JoshRosen May 12, 2015
e58a6b4
Add more tests for PackedRecordPointer encoding.
JoshRosen May 12, 2015
e995d1a
Introduce MAX_SHUFFLE_OUTPUT_PARTITIONS.
JoshRosen May 12, 2015
56781a1
Rename UnsafeShuffleSorter to UnsafeShuffleInMemorySorter
JoshRosen May 12, 2015
0ad34da
Fix off-by-one in nextInt() call
JoshRosen May 12, 2015
85da63f
Cleanup in UnsafeShuffleSorterIterator.
JoshRosen May 12, 2015
fdcac08
Guard against overflow when expanding sort buffer.
JoshRosen May 12, 2015
2d4e4f4
Address some minor comments in UnsafeShuffleExternalSorter.
JoshRosen May 12, 2015
57312c9
Clarify fileBufferSize units
JoshRosen May 12, 2015
6276168
Remove ability to disable spilling in UnsafeShuffleExternalSorter.
JoshRosen May 12, 2015
4a2c785
rename 'sort buffer' to 'pointer array'
JoshRosen May 12, 2015
e3b8855
Cleanup in UnsafeShuffleWriter
JoshRosen May 12, 2015
c2ce78e
Fix a missed usage of MAX_PARTITION_ID
JoshRosen May 12, 2015
d5779c6
Merge remote-tracking branch 'origin/master' into unsafe-sort
JoshRosen May 12, 2015
5e189c6
Track time spend closing / flushing files; split TimeTrackingOutputSt…
JoshRosen May 12, 2015
df07699
Attempt to clarify confusing metrics update code
JoshRosen May 12, 2015
de40b9d
More comments to try to explain metrics code
JoshRosen May 12, 2015
4023fa4
Add @Private annotation to some Java classes.
JoshRosen May 12, 2015
51812a7
Change shuffle manager sort name to tungsten-sort
JoshRosen May 13, 2015
52a9981
Fix some bugs in the address packing code.
JoshRosen May 13, 2015
d494ffe
Fix deserialization of JavaSerializer instances.
JoshRosen May 13, 2015
7610f2f
Add tests for proper cleanup of shuffle data.
JoshRosen May 13, 2015
ef0a86e
Fix scalastyle errors
JoshRosen May 13, 2015
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Update to reflect upstream ShuffleBlockManager -> ShuffleBlockResolve…
…r rename.
  • Loading branch information
JoshRosen committed May 9, 2015
commit 1929a7439040e7c2439e386eb0994236e07c6780
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@
import org.apache.spark.serializer.SerializationStream;
import org.apache.spark.serializer.Serializer;
import org.apache.spark.serializer.SerializerInstance;
import org.apache.spark.shuffle.IndexShuffleBlockManager;
import org.apache.spark.shuffle.IndexShuffleBlockResolver;
import org.apache.spark.shuffle.ShuffleMemoryManager;
import org.apache.spark.shuffle.ShuffleWriter;
import org.apache.spark.storage.BlockManager;
Expand All @@ -59,7 +59,7 @@ public class UnsafeShuffleWriter<K, V> extends ShuffleWriter<K, V> {
private static final ClassTag<Object> OBJECT_CLASS_TAG = ClassTag$.MODULE$.Object();

private final BlockManager blockManager;
private final IndexShuffleBlockManager shuffleBlockManager;
private final IndexShuffleBlockResolver shuffleBlockResolver;
private final TaskMemoryManager memoryManager;
private final ShuffleMemoryManager shuffleMemoryManager;
private final SerializerInstance serializer;
Expand Down Expand Up @@ -87,15 +87,15 @@ public class UnsafeShuffleWriter<K, V> extends ShuffleWriter<K, V> {

public UnsafeShuffleWriter(
BlockManager blockManager,
IndexShuffleBlockManager shuffleBlockManager,
IndexShuffleBlockResolver shuffleBlockResolver,
TaskMemoryManager memoryManager,
ShuffleMemoryManager shuffleMemoryManager,
UnsafeShuffleHandle<K, V> handle,
int mapId,
TaskContext taskContext,
SparkConf sparkConf) {
this.blockManager = blockManager;
this.shuffleBlockManager = shuffleBlockManager;
this.shuffleBlockResolver = shuffleBlockResolver;
this.memoryManager = memoryManager;
this.shuffleMemoryManager = shuffleMemoryManager;
this.mapId = mapId;
Expand Down Expand Up @@ -162,7 +162,7 @@ void closeAndWriteOutput() throws IOException {
}
}
}
shuffleBlockManager.writeIndexFile(shuffleId, mapId, partitionLengths);
shuffleBlockResolver.writeIndexFile(shuffleId, mapId, partitionLengths);
mapStatus = MapStatus$.MODULE$.apply(blockManager.shuffleServerId(), partitionLengths);
}

Expand Down Expand Up @@ -192,7 +192,7 @@ void forceSorterToSpill() throws IOException {
}

private long[] mergeSpills(SpillInfo[] spills) throws IOException {
final File outputFile = shuffleBlockManager.getDataFile(shuffleId, mapId);
final File outputFile = shuffleBlockResolver.getDataFile(shuffleId, mapId);
try {
if (spills.length == 0) {
new FileOutputStream(outputFile).close(); // Create an empty file
Expand Down Expand Up @@ -323,7 +323,7 @@ public Option<MapStatus> stop(boolean success) {
return Option.apply(mapStatus);
} else {
// The map task failed, so delete our output data.
shuffleBlockManager.removeDataByMap(shuffleId, mapId);
shuffleBlockResolver.removeDataByMap(shuffleId, mapId);
return Option.apply(null);
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ private[spark] class UnsafeShuffleManager(conf: SparkConf) extends ShuffleManage
// TODO: do we need to do anything to register the shuffle here?
new UnsafeShuffleWriter(
env.blockManager,
shuffleBlockResolver.asInstanceOf[IndexShuffleBlockManager],
shuffleBlockResolver.asInstanceOf[IndexShuffleBlockResolver],
context.taskMemoryManager(),
env.shuffleMemoryManager,
unsafeShuffleHandle,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@

import org.apache.spark.*;
import org.apache.spark.serializer.Serializer;
import org.apache.spark.shuffle.IndexShuffleBlockManager;
import org.apache.spark.shuffle.IndexShuffleBlockResolver;
import org.apache.spark.executor.ShuffleWriteMetrics;
import org.apache.spark.executor.TaskMetrics;
import org.apache.spark.serializer.SerializerInstance;
Expand All @@ -67,7 +67,7 @@ public class UnsafeShuffleWriterSuite {

@Mock(answer = RETURNS_SMART_NULLS) ShuffleMemoryManager shuffleMemoryManager;
@Mock(answer = RETURNS_SMART_NULLS) BlockManager blockManager;
@Mock(answer = RETURNS_SMART_NULLS) IndexShuffleBlockManager shuffleBlockManager;
@Mock(answer = RETURNS_SMART_NULLS) IndexShuffleBlockResolver shuffleBlockResolver;
@Mock(answer = RETURNS_SMART_NULLS) DiskBlockManager diskBlockManager;
@Mock(answer = RETURNS_SMART_NULLS) TaskContext taskContext;
@Mock(answer = RETURNS_SMART_NULLS) ShuffleDependency<Object, Object, Object> shuffleDep;
Expand Down Expand Up @@ -124,14 +124,14 @@ public DiskBlockObjectWriter answer(InvocationOnMock invocationOnMock) throws Th
when(blockManager.wrapForCompression(any(BlockId.class), any(InputStream.class)))
.then(returnsSecondArg());

when(shuffleBlockManager.getDataFile(anyInt(), anyInt())).thenReturn(mergedOutputFile);
when(shuffleBlockResolver.getDataFile(anyInt(), anyInt())).thenReturn(mergedOutputFile);
doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocationOnMock) throws Throwable {
partitionSizesInMergedFile = (long[]) invocationOnMock.getArguments()[2];
return null;
}
}).when(shuffleBlockManager).writeIndexFile(anyInt(), anyInt(), any(long[].class));
}).when(shuffleBlockResolver).writeIndexFile(anyInt(), anyInt(), any(long[].class));

when(diskBlockManager.createTempShuffleBlock()).thenAnswer(
new Answer<Tuple2<TempLocalBlockId, File>>() {
Expand All @@ -157,7 +157,7 @@ private UnsafeShuffleWriter<Object, Object> createWriter(boolean transferToEnabl
conf.set("spark.file.transferTo", String.valueOf(transferToEnabled));
return new UnsafeShuffleWriter<Object, Object>(
blockManager,
shuffleBlockManager,
shuffleBlockResolver,
taskMemoryManager,
shuffleMemoryManager,
new UnsafeShuffleHandle<Object, Object>(0, 1, shuffleDep),
Expand Down