You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2019/02/05 17:33:31 UTC
[beam] branch release-2.10.0 updated: Merge pull request #7714:
[BEAM-6579] Fix create disposition for the case of temporary tables
This is an automated email from the ASF dual-hosted git repository.
kenn pushed a commit to branch release-2.10.0
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/release-2.10.0 by this push:
new e46bdcb Merge pull request #7714: [BEAM-6579] Fix create disposition for the case of temporary tables
new 8e610ad Merge pull request #7733: Cherrypick #7714 for 2.10.0: [BEAM-6579] Fix create disposition for the case of temporary tables
e46bdcb is described below
commit e46bdcb63bb244f88eddf0820beea0d709899117
Author: reuvenlax <re...@google.com>
AuthorDate: Mon Feb 4 22:09:49 2019 -0800
Merge pull request #7714: [BEAM-6579] Fix create disposition for the case of temporary tables
---
.../beam/sdk/io/gcp/bigquery/BatchLoads.java | 29 +++++++++--
.../beam/sdk/io/gcp/bigquery/BigQueryIO.java | 31 ++++++++++++
.../beam/sdk/io/gcp/bigquery/WritePartition.java | 37 +++++++++++---
.../beam/sdk/io/gcp/bigquery/WriteTables.java | 26 +++++++---
.../sdk/io/gcp/bigquery/BigQueryIOWriteTest.java | 59 ++++++++++++++++++----
5 files changed, 154 insertions(+), 28 deletions(-)
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java
index 67bcf5a..abbcd7b 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java
@@ -91,11 +91,11 @@ class BatchLoads<DestinationT>
@VisibleForTesting
// Maximum number of files in a single partition.
- static final int MAX_NUM_FILES = 10000;
+ static final int DEFAULT_MAX_FILES_PER_PARTITION = 10000;
@VisibleForTesting
// Maximum number of bytes in a single partition -- 11 TiB just under BQ's 12 TiB limit.
- static final long MAX_SIZE_BYTES = 11 * (1L << 40);
+ static final long DEFAULT_MAX_BYTES_PER_PARTITION = 11 * (1L << 40);
// The maximum size of a single file - 4TiB, just under the 5 TiB limit.
static final long DEFAULT_MAX_FILE_SIZE = 4 * (1L << 40);
@@ -123,6 +123,8 @@ class BatchLoads<DestinationT>
private final Coder<DestinationT> destinationCoder;
private int maxNumWritersPerBundle;
private long maxFileSize;
+ private int maxFilesPerPartition;
+ private long maxBytesPerPartition;
private int numFileShards;
private Duration triggeringFrequency;
private ValueProvider<String> customGcsTempLocation;
@@ -149,6 +151,8 @@ class BatchLoads<DestinationT>
this.maxNumWritersPerBundle = DEFAULT_MAX_NUM_WRITERS_PER_BUNDLE;
this.maxFileSize = DEFAULT_MAX_FILE_SIZE;
this.numFileShards = DEFAULT_NUM_FILE_SHARDS;
+ this.maxFilesPerPartition = DEFAULT_MAX_FILES_PER_PARTITION;
+ this.maxBytesPerPartition = DEFAULT_MAX_BYTES_PER_PARTITION;
this.triggeringFrequency = null;
this.customGcsTempLocation = customGcsTempLocation;
this.loadJobProjectId = loadJobProjectId;
@@ -190,6 +194,16 @@ class BatchLoads<DestinationT>
this.maxFileSize = maxFileSize;
}
+ @VisibleForTesting
+ void setMaxFilesPerPartition(int maxFilesPerPartition) {
+ this.maxFilesPerPartition = maxFilesPerPartition;
+ }
+
+ @VisibleForTesting
+ void setMaxBytesPerPartition(long maxBytesPerPartition) {
+ this.maxBytesPerPartition = maxBytesPerPartition;
+ }
+
@Override
public void validate(PipelineOptions options) {
// We will use a BigQuery load job -- validate the temp location.
@@ -278,6 +292,8 @@ class BatchLoads<DestinationT>
singletonTable,
dynamicDestinations,
tempFilePrefixView,
+ maxFilesPerPartition,
+ maxBytesPerPartition,
multiPartitionsTag,
singlePartitionTag))
.withSideInputs(tempFilePrefixView)
@@ -345,6 +361,8 @@ class BatchLoads<DestinationT>
singletonTable,
dynamicDestinations,
tempFilePrefixView,
+ maxFilesPerPartition,
+ maxBytesPerPartition,
multiPartitionsTag,
singlePartitionTag))
.withSideInputs(tempFilePrefixView)
@@ -515,7 +533,8 @@ class BatchLoads<DestinationT>
ShardedKeyCoder.of(NullableCoder.of(destinationCoder)),
ListCoder.of(StringUtf8Coder.of()));
- // If WriteBundlesToFiles produced more than MAX_NUM_FILES files or MAX_SIZE_BYTES bytes, then
+ // If WriteBundlesToFiles produced more than DEFAULT_MAX_FILES_PER_PARTITION files or
+ // DEFAULT_MAX_BYTES_PER_PARTITION bytes, then
// the import needs to be split into multiple partitions, and those partitions will be
// specified in multiPartitionsTag.
return input
@@ -526,7 +545,7 @@ class BatchLoads<DestinationT>
.apply(
"MultiPartitionsWriteTables",
new WriteTables<>(
- false,
+ true,
bigQueryServices,
jobIdTokenView,
WriteDisposition.WRITE_EMPTY,
@@ -558,7 +577,7 @@ class BatchLoads<DestinationT>
.apply(
"SinglePartitionWriteTables",
new WriteTables<>(
- true,
+ false,
bigQueryServices,
loadJobIdPrefixView,
writeDisposition,
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java
index b4dba1b..56015e2 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java
@@ -1055,6 +1055,8 @@ public class BigQueryIO {
.setExtendedErrorInfo(false)
.setSkipInvalidRows(false)
.setIgnoreUnknownValues(false)
+ .setMaxFilesPerPartition(BatchLoads.DEFAULT_MAX_FILES_PER_PARTITION)
+ .setMaxBytesPerPartition(BatchLoads.DEFAULT_MAX_BYTES_PER_PARTITION)
.build();
}
@@ -1147,6 +1149,10 @@ public class BigQueryIO {
abstract int getNumFileShards();
+ abstract int getMaxFilesPerPartition();
+
+ abstract long getMaxBytesPerPartition();
+
@Nullable
abstract Duration getTriggeringFrequency();
@@ -1202,6 +1208,10 @@ public class BigQueryIO {
abstract Builder<T> setNumFileShards(int numFileShards);
+ abstract Builder<T> setMaxFilesPerPartition(int maxFilesPerPartition);
+
+ abstract Builder<T> setMaxBytesPerPartition(long maxBytesPerPartition);
+
abstract Builder<T> setTriggeringFrequency(Duration triggeringFrequency);
abstract Builder<T> setMethod(Method method);
@@ -1544,6 +1554,24 @@ public class BigQueryIO {
return toBuilder().setMaxFileSize(maxFileSize).build();
}
+ @VisibleForTesting
+ Write<T> withMaxFilesPerPartition(int maxFilesPerPartition) {
+ checkArgument(
+ maxFilesPerPartition > 0,
+ "maxFilesPerPartition must be > 0, but was: %s",
+ maxFilesPerPartition);
+ return toBuilder().setMaxFilesPerPartition(maxFilesPerPartition).build();
+ }
+
+ @VisibleForTesting
+ Write<T> withMaxBytesPerPartition(long maxBytesPerPartition) {
+ checkArgument(
+ maxBytesPerPartition > 0,
+ "maxFilesPerPartition must be > 0, but was: %s",
+ maxBytesPerPartition);
+ return toBuilder().setMaxBytesPerPartition(maxBytesPerPartition).build();
+ }
+
@Override
public void validate(PipelineOptions pipelineOptions) {
BigQueryOptions options = pipelineOptions.as(BigQueryOptions.class);
@@ -1735,6 +1763,9 @@ public class BigQueryIO {
if (getMaxFileSize() != null) {
batchLoads.setMaxFileSize(getMaxFileSize());
}
+ batchLoads.setMaxFilesPerPartition(getMaxFilesPerPartition());
+ batchLoads.setMaxBytesPerPartition(getMaxBytesPerPartition());
+
// When running in streaming (unbounded mode) we want to retry failed load jobs
// indefinitely. Failing the bundle is expensive, so we set a fairly high limit on retries.
if (IsBounded.UNBOUNDED.equals(input.isBounded())) {
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WritePartition.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WritePartition.java
index 3d2b38a..52204f5 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WritePartition.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WritePartition.java
@@ -40,6 +40,9 @@ class WritePartition<DestinationT>
private final boolean singletonTable;
private final DynamicDestinations<?, DestinationT> dynamicDestinations;
private final PCollectionView<String> tempFilePrefix;
+ private final int maxNumFiles;
+ private final long maxSizeBytes;
+
@Nullable private TupleTag<KV<ShardedKey<DestinationT>, List<String>>> multiPartitionsTag;
private TupleTag<KV<ShardedKey<DestinationT>, List<String>>> singlePartitionTag;
@@ -47,6 +50,17 @@ class WritePartition<DestinationT>
private int numFiles = 0;
private long byteSize = 0;
private List<String> filenames = Lists.newArrayList();
+ private final int maxNumFiles;
+ private final long maxSizeBytes;
+
+ private PartitionData(int maxNumFiles, long maxSizeBytes) {
+ this.maxNumFiles = maxNumFiles;
+ this.maxSizeBytes = maxSizeBytes;
+ }
+
+ static PartitionData withMaximums(int maxNumFiles, long maxSizeBytes) {
+ return new PartitionData(maxNumFiles, maxSizeBytes);
+ }
int getNumFiles() {
return numFiles;
@@ -75,17 +89,23 @@ class WritePartition<DestinationT>
// Check to see whether we can add to this partition without exceeding the maximum partition
// size.
boolean canAccept(int numFiles, long numBytes) {
- return this.numFiles + numFiles <= BatchLoads.MAX_NUM_FILES
- && this.byteSize + numBytes <= BatchLoads.MAX_SIZE_BYTES;
+ if (filenames.isEmpty()) {
+ return true;
+ }
+ return this.numFiles + numFiles <= maxNumFiles && this.byteSize + numBytes <= maxSizeBytes;
}
}
private static class DestinationData {
private List<PartitionData> partitions = Lists.newArrayList();
- DestinationData() {
+ private DestinationData() {}
+
+ private static DestinationData create(int maxNumFiles, long maxSizeBytes) {
+ DestinationData destinationData = new DestinationData();
// Always start out with a single empty partition.
- partitions.add(new PartitionData());
+ destinationData.partitions.add(new PartitionData(maxNumFiles, maxSizeBytes));
+ return destinationData;
}
List<PartitionData> getPartitions() {
@@ -105,11 +125,15 @@ class WritePartition<DestinationT>
boolean singletonTable,
DynamicDestinations<?, DestinationT> dynamicDestinations,
PCollectionView<String> tempFilePrefix,
+ int maxNumFiles,
+ long maxSizeBytes,
TupleTag<KV<ShardedKey<DestinationT>, List<String>>> multiPartitionsTag,
TupleTag<KV<ShardedKey<DestinationT>, List<String>>> singlePartitionTag) {
this.singletonTable = singletonTable;
this.dynamicDestinations = dynamicDestinations;
this.tempFilePrefix = tempFilePrefix;
+ this.maxNumFiles = maxNumFiles;
+ this.maxSizeBytes = maxSizeBytes;
this.multiPartitionsTag = multiPartitionsTag;
this.singlePartitionTag = singlePartitionTag;
}
@@ -138,12 +162,13 @@ class WritePartition<DestinationT>
for (WriteBundlesToFiles.Result<DestinationT> fileResult : results) {
DestinationT destination = fileResult.destination;
DestinationData destinationData =
- currentResults.computeIfAbsent(destination, k -> new DestinationData());
+ currentResults.computeIfAbsent(
+ destination, k -> DestinationData.create(maxNumFiles, maxSizeBytes));
PartitionData latestPartition = destinationData.getLatestPartition();
if (!latestPartition.canAccept(1, fileResult.fileByteSize)) {
// Too much data, roll over to a new partition.
- latestPartition = new PartitionData();
+ latestPartition = PartitionData.withMaximums(maxNumFiles, maxSizeBytes);
destinationData.addPartition(latestPartition);
}
latestPartition.addFilename(fileResult.filename);
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WriteTables.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WriteTables.java
index 4f7b5c4..3d94857 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WriteTables.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WriteTables.java
@@ -83,7 +83,7 @@ class WriteTables<DestinationT>
PCollection<KV<TableDestination, String>>> {
private static final Logger LOG = LoggerFactory.getLogger(WriteTables.class);
- private final boolean singlePartition;
+ private final boolean tempTable;
private final BigQueryServices bqServices;
private final PCollectionView<String> loadJobIdPrefixView;
private final WriteDisposition firstPaneWriteDisposition;
@@ -175,14 +175,24 @@ class WriteTables<DestinationT>
BigQueryHelpers.createJobId(
c.sideInput(loadJobIdPrefixView), tableDestination, partition, c.pane().getIndex());
- if (!singlePartition) {
+ if (tempTable) {
+ // This is a temp table. Create a new one for each partition and each pane.
tableReference.setTableId(jobIdPrefix);
}
- WriteDisposition writeDisposition =
- (c.pane().getIndex() == 0) ? firstPaneWriteDisposition : WriteDisposition.WRITE_APPEND;
- CreateDisposition createDisposition =
- (c.pane().getIndex() == 0) ? firstPaneCreateDisposition : CreateDisposition.CREATE_NEVER;
+ WriteDisposition writeDisposition = firstPaneWriteDisposition;
+ CreateDisposition createDisposition = firstPaneCreateDisposition;
+ if (c.pane().getIndex() > 0 && !tempTable) {
+ // If writing directly to the destination, then the table is created on the first write
+ // and we should change the disposition for subsequent writes.
+ writeDisposition = WriteDisposition.WRITE_APPEND;
+ createDisposition = CreateDisposition.CREATE_NEVER;
+ } else if (tempTable) {
+ // In this case, we are writing to a temp table and always need to create it.
+ // WRITE_TRUNCATE is set so that we properly handle retries of this pane.
+ writeDisposition = WriteDisposition.WRITE_TRUNCATE;
+ createDisposition = CreateDisposition.CREATE_IF_NEEDED;
+ }
BigQueryHelpers.PendingJob retryJob =
startLoad(
@@ -252,7 +262,7 @@ class WriteTables<DestinationT>
}
public WriteTables(
- boolean singlePartition,
+ boolean tempTable,
BigQueryServices bqServices,
PCollectionView<String> loadJobIdPrefixView,
WriteDisposition writeDisposition,
@@ -262,7 +272,7 @@ class WriteTables<DestinationT>
@Nullable ValueProvider<String> loadJobProjectId,
int maxRetryJobs,
boolean ignoreUnknownValues) {
- this.singlePartition = singlePartition;
+ this.tempTable = tempTable;
this.bqServices = bqServices;
this.loadJobIdPrefixView = loadJobIdPrefixView;
this.firstPaneWriteDisposition = writeDisposition;
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java
index 28618e4..47297e2 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java
@@ -64,11 +64,9 @@ import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.testing.ExpectedLogs;
-import org.apache.beam.sdk.testing.NeedsRunner;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.testing.TestStream;
-import org.apache.beam.sdk.testing.UsesTestStream;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFnTester;
import org.apache.beam.sdk.transforms.MapElements;
@@ -103,7 +101,6 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
-import org.junit.experimental.categories.Category;
import org.junit.rules.ExpectedException;
import org.junit.rules.TemporaryFolder;
import org.junit.rules.TestRule;
@@ -361,7 +358,6 @@ public class BigQueryIOWriteTest implements Serializable {
}
@Test
- @Category({NeedsRunner.class, UsesTestStream.class})
public void testTriggeredFileLoads() throws Exception {
List<TableRow> elements = Lists.newArrayList();
for (int i = 0; i < 30; ++i) {
@@ -402,6 +398,48 @@ public class BigQueryIOWriteTest implements Serializable {
}
@Test
+ public void testTriggeredFileLoadsWithTempTables() throws Exception {
+ List<TableRow> elements = Lists.newArrayList();
+ for (int i = 0; i < 30; ++i) {
+ elements.add(new TableRow().set("number", i));
+ }
+
+ TestStream<TableRow> testStream =
+ TestStream.create(TableRowJsonCoder.of())
+ .addElements(
+ elements.get(0), Iterables.toArray(elements.subList(1, 10), TableRow.class))
+ .advanceProcessingTime(Duration.standardMinutes(1))
+ .addElements(
+ elements.get(10), Iterables.toArray(elements.subList(11, 20), TableRow.class))
+ .advanceProcessingTime(Duration.standardMinutes(1))
+ .addElements(
+ elements.get(20), Iterables.toArray(elements.subList(21, 30), TableRow.class))
+ .advanceWatermarkToInfinity();
+
+ p.apply(testStream)
+ .apply(
+ BigQueryIO.writeTableRows()
+ .to("project-id:dataset-id.table-id")
+ .withSchema(
+ new TableSchema()
+ .setFields(
+ ImmutableList.of(
+ new TableFieldSchema().setName("number").setType("INTEGER"))))
+ .withTestServices(fakeBqServices)
+ .withTriggeringFrequency(Duration.standardSeconds(30))
+ .withNumFileShards(2)
+ .withMaxBytesPerPartition(1)
+ .withMaxFilesPerPartition(1)
+ .withMethod(BigQueryIO.Write.Method.FILE_LOADS)
+ .withoutValidation());
+ p.run();
+
+ assertThat(
+ fakeDatasetService.getAllRows("project-id", "dataset-id", "table-id"),
+ containsInAnyOrder(Iterables.toArray(elements, TableRow.class)));
+ }
+
+ @Test
public void testFailuresNoRetryPolicy() throws Exception {
TableRow row1 = new TableRow().set("name", "a").set("number", "1");
TableRow row2 = new TableRow().set("name", "b").set("number", "2");
@@ -994,7 +1032,7 @@ public class BigQueryIOWriteTest implements Serializable {
@Test
public void testWritePartitionSinglePartition() throws Exception {
- long numFiles = BatchLoads.MAX_NUM_FILES;
+ long numFiles = BatchLoads.DEFAULT_MAX_FILES_PER_PARTITION;
long fileSize = 1;
// One partition is needed.
@@ -1004,10 +1042,11 @@ public class BigQueryIOWriteTest implements Serializable {
@Test
public void testWritePartitionManyFiles() throws Exception {
- long numFiles = BatchLoads.MAX_NUM_FILES * 3;
+ long numFiles = BatchLoads.DEFAULT_MAX_FILES_PER_PARTITION * 3;
long fileSize = 1;
- // One partition is needed for each group of BigQueryWrite.MAX_NUM_FILES files.
+ // One partition is needed for each group of BigQueryWrite.DEFAULT_MAX_FILES_PER_PARTITION
+ // files.
long expectedNumPartitions = 3;
testWritePartition(2, numFiles, fileSize, expectedNumPartitions);
}
@@ -1015,7 +1054,7 @@ public class BigQueryIOWriteTest implements Serializable {
@Test
public void testWritePartitionLargeFileSize() throws Exception {
long numFiles = 10;
- long fileSize = BatchLoads.MAX_SIZE_BYTES / 3;
+ long fileSize = BatchLoads.DEFAULT_MAX_BYTES_PER_PARTITION / 3;
// One partition is needed for each group of three files.
long expectedNumPartitions = 4;
@@ -1075,6 +1114,8 @@ public class BigQueryIOWriteTest implements Serializable {
isSingleton,
dynamicDestinations,
tempFilePrefixView,
+ BatchLoads.DEFAULT_MAX_FILES_PER_PARTITION,
+ BatchLoads.DEFAULT_MAX_BYTES_PER_PARTITION,
multiPartitionsTag,
singlePartitionTag);
@@ -1185,7 +1226,7 @@ public class BigQueryIOWriteTest implements Serializable {
fakeJobService.setNumFailuresExpected(3);
WriteTables<String> writeTables =
new WriteTables<>(
- false,
+ true,
fakeBqServices,
jobIdTokenView,
BigQueryIO.Write.WriteDisposition.WRITE_EMPTY,