Skip to content
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.

Sign up for GitHub

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

Jump to bottom

Fix for #22951 #22953

Closed
wants to merge 15 commits into from
Closed

Fix for #22951 #22953

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
2eacbd4
Fix for #22951
nbali Aug 30, 2022
ff6b5a0
Compilation fix
nbali Aug 30, 2022
5f79ed6
Spotless fix
nbali Aug 30, 2022
bcd4ba9
Fix for #22951 - PR CR - using the same chunk size as the network layer
nbali Sep 6, 2022
040b744
Fix for #22951 - PR CR - added new test cases into GroupIntoBatchesTest
nbali Sep 6, 2022
b1b732c
Fix for #22951 - PR CR - added new test cases into GroupIntoBatchesTr…
nbali Sep 7, 2022
70a92b6
Fix for #22951 - PR CR - guaranteed element order for the newly intro…
nbali Sep 7, 2022
6abe4cd
#22591 Firing GroupIntoBatches earlier if byte size would go over the…
nbali Nov 15, 2022
3e53b0f
Spotless
nbali Nov 15, 2022
f00c85d
#22951 Adding byte size limit to WriteFiles transform as well
nbali Nov 15, 2022
5840e55
#22951 Enforcing the new expectations for the 'byteSize' limit in uni…
nbali Nov 16, 2022
6d016f1
Fixing State.readLater() usage + Unit test fix to reflect changed exp…
nbali Nov 16, 2022
9f874f7
Merge branch 'master' into fix-for-22951
nbali Nov 17, 2022
5a3ee7f
Merge branch 'master' into fix-for-22951
nbali Nov 18, 2022
fa1bd88
Apply suggestions from code review
lukecwik Nov 30, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
52 changes: 44 additions & 8 deletions sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,11 @@ public class GroupIntoBatches<K, InputT>
*/
@AutoValue
public abstract static class BatchingParams<InputT> implements Serializable {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It looks like your adding support for GroupIntoBatches to limit on count and byte size at the same time.

Can you add tests that cover this new scenario to:

  • GroupIntoBatchesTest
  • GroupIntoBatchesTranslationTest

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

see 040b744 (#22953) and b1b732c (#22953)

public static <InputT> BatchingParams<InputT> createDefault() {
lukecwik marked this conversation as resolved.
Show resolved Hide resolved
return new AutoValue_GroupIntoBatches_BatchingParams(
Long.MAX_VALUE, Long.MAX_VALUE, null, Duration.ZERO);
}

public static <InputT> BatchingParams<InputT> create(
long batchSize,
long batchSizeBytes,
Expand Down Expand Up @@ -170,8 +175,7 @@ private GroupIntoBatches(BatchingParams<InputT> params) {
/** Aim to create batches each with the specified element count. */
public static <K, InputT> GroupIntoBatches<K, InputT> ofSize(long batchSize) {
Preconditions.checkState(batchSize < Long.MAX_VALUE);
return new GroupIntoBatches<>(
BatchingParams.create(batchSize, Long.MAX_VALUE, null, Duration.ZERO));
return new GroupIntoBatches<K, InputT>(BatchingParams.createDefault()).withSize(batchSize);
}

/**
Expand All @@ -185,9 +189,8 @@ public static <K, InputT> GroupIntoBatches<K, InputT> ofSize(long batchSize) {
* {@link #ofByteSize(long, SerializableFunction)} to specify code to calculate the byte size.
*/
public static <K, InputT> GroupIntoBatches<K, InputT> ofByteSize(long batchSizeBytes) {
Preconditions.checkState(batchSizeBytes < Long.MAX_VALUE);
return new GroupIntoBatches<>(
BatchingParams.create(Long.MAX_VALUE, batchSizeBytes, null, Duration.ZERO));
return new GroupIntoBatches<K, InputT>(BatchingParams.createDefault())
.withByteSize(batchSizeBytes);
}

/**
Expand All @@ -196,16 +199,49 @@ public static <K, InputT> GroupIntoBatches<K, InputT> ofByteSize(long batchSizeB
*/
public static <K, InputT> GroupIntoBatches<K, InputT> ofByteSize(
long batchSizeBytes, SerializableFunction<InputT, Long> getElementByteSize) {
Preconditions.checkState(batchSizeBytes < Long.MAX_VALUE);
return new GroupIntoBatches<>(
BatchingParams.create(Long.MAX_VALUE, batchSizeBytes, getElementByteSize, Duration.ZERO));
return new GroupIntoBatches<K, InputT>(BatchingParams.createDefault())
.withByteSize(batchSizeBytes, getElementByteSize);
}

/** Returns user supplied parameters for batching. */
public BatchingParams<InputT> getBatchingParams() {
return params;
}

/** @see #ofSize(long) */
public GroupIntoBatches<K, InputT> withSize(long batchSize) {
Preconditions.checkState(batchSize < Long.MAX_VALUE);
return new GroupIntoBatches<>(
BatchingParams.create(
batchSize,
params.getBatchSizeBytes(),
params.getElementByteSize(),
params.getMaxBufferingDuration()));
}

/** @see #ofByteSize(long) */
public GroupIntoBatches<K, InputT> withByteSize(long batchSizeBytes) {
Preconditions.checkState(batchSizeBytes < Long.MAX_VALUE);
return new GroupIntoBatches<>(
BatchingParams.create(
params.getBatchSize(),
batchSizeBytes,
params.getElementByteSize(),
params.getMaxBufferingDuration()));
}

/** @see #ofByteSize(long, SerializableFunction) */
public GroupIntoBatches<K, InputT> withByteSize(
long batchSizeBytes, SerializableFunction<InputT, Long> getElementByteSize) {
Preconditions.checkState(batchSizeBytes < Long.MAX_VALUE);
return new GroupIntoBatches<>(
BatchingParams.create(
params.getBatchSize(),
batchSizeBytes,
getElementByteSize,
params.getMaxBufferingDuration()));
}

/**
* Sets a time limit (in processing time) on how long an incomplete batch of elements is allowed
* to be buffered. Once a batch is flushed to output, the timer is reset. The provided limit must
Expand Down
4 changes: 4 additions & 0 deletions ...o/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,9 @@ class BatchLoads<DestinationT, ElementT>
// If user triggering is supplied, we will trigger the file write after this many records are
// written.
static final int FILE_TRIGGERING_RECORD_COUNT = 500000;
// If user triggering is supplied, we will trigger the file write after this many bytes are
// written.
static final long FILE_TRIGGERING_BYTE_COUNT = 100 * (1L << 20); // 100MiB
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It looks like we already have a memory limit for writing of 20 parallel writers with 64mb buffers. Should we limit this triggering to be 64mbs as well so that it fits in one chunk?

CC: @reuvenlax Any suggestions here?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@lukecwik Having the same limit as a buffer actually makes sense to me, but can you direct me towards where might I find that limit? I can see it in the comments for DEFAULT_MAX_NUM_WRITERS_PER_BUNDLE, but instead of hardcoding 64MB here as well, I would rather reference the original limit directly.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The default comes from this constant:
https://www.javadoc.io/static/com.google.cloud.bigdataoss/util/1.9.17/com/google/cloud/hadoop/util/AsyncWriteChannelOptions.html#UPLOAD_CHUNK_SIZE_DEFAULT

The user can override the default using this pipeline option:

beam/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java

Line 91 in b2a6f46

Integer getGcsUploadBufferSizeBytes();

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done in bcd4ba9 (#22953)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@lukecwik
On second thought, I think there is a problem with using this 64MB default. We only flush the batch inside GroupIntoBatches, once the storedBatchSizeBytes is greater than or equal to the limit. So if we make the limit 64MB, more than likely we will flush just a bit more than 64MB so we won't fit into the 64MB buffer.

So either the triggering byte count should be x% smaller than the 64MB default, or GroupIntoBatches has to be modified that it the current element would make it go over the byte size limit, then fire the batch without that element being added to the it first. The second seems like a better solution, but I assume doing the storedBatchSizeBytes.read() sooner would have performance impact.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry, didn't see this comment but I agree that we should change GroupIntoBatches to ensure that if we add an element that would make it go over the limit we would flush a batch.

Pseudo-code would be like:

byteSize = measure(obj)
if (byteSize >= byteSizeLimit) {
  output obj as single element batch
  continue
}
if (byteSize + previousNumBytes > byteSizeLimit) {
  output all buffered elements as batch
} else {
  add obj to batch
}

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I used a different algo, but IMO it stays closed to the original concept of the transform now.


// If using auto-sharding for unbounded data, we batch the records before triggering file write
// to avoid generating too many small files.
Expand Down Expand Up @@ -647,6 +650,7 @@ PCollection<WriteBundlesToFiles.Result<DestinationT>> writeDynamicallyShardedFil
return input
.apply(
GroupIntoBatches.<DestinationT, ElementT>ofSize(FILE_TRIGGERING_RECORD_COUNT)
.withByteSize(FILE_TRIGGERING_BYTE_COUNT)
.withMaxBufferingDuration(maxBufferingDuration)
.withShardedKey())
.setCoder(
Expand Down