You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by tg...@apache.org on 2017/03/20 16:21:29 UTC
[1/3] beam git commit: Cleanup BigQueryIO
Repository: beam
Updated Branches:
refs/heads/master d47ae8763 -> 6816d9870
Cleanup BigQueryIO
* Input type T
* Allow data-dependent tables, based on ValueInSingleWindow
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/097573ae
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/097573ae
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/097573ae
Branch: refs/heads/master
Commit: 097573ae41b0bdbf048fca4281ca678955ed32a6
Parents: d47ae87
Author: Reuven Lax <re...@google.com>
Authored: Mon Mar 13 15:25:25 2017 -0700
Committer: Thomas Groh <tg...@google.com>
Committed: Mon Mar 20 09:20:53 2017 -0700
----------------------------------------------------------------------
.../beam/examples/complete/AutoComplete.java | 2 +-
.../examples/complete/StreamingWordExtract.java | 2 +-
.../examples/complete/TrafficMaxLaneFlow.java | 2 +-
.../beam/examples/complete/TrafficRoutes.java | 2 +-
.../examples/cookbook/BigQueryTornadoes.java | 10 +-
.../cookbook/CombinePerKeyExamples.java | 2 +-
.../beam/examples/cookbook/FilterExamples.java | 2 +-
.../examples/cookbook/MaxPerKeyExamples.java | 2 +-
.../beam/examples/cookbook/TriggerExample.java | 4 +-
.../complete/game/utils/WriteToBigQuery.java | 10 +-
.../game/utils/WriteWindowedToBigQuery.java | 6 +-
.../beam/sdk/io/gcp/bigquery/BigQueryIO.java | 215 ++++++++++++-------
.../sdk/io/gcp/bigquery/BigQueryIOTest.java | 118 +++++-----
13 files changed, 221 insertions(+), 156 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/097573ae/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java
----------------------------------------------------------------------
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java b/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java
index fba3dc0..e6621ce 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java
@@ -491,7 +491,7 @@ public class AutoComplete {
toWrite
.apply(ParDo.of(new FormatForBigquery()))
- .apply(BigQueryIO.write()
+ .apply(BigQueryIO.writeTableRows()
.to(tableRef)
.withSchema(FormatForBigquery.getSchema())
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
http://git-wip-us.apache.org/repos/asf/beam/blob/097573ae/examples/java/src/main/java/org/apache/beam/examples/complete/StreamingWordExtract.java
----------------------------------------------------------------------
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/StreamingWordExtract.java b/examples/java/src/main/java/org/apache/beam/examples/complete/StreamingWordExtract.java
index 2e7d451..20cee01 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/StreamingWordExtract.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/StreamingWordExtract.java
@@ -136,7 +136,7 @@ public class StreamingWordExtract {
.apply(ParDo.of(new ExtractWords()))
.apply(ParDo.of(new Uppercase()))
.apply(ParDo.of(new StringToRowConverter()))
- .apply(BigQueryIO.write().to(tableSpec)
+ .apply(BigQueryIO.writeTableRows().to(tableSpec)
.withSchema(StringToRowConverter.getSchema()));
PipelineResult result = pipeline.run();
http://git-wip-us.apache.org/repos/asf/beam/blob/097573ae/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficMaxLaneFlow.java
----------------------------------------------------------------------
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficMaxLaneFlow.java b/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficMaxLaneFlow.java
index c9508eb..e57da93 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficMaxLaneFlow.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficMaxLaneFlow.java
@@ -348,7 +348,7 @@ public class TrafficMaxLaneFlow {
Duration.standardMinutes(options.getWindowDuration())).
every(Duration.standardMinutes(options.getWindowSlideEvery()))))
.apply(new MaxLaneFlow())
- .apply(BigQueryIO.write().to(tableRef)
+ .apply(BigQueryIO.writeTableRows().to(tableRef)
.withSchema(FormatMaxesFn.getSchema()));
// Run the pipeline.
http://git-wip-us.apache.org/repos/asf/beam/blob/097573ae/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficRoutes.java
----------------------------------------------------------------------
diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficRoutes.java b/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficRoutes.java
index fc5eb89..b1f938b 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficRoutes.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/complete/TrafficRoutes.java
@@ -359,7 +359,7 @@ public class TrafficRoutes {
Duration.standardMinutes(options.getWindowDuration())).
every(Duration.standardMinutes(options.getWindowSlideEvery()))))
.apply(new TrackSpeed())
- .apply(BigQueryIO.write().to(tableRef)
+ .apply(BigQueryIO.writeTableRows().to(tableRef)
.withSchema(FormatStatsFn.getSchema()));
// Run the pipeline.
http://git-wip-us.apache.org/repos/asf/beam/blob/097573ae/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryTornadoes.java
----------------------------------------------------------------------
diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryTornadoes.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryTornadoes.java
index f8bc104..07a3edd 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryTornadoes.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryTornadoes.java
@@ -158,11 +158,11 @@ public class BigQueryTornadoes {
p.apply(BigQueryIO.read().from(options.getInput()))
.apply(new CountTornadoes())
- .apply(BigQueryIO.write()
- .to(options.getOutput())
- .withSchema(schema)
- .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
- .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE));
+ .apply(BigQueryIO.writeTableRows()
+ .to(options.getOutput())
+ .withSchema(schema)
+ .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
+ .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE));
p.run().waitUntilFinish();
}
http://git-wip-us.apache.org/repos/asf/beam/blob/097573ae/examples/java/src/main/java/org/apache/beam/examples/cookbook/CombinePerKeyExamples.java
----------------------------------------------------------------------
diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/CombinePerKeyExamples.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/CombinePerKeyExamples.java
index a7016b0..8d13b90 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/CombinePerKeyExamples.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/CombinePerKeyExamples.java
@@ -202,7 +202,7 @@ public class CombinePerKeyExamples {
p.apply(BigQueryIO.read().from(options.getInput()))
.apply(new PlaysForWord())
- .apply(BigQueryIO.write()
+ .apply(BigQueryIO.writeTableRows()
.to(options.getOutput())
.withSchema(schema)
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
http://git-wip-us.apache.org/repos/asf/beam/blob/097573ae/examples/java/src/main/java/org/apache/beam/examples/cookbook/FilterExamples.java
----------------------------------------------------------------------
diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/FilterExamples.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/FilterExamples.java
index 7adf7c6..815ac7b 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/FilterExamples.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/FilterExamples.java
@@ -241,7 +241,7 @@ public class FilterExamples {
p.apply(BigQueryIO.read().from(options.getInput()))
.apply(ParDo.of(new ProjectionFn()))
.apply(new BelowGlobalMean(options.getMonthFilter()))
- .apply(BigQueryIO.write()
+ .apply(BigQueryIO.writeTableRows()
.to(options.getOutput())
.withSchema(schema)
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
http://git-wip-us.apache.org/repos/asf/beam/blob/097573ae/examples/java/src/main/java/org/apache/beam/examples/cookbook/MaxPerKeyExamples.java
----------------------------------------------------------------------
diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/MaxPerKeyExamples.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/MaxPerKeyExamples.java
index a8dc7f9..295b3f4 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/MaxPerKeyExamples.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/MaxPerKeyExamples.java
@@ -151,7 +151,7 @@ public class MaxPerKeyExamples {
p.apply(BigQueryIO.read().from(options.getInput()))
.apply(new MaxMeanTemp())
- .apply(BigQueryIO.write()
+ .apply(BigQueryIO.writeTableRows()
.to(options.getOutput())
.withSchema(schema)
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
http://git-wip-us.apache.org/repos/asf/beam/blob/097573ae/examples/java/src/main/java/org/apache/beam/examples/cookbook/TriggerExample.java
----------------------------------------------------------------------
diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/TriggerExample.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/TriggerExample.java
index 7048bde..0b5d9ad 100644
--- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/TriggerExample.java
+++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/TriggerExample.java
@@ -452,7 +452,9 @@ public class TriggerExample {
.apply(new CalculateTotalFlow(options.getWindowDuration()));
for (int i = 0; i < resultList.size(); i++){
- resultList.get(i).apply(BigQueryIO.write().to(tableRef).withSchema(getSchema()));
+ resultList.get(i).apply(BigQueryIO.writeTableRows()
+ .to(tableRef)
+ .withSchema(getSchema()));
}
PipelineResult result = pipeline.run();
http://git-wip-us.apache.org/repos/asf/beam/blob/097573ae/examples/java8/src/main/java/org/apache/beam/examples/complete/game/utils/WriteToBigQuery.java
----------------------------------------------------------------------
diff --git a/examples/java8/src/main/java/org/apache/beam/examples/complete/game/utils/WriteToBigQuery.java b/examples/java8/src/main/java/org/apache/beam/examples/complete/game/utils/WriteToBigQuery.java
index c124624..d60510f 100644
--- a/examples/java8/src/main/java/org/apache/beam/examples/complete/game/utils/WriteToBigQuery.java
+++ b/examples/java8/src/main/java/org/apache/beam/examples/complete/game/utils/WriteToBigQuery.java
@@ -121,12 +121,10 @@ public class WriteToBigQuery<InputT>
public PDone expand(PCollection<InputT> teamAndScore) {
return teamAndScore
.apply("ConvertToRow", ParDo.of(new BuildRowFn()))
- .apply(BigQueryIO.write()
- .to(getTable(teamAndScore.getPipeline(),
- tableName))
- .withSchema(getSchema())
- .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
- .withWriteDisposition(WriteDisposition.WRITE_APPEND));
+ .apply(BigQueryIO.writeTableRows().to(getTable(teamAndScore.getPipeline(), tableName))
+ .withSchema(getSchema())
+ .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
+ .withWriteDisposition(WriteDisposition.WRITE_APPEND));
}
/** Utility to construct an output table reference. */
http://git-wip-us.apache.org/repos/asf/beam/blob/097573ae/examples/java8/src/main/java/org/apache/beam/examples/complete/game/utils/WriteWindowedToBigQuery.java
----------------------------------------------------------------------
diff --git a/examples/java8/src/main/java/org/apache/beam/examples/complete/game/utils/WriteWindowedToBigQuery.java b/examples/java8/src/main/java/org/apache/beam/examples/complete/game/utils/WriteWindowedToBigQuery.java
index 7d16fa9..b8e12c1 100644
--- a/examples/java8/src/main/java/org/apache/beam/examples/complete/game/utils/WriteWindowedToBigQuery.java
+++ b/examples/java8/src/main/java/org/apache/beam/examples/complete/game/utils/WriteWindowedToBigQuery.java
@@ -19,6 +19,7 @@ package org.apache.beam.examples.complete.game.utils;
import com.google.api.services.bigquery.model.TableRow;
import java.util.Map;
+
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
@@ -60,9 +61,8 @@ public class WriteWindowedToBigQuery<T>
public PDone expand(PCollection<T> teamAndScore) {
return teamAndScore
.apply("ConvertToRow", ParDo.of(new BuildRowFn()))
- .apply(BigQueryIO.write()
- .to(getTable(teamAndScore.getPipeline(),
- tableName))
+ .apply(BigQueryIO.writeTableRows()
+ .to(getTable(teamAndScore.getPipeline(), tableName))
.withSchema(getSchema())
.withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
.withWriteDisposition(WriteDisposition.WRITE_APPEND));
http://git-wip-us.apache.org/repos/asf/beam/blob/097573ae/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java
----------------------------------------------------------------------
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 2ae357e..d4ca23c 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
@@ -93,6 +93,7 @@ import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider;
import org.apache.beam.sdk.runners.PipelineRunner;
+import org.apache.beam.sdk.testing.ValueInSingleWindow;
import org.apache.beam.sdk.transforms.Aggregator;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
@@ -130,6 +131,7 @@ import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.PDone;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.TupleTagList;
+import org.apache.beam.sdk.values.TypeDescriptor;
import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -186,7 +188,11 @@ import org.slf4j.LoggerFactory;
* <h3>Writing</h3>
*
* <p>To write to a BigQuery table, apply a {@link BigQueryIO.Write} transformation.
- * This consumes a {@link PCollection} of {@link TableRow TableRows} as input.
+ * This consumes either a {@link PCollection} of {@link TableRow TableRows} as input when using
+ * {@link BigQueryIO#writeTableRows()} or of a user-defined type when using
+ * {@link BigQueryIO#write()}. When using a user-defined type, a function must be provided to
+ * turn this type into a {@link TableRow} using
+ * {@link BigQueryIO.Write#withFormatFunction(SerializableFunction)}.
* <pre>{@code
* PCollection<TableRow> quotes = ...
*
@@ -195,7 +201,7 @@ import org.slf4j.LoggerFactory;
* fields.add(new TableFieldSchema().setName("quote").setType("STRING"));
* TableSchema schema = new TableSchema().setFields(fields);
*
- * quotes.apply(BigQueryIO.Write
+ * quotes.apply(BigQueryIO.writeTableRows()
* .to("my-project:output.output_table")
* .withSchema(schema)
* .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE));
@@ -434,6 +440,19 @@ public class BigQueryIO {
return NestedValueProvider.of(table, new TableRefToTableSpec());
}
+
+ /**
+ * A formatting function that maps a TableRow to itself. This allows sending a
+ * {@code PCollection<TableRow>} directly to BigQueryIO.Write.
+ */
+ private static final SerializableFunction<TableRow, TableRow> IDENTITY_FORMATTER =
+ new SerializableFunction<TableRow, TableRow>() {
+ @Override
+ public TableRow apply(TableRow input) {
+ return input;
+ }
+ };
+
/**
* A {@link PTransform} that reads from a BigQuery table and returns a
* {@link PCollection} of {@link TableRow TableRows} containing each of the rows of the table.
@@ -527,7 +546,7 @@ public class BigQueryIO {
}
/**
- * Same as {@code from(String)}, but with a {@link ValueProvider}.
+ * Same as {@code fromQuery(String)}, but with a {@link ValueProvider}.
*/
public Read fromQuery(ValueProvider<String> query) {
ensureFromNotCalledYet();
@@ -535,7 +554,7 @@ public class BigQueryIO {
}
/**
- * Reads a BigQuery table specified as a {@link TableReference} object.
+ * Read from table specified by a {@link TableReference}.
*/
public Read from(TableReference table) {
return from(StaticValueProvider.of(toTableSpec(table)));
@@ -1352,8 +1371,9 @@ public class BigQueryIO {
/////////////////////////////////////////////////////////////////////////////
/**
- * A {@link PTransform} that writes a {@link PCollection} containing {@link TableRow TableRows}
- * to a BigQuery table.
+ * A {@link PTransform} that writes a {@link PCollection} to a BigQuery table. A formatting
+ * function must be provided to convert each input element into a {@link TableRow} using
+ * {@link Write#withFormatFunction(SerializableFunction)}.
*
* <p>In BigQuery, each table has an encosing dataset. The dataset being written must already
* exist.
@@ -1379,8 +1399,8 @@ public class BigQueryIO {
* }
* }}</pre>
*/
- public static Write write() {
- return new AutoValue_BigQueryIO_Write.Builder()
+ public static <T> Write<T> write() {
+ return new AutoValue_BigQueryIO_Write.Builder<T>()
.setValidate(true)
.setBigQueryServices(new BigQueryServicesImpl())
.setCreateDisposition(Write.CreateDisposition.CREATE_IF_NEEDED)
@@ -1388,9 +1408,17 @@ public class BigQueryIO {
.build();
}
+ /**
+ * A {@link PTransform} that writes a {@link PCollection} containing {@link TableRow TableRows}
+ * to a BigQuery table.
+ */
+ public static Write<TableRow> writeTableRows() {
+ return BigQueryIO.<TableRow>write().withFormatFunction(IDENTITY_FORMATTER);
+ }
+
/** Implementation of {@link #write}. */
@AutoValue
- public abstract static class Write extends PTransform<PCollection<TableRow>, PDone> {
+ public abstract static class Write<T> extends PTransform<PCollection<T>, PDone> {
@VisibleForTesting
// Maximum number of files in a single partition.
static final int MAX_NUM_FILES = 10000;
@@ -1407,7 +1435,9 @@ public class BigQueryIO {
private static final int LOAD_JOB_POLL_MAX_RETRIES = Integer.MAX_VALUE;
@Nullable abstract ValueProvider<String> getJsonTableRef();
- @Nullable abstract SerializableFunction<BoundedWindow, TableReference> getTableRefFunction();
+ @Nullable abstract SerializableFunction<ValueInSingleWindow<T>, TableReference>
+ getTableRefFunction();
+ @Nullable abstract SerializableFunction<T, TableRow> getFormatFunction();
/** Table schema. The schema is required only if the table does not exist. */
@Nullable abstract ValueProvider<String> getJsonSchema();
abstract CreateDisposition getCreateDisposition();
@@ -1417,21 +1447,22 @@ public class BigQueryIO {
abstract boolean getValidate();
abstract BigQueryServices getBigQueryServices();
- abstract Builder toBuilder();
+ abstract Builder<T> toBuilder();
@AutoValue.Builder
- abstract static class Builder {
- abstract Builder setJsonTableRef(ValueProvider<String> jsonTableRef);
- abstract Builder setTableRefFunction(
- SerializableFunction<BoundedWindow, TableReference> tableRefFunction);
- abstract Builder setJsonSchema(ValueProvider<String> jsonSchema);
- abstract Builder setCreateDisposition(CreateDisposition createDisposition);
- abstract Builder setWriteDisposition(WriteDisposition writeDisposition);
- abstract Builder setTableDescription(String tableDescription);
- abstract Builder setValidate(boolean validate);
- abstract Builder setBigQueryServices(BigQueryServices bigQueryServices);
+ abstract static class Builder<T> {
+ abstract Builder<T> setJsonTableRef(ValueProvider<String> jsonTableRef);
+ abstract Builder<T> setTableRefFunction(
+ SerializableFunction<ValueInSingleWindow<T>, TableReference> tableRefFunction);
+ abstract Builder<T> setFormatFunction(SerializableFunction<T, TableRow> formatFunction);
+ abstract Builder<T> setJsonSchema(ValueProvider<String> jsonSchema);
+ abstract Builder<T> setCreateDisposition(CreateDisposition createDisposition);
+ abstract Builder<T> setWriteDisposition(WriteDisposition writeDisposition);
+ abstract Builder<T> setTableDescription(String tableDescription);
+ abstract Builder<T> setValidate(boolean validate);
+ abstract Builder<T> setBigQueryServices(BigQueryServices bigQueryServices);
- abstract Write build();
+ abstract Write<T> build();
}
/**
@@ -1510,17 +1541,17 @@ public class BigQueryIO {
/**
* Writes to the given table, specified in the format described in {@link #parseTableSpec}.
*/
- public Write to(String tableSpec) {
+ public Write<T> to(String tableSpec) {
return to(StaticValueProvider.of(tableSpec));
}
/** Writes to the given table, specified as a {@link TableReference}. */
- public Write to(TableReference table) {
+ public Write<T> to(TableReference table) {
return to(StaticValueProvider.of(toTableSpec(table)));
}
/** Same as {@link #to(String)}, but with a {@link ValueProvider}. */
- public Write to(ValueProvider<String> tableSpec) {
+ public Write<T> to(ValueProvider<String> tableSpec) {
ensureToNotCalledYet();
return toBuilder()
.setJsonTableRef(
@@ -1531,43 +1562,41 @@ public class BigQueryIO {
}
/**
- * Creates a write transformation from a function that maps windows to table specifications.
- * Each time a new window is encountered, this function will be called and the resulting table
- * will be created. Records within that window will be written to the associated table.
- *
- * <p>See {@link #parseTableSpec(String)} for the format that {@code tableSpecFunction} should
- * return.
- *
- * <p>{@code tableSpecFunction} should be deterministic. When given the same window, it should
- * always return the same table specification.
+ * Writes to table specified by the specified table function. The table is a function of
+ * {@link ValueInSingleWindow}, so can be determined by the value or by the window.
*/
- public Write to(SerializableFunction<BoundedWindow, String> tableSpecFunction) {
- return toTableReference(new TranslateTableSpecFunction(tableSpecFunction));
+ public Write<T> to(SerializableFunction<ValueInSingleWindow<T>, String> tableSpecFunction) {
+ return toTableReference(new TranslateTableSpecFunction<T>(tableSpecFunction));
}
/**
- * Creates a write transformation from a function that maps windows to {@link TableReference}
- * objects.
- *
- * <p>{@code tableRefFunction} should be deterministic. When given the same window, it should
- * always return the same table reference.
+ * Like {@link BigQueryIO.Write#to(SerializableFunction)}, but the function returns a
+ * {@link TableReference} instead of a string table specification.
*/
- private Write toTableReference(
- SerializableFunction<BoundedWindow, TableReference> tableRefFunction) {
+ private Write<T> toTableReference(
+ SerializableFunction<ValueInSingleWindow<T>, TableReference> tableRefFunction) {
ensureToNotCalledYet();
return toBuilder().setTableRefFunction(tableRefFunction).build();
}
- private static class TranslateTableSpecFunction implements
- SerializableFunction<BoundedWindow, TableReference> {
- private SerializableFunction<BoundedWindow, String> tableSpecFunction;
+ /**
+ * Formats the user's type into a {@link TableRow} to be written to BigQuery.
+ */
+ public Write<T> withFormatFunction(SerializableFunction<T, TableRow> formatFunction) {
+ return toBuilder().setFormatFunction(formatFunction).build();
+ }
+
+ private static class TranslateTableSpecFunction<T> implements
+ SerializableFunction<ValueInSingleWindow<T>, TableReference> {
+ private SerializableFunction<ValueInSingleWindow<T>, String> tableSpecFunction;
- TranslateTableSpecFunction(SerializableFunction<BoundedWindow, String> tableSpecFunction) {
+ TranslateTableSpecFunction(
+ SerializableFunction<ValueInSingleWindow<T>, String> tableSpecFunction) {
this.tableSpecFunction = tableSpecFunction;
}
@Override
- public TableReference apply(BoundedWindow value) {
+ public TableReference apply(ValueInSingleWindow<T> value) {
return parseTableSpec(tableSpecFunction.apply(value));
}
}
@@ -1579,43 +1608,43 @@ public class BigQueryIO {
* exist, and {@link CreateDisposition} is set to
* {@link CreateDisposition#CREATE_IF_NEEDED}.
*/
- public Write withSchema(TableSchema schema) {
+ public Write<T> withSchema(TableSchema schema) {
return toBuilder()
.setJsonSchema(StaticValueProvider.of(toJsonString(schema)))
.build();
}
/**
- * Like {@link #withSchema(TableSchema)}, but with a {@link ValueProvider}.
+ * Use the specified schema for rows to be written.
*/
- public Write withSchema(ValueProvider<TableSchema> schema) {
+ public Write<T> withSchema(ValueProvider<TableSchema> schema) {
return toBuilder()
.setJsonSchema(NestedValueProvider.of(schema, new TableSchemaToJsonSchema()))
.build();
}
/** Specifies whether the table should be created if it does not exist. */
- public Write withCreateDisposition(CreateDisposition createDisposition) {
+ public Write<T> withCreateDisposition(CreateDisposition createDisposition) {
return toBuilder().setCreateDisposition(createDisposition).build();
}
/** Specifies what to do with existing data in the table, in case the table already exists. */
- public Write withWriteDisposition(WriteDisposition writeDisposition) {
+ public Write<T> withWriteDisposition(WriteDisposition writeDisposition) {
return toBuilder().setWriteDisposition(writeDisposition).build();
}
/** Specifies the table description. */
- public Write withTableDescription(String tableDescription) {
+ public Write<T> withTableDescription(String tableDescription) {
return toBuilder().setTableDescription(tableDescription).build();
}
/** Disables BigQuery table validation. */
- public Write withoutValidation() {
+ public Write<T> withoutValidation() {
return toBuilder().setValidate(false).build();
}
@VisibleForTesting
- Write withTestServices(BigQueryServices testServices) {
+ Write<T> withTestServices(BigQueryServices testServices) {
return toBuilder().setBigQueryServices(testServices).build();
}
@@ -1640,7 +1669,7 @@ public class BigQueryIO {
}
@Override
- public void validate(PCollection<TableRow> input) {
+ public void validate(PCollection<T> input) {
BigQueryOptions options = input.getPipeline().getOptions().as(BigQueryOptions.class);
// Exactly one of the table and table reference can be configured.
@@ -1652,6 +1681,10 @@ public class BigQueryIO {
"Cannot set both a table reference and a table function for a BigQueryIO.Write"
+ " transform");
+ checkArgument(getFormatFunction() != null,
+ "A function must be provided to convert type into a TableRow. "
+ + "use BigQueryIO.Write.withFormatFunction to provide a formatting function.");
+
// Require a schema if creating one or more tables.
checkArgument(
getCreateDisposition() != CreateDisposition.CREATE_IF_NEEDED || getJsonSchema() != null,
@@ -1714,14 +1747,14 @@ public class BigQueryIO {
}
@Override
- public PDone expand(PCollection<TableRow> input) {
+ public PDone expand(PCollection<T> input) {
Pipeline p = input.getPipeline();
BigQueryOptions options = p.getOptions().as(BigQueryOptions.class);
// When writing an Unbounded PCollection, or when a tablespec function is defined, we use
// StreamWithDeDup and BigQuery's streaming import API.
if (input.isBounded() == IsBounded.UNBOUNDED || getTableRefFunction() != null) {
- return input.apply(new StreamWithDeDup(this));
+ return input.apply(new StreamWithDeDup<T>(this));
}
ValueProvider<TableReference> table = getTableWithDefaultProject(options);
@@ -1754,12 +1787,23 @@ public class BigQueryIO {
}))
.apply(View.<String>asSingleton());
- PCollection<TableRow> inputInGlobalWindow =
+ PCollection<T> typedInputInGlobalWindow =
input.apply(
- Window.<TableRow>into(new GlobalWindows())
+ Window.<T>into(new GlobalWindows())
.triggering(DefaultTrigger.of())
.discardingFiredPanes());
+ // Avoid applying the formatFunction if it is the identity formatter.
+ PCollection<TableRow> inputInGlobalWindow;
+ if (getFormatFunction() == IDENTITY_FORMATTER) {
+ inputInGlobalWindow = (PCollection<TableRow>) typedInputInGlobalWindow;
+ } else {
+ inputInGlobalWindow = typedInputInGlobalWindow
+ .apply(MapElements.via(getFormatFunction())
+ .withOutputType(new TypeDescriptor<TableRow>() {
+ }));
+ }
+ // PCollection of filename, file byte size.
PCollection<KV<String, Long>> results = inputInGlobalWindow
.apply("WriteBundles",
ParDo.of(new WriteBundles(tempFilePrefix)));
@@ -1769,6 +1813,8 @@ public class BigQueryIO {
TupleTag<KV<Long, List<String>>> singlePartitionTag =
new TupleTag<KV<Long, List<String>>>("singlePartitionTag") {};
+ // Turn the list of files and record counts in a PCollectionView that can be used as a
+ // side input.
PCollectionView<Iterable<KV<String, Long>>> resultsView = results
.apply("ResultsView", View.<KV<String, Long>>asIterable());
PCollectionTuple partitions = singleton.apply(ParDo
@@ -1779,7 +1825,9 @@ public class BigQueryIO {
.withSideInputs(resultsView)
.withOutputTags(multiPartitionsTag, TupleTagList.of(singlePartitionTag)));
- // Write multiple partitions to separate temporary tables
+ // If WriteBundles produced more than MAX_NUM_FILES files or MAX_SIZE_BYTES bytes, then
+ // the import needs to be split into multiple partitions, and those partitions will be
+ // specified in multiPartitionsTag.
PCollection<String> tempTables = partitions.get(multiPartitionsTag)
.apply("MultiPartitionsGroupByKey", GroupByKey.<Long, List<String>>create())
.apply("MultiPartitionsWriteTables", ParDo.of(new WriteTables(
@@ -1944,6 +1992,7 @@ public class BigQueryIO {
NestedValueProvider.of(getJsonTableRef(), new JsonTableRefToTableRef());
}
+
static class TableRowWriter {
private static final Coder<TableRow> CODER = TableRowJsonCoder.of();
private static final byte[] NEWLINE = "\n".getBytes(StandardCharsets.UTF_8);
@@ -2652,20 +2701,24 @@ public class BigQueryIO {
* id is created by concatenating this randomUUID with a sequential number.
*/
@VisibleForTesting
- static class TagWithUniqueIdsAndTable
- extends DoFn<TableRow, KV<ShardedKey<String>, TableRowInfo>> {
+ static class TagWithUniqueIdsAndTable<T>
+ extends DoFn<T, KV<ShardedKey<String>, TableRowInfo>> {
/** TableSpec to write to. */
private final ValueProvider<String> tableSpec;
- /** User function mapping windows to {@link TableReference} in JSON. */
- private final SerializableFunction<BoundedWindow, TableReference> tableRefFunction;
+ /** User function mapping windowed values to {@link TableReference} in JSON. */
+ private final SerializableFunction<ValueInSingleWindow<T>, TableReference> tableRefFunction;
+
+ /** User function mapping user type to a TableRow. */
+ private final SerializableFunction<T, TableRow> formatFunction;
private transient String randomUUID;
private transient long sequenceNo = 0L;
TagWithUniqueIdsAndTable(BigQueryOptions options,
ValueProvider<TableReference> table,
- SerializableFunction<BoundedWindow, TableReference> tableRefFunction) {
+ SerializableFunction<ValueInSingleWindow<T>, TableReference> tableRefFunction,
+ SerializableFunction<T, TableRow> formatFunction) {
checkArgument(table == null ^ tableRefFunction == null,
"Exactly one of table or tableRefFunction should be set");
if (table != null) {
@@ -2681,6 +2734,7 @@ public class BigQueryIO {
tableSpec = null;
}
this.tableRefFunction = tableRefFunction;
+ this.formatFunction = formatFunction;
}
@@ -2694,12 +2748,13 @@ public class BigQueryIO {
public void processElement(ProcessContext context, BoundedWindow window) throws IOException {
String uniqueId = randomUUID + sequenceNo++;
ThreadLocalRandom randomGenerator = ThreadLocalRandom.current();
- String tableSpec = tableSpecFromWindow(
- context.getPipelineOptions().as(BigQueryOptions.class), window);
+ String tableSpec = tableSpecFromWindowedValue(
+ context.getPipelineOptions().as(BigQueryOptions.class),
+ ValueInSingleWindow.of(context.element(), context.timestamp(), window, context.pane()));
// We output on keys 0-50 to ensure that there's enough batching for
// BigQuery.
context.output(KV.of(ShardedKey.of(tableSpec, randomGenerator.nextInt(0, 50)),
- new TableRowInfo(context.element(), uniqueId)));
+ new TableRowInfo(formatFunction.apply(context.element()), uniqueId)));
}
@Override
@@ -2718,11 +2773,12 @@ public class BigQueryIO {
return tableSpec;
}
- private String tableSpecFromWindow(BigQueryOptions options, BoundedWindow window) {
+ private String tableSpecFromWindowedValue(BigQueryOptions options,
+ ValueInSingleWindow<T> value) {
if (tableSpec != null) {
return tableSpec.get();
} else {
- TableReference table = tableRefFunction.apply(window);
+ TableReference table = tableRefFunction.apply(value);
if (table.getProjectId() == null) {
table.setProjectId(options.getProject());
}
@@ -2737,11 +2793,11 @@ public class BigQueryIO {
* PTransform that performs streaming BigQuery write. To increase consistency,
* it leverages BigQuery best effort de-dup mechanism.
*/
- private static class StreamWithDeDup extends PTransform<PCollection<TableRow>, PDone> {
- private final Write write;
+ private static class StreamWithDeDup<T> extends PTransform<PCollection<T>, PDone> {
+ private final Write<T> write;
/** Constructor. */
- StreamWithDeDup(Write write) {
+ StreamWithDeDup(Write<T> write) {
this.write = write;
}
@@ -2751,7 +2807,7 @@ public class BigQueryIO {
}
@Override
- public PDone expand(PCollection<TableRow> input) {
+ public PDone expand(PCollection<T> input) {
// A naive implementation would be to simply stream data directly to BigQuery.
// However, this could occasionally lead to duplicated data, e.g., when
// a VM that runs this code is restarted and the code is re-run.
@@ -2762,9 +2818,10 @@ public class BigQueryIO {
// To use this mechanism, each input TableRow is tagged with a generated
// unique id, which is then passed to BigQuery and used to ignore duplicates.
- PCollection<KV<ShardedKey<String>, TableRowInfo>> tagged = input.apply(ParDo.of(
- new TagWithUniqueIdsAndTable(input.getPipeline().getOptions().as(BigQueryOptions.class),
- write.getTable(), write.getTableRefFunction())));
+ PCollection<KV<ShardedKey<String>, TableRowInfo>> tagged =
+ input.apply(ParDo.of(new TagWithUniqueIdsAndTable<T>(
+ input.getPipeline().getOptions().as(BigQueryOptions.class), write.getTable(),
+ write.getTableRefFunction(), write.getFormatFunction())));
// To prevent having the same TableRow processed more than once with regenerated
// different unique ids, this implementation relies on "checkpointing", which is
http://git-wip-us.apache.org/repos/asf/beam/blob/097573ae/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
index 8a53d02..cf96acc 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
@@ -30,7 +30,6 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.eq;
@@ -109,6 +108,7 @@ import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.PassThroughThenCleanup;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.PassThroughThenCleanup.CleanupOperation;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Status;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TransformingSource;
+import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.TableRowWriter;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
@@ -132,6 +132,7 @@ import org.apache.beam.sdk.testing.RunnableOnService;
import org.apache.beam.sdk.testing.SourceTestUtils;
import org.apache.beam.sdk.testing.SourceTestUtils.ExpectedSplitOutcome;
import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.ValueInSingleWindow;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnTester;
@@ -705,7 +706,7 @@ public class BigQueryIOTest implements Serializable {
}
private void checkWriteObjectWithValidate(
- BigQueryIO.Write write, String project, String dataset, String table,
+ BigQueryIO.Write<TableRow> write, String project, String dataset, String table,
TableSchema schema, CreateDisposition createDisposition,
WriteDisposition writeDisposition, String tableDescription, boolean validate) {
assertEquals(project, write.getTable().get().getProjectId());
@@ -744,6 +745,7 @@ public class BigQueryIOTest implements Serializable {
BigQueryIO.Read read =
BigQueryIO.read().from("foo.com:project:somedataset.sometable").withoutValidation();
checkReadTableObjectWithValidate(read, "foo.com:project", "somedataset", "sometable", false);
+
}
@Test
@@ -963,7 +965,7 @@ public class BigQueryIOTest implements Serializable {
new TableRow().set("name", "b").set("number", 2),
new TableRow().set("name", "c").set("number", 3))
.withCoder(TableRowJsonCoder.of()))
- .apply(BigQueryIO.write().to("dataset-id.table-id")
+ .apply(BigQueryIO.writeTableRows().to("dataset-id.table-id")
.withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
.withSchema(new TableSchema().setFields(
ImmutableList.of(
@@ -997,7 +999,7 @@ public class BigQueryIOTest implements Serializable {
new TableRow().set("name", "d").set("number", 4))
.withCoder(TableRowJsonCoder.of()))
.setIsBoundedInternal(PCollection.IsBounded.UNBOUNDED)
- .apply(BigQueryIO.write().to("project-id:dataset-id.table-id")
+ .apply(BigQueryIO.writeTableRows().to("project-id:dataset-id.table-id")
.withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
.withSchema(new TableSchema().setFields(
ImmutableList.of(
@@ -1022,12 +1024,11 @@ public class BigQueryIOTest implements Serializable {
* <p>Logically, creates multiple global windows, and the user provides a function that
* decides which global window a value should go into.
*/
- private static class PartitionedGlobalWindows extends
+ private static class PartitionedGlobalWindows<T> extends
+ NonMergingWindowFn<T, PartitionedGlobalWindow> {
+ private SerializableFunction<T, String> extractPartition;
- NonMergingWindowFn<TableRow, PartitionedGlobalWindow> {
- private SerializableFunction<TableRow, String> extractPartition;
-
- public PartitionedGlobalWindows(SerializableFunction<TableRow, String> extractPartition) {
+ public PartitionedGlobalWindows(SerializableFunction<T, String> extractPartition) {
this.extractPartition = extractPartition;
}
@@ -1119,42 +1120,45 @@ public class BigQueryIOTest implements Serializable {
FakeBigQueryServices fakeBqServices = new FakeBigQueryServices()
.withDatasetService(datasetService);
- List<TableRow> inserts = new ArrayList<>();
+ List<Integer> inserts = new ArrayList<>();
for (int i = 0; i < 10; i++) {
- inserts.add(new TableRow().set("name", "number" + i).set("number", i));
+ inserts.add(i);
}
// Create a windowing strategy that puts the input into five different windows depending on
// record value.
- WindowFn<TableRow, PartitionedGlobalWindow> window = new PartitionedGlobalWindows(
- new SerializableFunction<TableRow, String>() {
+ WindowFn<Integer, PartitionedGlobalWindow> window = new PartitionedGlobalWindows(
+ new SerializableFunction<Integer, String>() {
@Override
- public String apply(TableRow value) {
- try {
- int intValue = (Integer) value.get("number") % 5;
- return Integer.toString(intValue);
- } catch (NumberFormatException e) {
- fail(e.toString());
- }
- return value.toString();
+ public String apply(Integer i) {
+ return Integer.toString(i % 5);
}
}
);
- SerializableFunction<BoundedWindow, String> tableFunction =
- new SerializableFunction<BoundedWindow, String>() {
+ SerializableFunction<ValueInSingleWindow<Integer>, String> tableFunction =
+ new SerializableFunction<ValueInSingleWindow<Integer>, String>() {
@Override
- public String apply(BoundedWindow input) {
- return "project-id:dataset-id.table-id-" + ((PartitionedGlobalWindow) input).value;
+ public String apply(ValueInSingleWindow<Integer> input) {
+ PartitionedGlobalWindow window = (PartitionedGlobalWindow) input.getWindow();
+ // Check that we can access the element as well here.
+ checkArgument(window.value.equals(Integer.toString(input.getValue() % 5)),
+ "Incorrect element");
+ return "project-id:dataset-id.table-id-" + window.value;
}
};
Pipeline p = TestPipeline.create(bqOptions);
- p.apply(Create.of(inserts).withCoder(TableRowJsonCoder.of()))
+ p.apply(Create.of(inserts))
.setIsBoundedInternal(PCollection.IsBounded.UNBOUNDED)
- .apply(Window.<TableRow>into(window))
- .apply(BigQueryIO.write()
+ .apply(Window.<Integer>into(window))
+ .apply(BigQueryIO.<Integer>write()
.to(tableFunction)
+ .withFormatFunction(new SerializableFunction<Integer, TableRow>() {
+ @Override
+ public TableRow apply(Integer i) {
+ return new TableRow().set("name", "number" + i).set("number", i);
+ }})
.withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
.withSchema(new TableSchema().setFields(
ImmutableList.of(
@@ -1205,7 +1209,7 @@ public class BigQueryIOTest implements Serializable {
new TableRow().set("name", "b").set("number", 2),
new TableRow().set("name", "c").set("number", 3))
.withCoder(TableRowJsonCoder.of()))
- .apply(BigQueryIO.write().to("project-id:dataset-id.table-id")
+ .apply(BigQueryIO.writeTableRows().to("project-id:dataset-id.table-id")
.withCreateDisposition(CreateDisposition.CREATE_NEVER)
.withTestServices(fakeBqServices)
.withoutValidation());
@@ -1238,7 +1242,7 @@ public class BigQueryIOTest implements Serializable {
new TableRow().set("name", "b").set("number", 2),
new TableRow().set("name", "c").set("number", 3))
.withCoder(TableRowJsonCoder.of()))
- .apply(BigQueryIO.write().to("dataset-id.table-id")
+ .apply(BigQueryIO.writeTableRows().to("dataset-id.table-id")
.withCreateDisposition(CreateDisposition.CREATE_NEVER)
.withTestServices(fakeBqServices)
.withoutValidation());
@@ -1327,8 +1331,8 @@ public class BigQueryIOTest implements Serializable {
@Test
public void testBuildWrite() {
- BigQueryIO.Write write =
- BigQueryIO.write().to("foo.com:project:somedataset.sometable");
+ BigQueryIO.Write<TableRow> write =
+ BigQueryIO.writeTableRows().to("foo.com:project:somedataset.sometable");
checkWriteObject(
write, "foo.com:project", "somedataset", "sometable",
null, CreateDisposition.CREATE_IF_NEEDED, WriteDisposition.WRITE_EMPTY, null);
@@ -1354,7 +1358,7 @@ public class BigQueryIOTest implements Serializable {
options.as(StreamingOptions.class).setStreaming(streaming);
DisplayDataEvaluator evaluator = DisplayDataEvaluator.create(options);
- BigQueryIO.Write write = BigQueryIO.write()
+ BigQueryIO.Write write = BigQueryIO.writeTableRows()
.to("project:dataset.table")
.withSchema(new TableSchema().set("col1", "type1").set("col2", "type2"))
.withTestServices(new FakeBigQueryServices()
@@ -1375,7 +1379,8 @@ public class BigQueryIOTest implements Serializable {
// This test just checks that using withoutValidation will not trigger object
// construction errors.
BigQueryIO.Write write =
- BigQueryIO.write().to("foo.com:project:somedataset.sometable").withoutValidation();
+ BigQueryIO.<TableRow>write().to("foo.com:project:somedataset.sometable")
+ .withoutValidation();
checkWriteObjectWithValidate(
write,
"foo.com:project",
@@ -1390,10 +1395,11 @@ public class BigQueryIOTest implements Serializable {
@Test
public void testBuildWriteDefaultProject() {
- BigQueryIO.Write write = BigQueryIO.write().to("somedataset.sometable");
+ BigQueryIO.Write<TableRow> write = BigQueryIO.writeTableRows().to("somedataset.sometable");
checkWriteObject(
write, null, "somedataset", "sometable",
- null, CreateDisposition.CREATE_IF_NEEDED, WriteDisposition.WRITE_EMPTY, null);
+ null, CreateDisposition.CREATE_IF_NEEDED, WriteDisposition.WRITE_EMPTY,
+ null);
}
@Test
@@ -1402,7 +1408,7 @@ public class BigQueryIOTest implements Serializable {
.setProjectId("foo.com:project")
.setDatasetId("somedataset")
.setTableId("sometable");
- BigQueryIO.Write write = BigQueryIO.write().to(table);
+ BigQueryIO.Write<TableRow> write = BigQueryIO.writeTableRows().to(table);
checkWriteObject(
write, "foo.com:project", "somedataset", "sometable",
null, CreateDisposition.CREATE_IF_NEEDED, WriteDisposition.WRITE_EMPTY, null);
@@ -1411,8 +1417,8 @@ public class BigQueryIOTest implements Serializable {
@Test
public void testBuildWriteWithSchema() {
TableSchema schema = new TableSchema();
- BigQueryIO.Write write =
- BigQueryIO.write().to("foo.com:project:somedataset.sometable").withSchema(schema);
+ BigQueryIO.Write<TableRow> write =
+ BigQueryIO.<TableRow>write().to("foo.com:project:somedataset.sometable").withSchema(schema);
checkWriteObject(
write, "foo.com:project", "somedataset", "sometable",
schema, CreateDisposition.CREATE_IF_NEEDED, WriteDisposition.WRITE_EMPTY, null);
@@ -1420,7 +1426,7 @@ public class BigQueryIOTest implements Serializable {
@Test
public void testBuildWriteWithCreateDispositionNever() {
- BigQueryIO.Write write = BigQueryIO.write()
+ BigQueryIO.Write<TableRow> write = BigQueryIO.<TableRow>write()
.to("foo.com:project:somedataset.sometable")
.withCreateDisposition(CreateDisposition.CREATE_NEVER);
checkWriteObject(
@@ -1430,7 +1436,7 @@ public class BigQueryIOTest implements Serializable {
@Test
public void testBuildWriteWithCreateDispositionIfNeeded() {
- BigQueryIO.Write write = BigQueryIO.write()
+ BigQueryIO.Write<TableRow> write = BigQueryIO.writeTableRows()
.to("foo.com:project:somedataset.sometable")
.withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED);
checkWriteObject(
@@ -1440,7 +1446,7 @@ public class BigQueryIOTest implements Serializable {
@Test
public void testBuildWriteWithWriteDispositionTruncate() {
- BigQueryIO.Write write = BigQueryIO.write()
+ BigQueryIO.Write<TableRow> write = BigQueryIO.<TableRow>write()
.to("foo.com:project:somedataset.sometable")
.withWriteDisposition(WriteDisposition.WRITE_TRUNCATE);
checkWriteObject(
@@ -1450,7 +1456,7 @@ public class BigQueryIOTest implements Serializable {
@Test
public void testBuildWriteWithWriteDispositionAppend() {
- BigQueryIO.Write write = BigQueryIO.write()
+ BigQueryIO.Write<TableRow> write = BigQueryIO.writeTableRows()
.to("foo.com:project:somedataset.sometable")
.withWriteDisposition(WriteDisposition.WRITE_APPEND);
checkWriteObject(
@@ -1460,7 +1466,7 @@ public class BigQueryIOTest implements Serializable {
@Test
public void testBuildWriteWithWriteDispositionEmpty() {
- BigQueryIO.Write write = BigQueryIO.write()
+ BigQueryIO.Write<TableRow> write = BigQueryIO.<TableRow>write()
.to("foo.com:project:somedataset.sometable")
.withWriteDisposition(WriteDisposition.WRITE_EMPTY);
checkWriteObject(
@@ -1471,7 +1477,7 @@ public class BigQueryIOTest implements Serializable {
@Test
public void testBuildWriteWithWriteWithTableDescription() {
final String tblDescription = "foo bar table";
- BigQueryIO.Write write = BigQueryIO.write()
+ BigQueryIO.Write<TableRow> write = BigQueryIO.writeTableRows()
.to("foo.com:project:somedataset.sometable")
.withTableDescription(tblDescription);
checkWriteObject(
@@ -1491,7 +1497,7 @@ public class BigQueryIOTest implements Serializable {
TableSchema schema = new TableSchema().set("col1", "type1").set("col2", "type2");
final String tblDescription = "foo bar table";
- BigQueryIO.Write write = BigQueryIO.write()
+ BigQueryIO.Write<TableRow> write = BigQueryIO.writeTableRows()
.to(tableSpec)
.withSchema(schema)
.withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
@@ -1556,7 +1562,7 @@ public class BigQueryIOTest implements Serializable {
.or(Matchers.containsString("BigQuery dataset not found for table")));
tableRows
.apply(
- BigQueryIO.write().to(tableRef)
+ BigQueryIO.writeTableRows().to(tableRef)
.withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
.withSchema(new TableSchema())
.withTestServices(fakeBqServices));
@@ -1603,7 +1609,7 @@ public class BigQueryIOTest implements Serializable {
}))
.setCoder(TableRowJsonCoder.of());
tableRows
- .apply(BigQueryIO.write().to(tableRef)
+ .apply(BigQueryIO.writeTableRows().to(tableRef)
.withCreateDisposition(CreateDisposition.CREATE_NEVER)
.withoutValidation());
}
@@ -1674,8 +1680,10 @@ public class BigQueryIOTest implements Serializable {
@Test
public void testBigQueryIOGetName() {
- assertEquals("BigQueryIO.Read", BigQueryIO.read().from("somedataset.sometable").getName());
- assertEquals("BigQueryIO.Write", BigQueryIO.write().to("somedataset.sometable").getName());
+ assertEquals("BigQueryIO.Read",
+ BigQueryIO.read().from("somedataset.sometable").getName());
+ assertEquals("BigQueryIO.Write",
+ BigQueryIO.<TableRow>write().to("somedataset.sometable").getName());
}
@Test
@@ -1686,7 +1694,7 @@ public class BigQueryIOTest implements Serializable {
thrown.expectMessage("no schema was provided");
p
.apply(Create.empty(TableRowJsonCoder.of()))
- .apply(BigQueryIO.write()
+ .apply(BigQueryIO.writeTableRows()
.to("dataset.table")
.withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED));
}
@@ -2343,7 +2351,7 @@ public class BigQueryIOTest implements Serializable {
BigQueryOptions bqOptions = options.as(BigQueryOptions.class);
bqOptions.setTempLocation("gs://testbucket/testdir");
Pipeline pipeline = TestPipeline.create(options);
- BigQueryIO.Write write = BigQueryIO.write()
+ BigQueryIO.Write<TableRow> write = BigQueryIO.writeTableRows()
.to(options.getOutputTable())
.withSchema(NestedValueProvider.of(
options.getOutputSchema(), new JsonSchemaToTableSchema()))
@@ -2359,11 +2367,11 @@ public class BigQueryIOTest implements Serializable {
public void testTagWithUniqueIdsAndTableProjectNotNullWithNvp() {
BigQueryOptions bqOptions = PipelineOptionsFactory.as(BigQueryOptions.class);
bqOptions.setProject("project");
- BigQueryIO.TagWithUniqueIdsAndTable tag =
- new BigQueryIO.TagWithUniqueIdsAndTable(
+ BigQueryIO.TagWithUniqueIdsAndTable<TableRow> tag =
+ new BigQueryIO.TagWithUniqueIdsAndTable<TableRow>(
bqOptions, NestedValueProvider.of(
StaticValueProvider.of("data_set.table_name"),
- new BigQueryIO.TableSpecToTableRef()), null);
+ new BigQueryIO.TableSpecToTableRef()), null, null);
TableReference table = BigQueryIO.parseTableSpec(tag.getTableSpec().get());
assertNotNull(table.getProjectId());
}
[2/3] beam git commit: Move ValueInSingleWindow from testing package
to values package
Posted by tg...@apache.org.
Move ValueInSingleWindow from testing package to values package
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/b793fe0c
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/b793fe0c
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/b793fe0c
Branch: refs/heads/master
Commit: b793fe0ceba9f5e0a680f2e2fe0480264b546f3c
Parents: 097573a
Author: Reuven Lax <re...@google.com>
Authored: Wed Mar 15 09:05:55 2017 -0700
Committer: Thomas Groh <tg...@google.com>
Committed: Mon Mar 20 09:20:58 2017 -0700
----------------------------------------------------------------------
.../beam/runners/core/SplittableParDoTest.java | 2 +-
.../apache/beam/sdk/testing/GatherAllPanes.java | 1 +
.../org/apache/beam/sdk/testing/PAssert.java | 1 +
.../apache/beam/sdk/testing/PaneExtractors.java | 1 +
.../beam/sdk/testing/ValueInSingleWindow.java | 134 -------------------
.../apache/beam/sdk/transforms/DoFnTester.java | 2 +-
.../beam/sdk/values/ValueInSingleWindow.java | 134 +++++++++++++++++++
.../beam/sdk/testing/GatherAllPanesTest.java | 1 +
.../beam/sdk/testing/PaneExtractorsTest.java | 1 +
.../testing/ValueInSingleWindowCoderTest.java | 1 +
.../beam/sdk/io/gcp/bigquery/BigQueryIO.java | 2 +-
.../sdk/io/gcp/bigquery/BigQueryIOTest.java | 3 +-
12 files changed, 144 insertions(+), 139 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/b793fe0c/runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoTest.java
index 9402502..af547c2 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoTest.java
@@ -41,7 +41,6 @@ import org.apache.beam.sdk.coders.BigEndianIntegerCoder;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.SerializableCoder;
import org.apache.beam.sdk.testing.TestPipeline;
-import org.apache.beam.sdk.testing.ValueInSingleWindow;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnTester;
@@ -58,6 +57,7 @@ import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TimestampedValue;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.TupleTagList;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.junit.Rule;
http://git-wip-us.apache.org/repos/asf/beam/blob/b793fe0c/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/GatherAllPanes.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/GatherAllPanes.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/GatherAllPanes.java
index bf2cd0b..6b24d95 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/GatherAllPanes.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/GatherAllPanes.java
@@ -31,6 +31,7 @@ import org.apache.beam.sdk.util.IdentityWindowFn;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
/**
* Gathers all panes of each window into exactly one output.
http://git-wip-us.apache.org/repos/asf/beam/blob/b793fe0c/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java
index f6a409a..412753c 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java
@@ -72,6 +72,7 @@ import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionList;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.PDone;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
import org.joda.time.Duration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
http://git-wip-us.apache.org/repos/asf/beam/blob/b793fe0c/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PaneExtractors.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PaneExtractors.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PaneExtractors.java
index dd1fac9..f88efcb 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PaneExtractors.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PaneExtractors.java
@@ -27,6 +27,7 @@ import org.apache.beam.sdk.transforms.windowing.PaneInfo;
import org.apache.beam.sdk.transforms.windowing.PaneInfo.Timing;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
/**
* {@link PTransform PTransforms} which take an {@link Iterable} of {@link ValueInSingleWindow
http://git-wip-us.apache.org/repos/asf/beam/blob/b793fe0c/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/ValueInSingleWindow.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/ValueInSingleWindow.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/ValueInSingleWindow.java
deleted file mode 100644
index b746f6d..0000000
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/ValueInSingleWindow.java
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
- * 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.beam.sdk.testing;
-
-import static com.google.common.base.Preconditions.checkArgument;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.auto.value.AutoValue;
-import com.google.common.collect.ImmutableList;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.util.List;
-import javax.annotation.Nullable;
-import org.apache.beam.sdk.coders.InstantCoder;
-import org.apache.beam.sdk.coders.StandardCoder;
-import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
-import org.apache.beam.sdk.transforms.windowing.PaneInfo;
-import org.apache.beam.sdk.util.PropertyNames;
-import org.joda.time.Instant;
-
-/**
- * An immutable tuple of value, timestamp, window, and pane.
- *
- * @param <T> the type of the value
- */
-@AutoValue
-public abstract class ValueInSingleWindow<T> {
- /** Returns the value of this {@code ValueInSingleWindow}. */
- @Nullable
- public abstract T getValue();
-
- /** Returns the timestamp of this {@code ValueInSingleWindow}. */
- public abstract Instant getTimestamp();
-
- /** Returns the window of this {@code ValueInSingleWindow}. */
- public abstract BoundedWindow getWindow();
-
- /** Returns the pane of this {@code ValueInSingleWindow} in its window. */
- public abstract PaneInfo getPane();
-
- public static <T> ValueInSingleWindow<T> of(
- T value, Instant timestamp, BoundedWindow window, PaneInfo paneInfo) {
- return new AutoValue_ValueInSingleWindow<>(value, timestamp, window, paneInfo);
- }
-
- /** A coder for {@link ValueInSingleWindow}. */
- public static class Coder<T> extends StandardCoder<ValueInSingleWindow<T>> {
- private final org.apache.beam.sdk.coders.Coder<T> valueCoder;
- private final org.apache.beam.sdk.coders.Coder<BoundedWindow> windowCoder;
-
- public static <T> Coder<T> of(
- org.apache.beam.sdk.coders.Coder<T> valueCoder,
- org.apache.beam.sdk.coders.Coder<? extends BoundedWindow> windowCoder) {
- return new Coder<>(valueCoder, windowCoder);
- }
-
- @JsonCreator
- public static <T> Coder<T> of(
- @JsonProperty(PropertyNames.COMPONENT_ENCODINGS)
- List<org.apache.beam.sdk.coders.Coder<?>> components) {
- checkArgument(components.size() == 2, "Expecting 2 components, got %s", components.size());
- @SuppressWarnings("unchecked")
- org.apache.beam.sdk.coders.Coder<T> valueCoder =
- (org.apache.beam.sdk.coders.Coder<T>) components.get(0);
- @SuppressWarnings("unchecked")
- org.apache.beam.sdk.coders.Coder<BoundedWindow> windowCoder =
- (org.apache.beam.sdk.coders.Coder<BoundedWindow>) components.get(1);
- return new Coder<>(valueCoder, windowCoder);
- }
-
- @SuppressWarnings({"unchecked", "rawtypes"})
- Coder(
- org.apache.beam.sdk.coders.Coder<T> valueCoder,
- org.apache.beam.sdk.coders.Coder<? extends BoundedWindow> windowCoder) {
- this.valueCoder = valueCoder;
- this.windowCoder = (org.apache.beam.sdk.coders.Coder) windowCoder;
- }
-
- @Override
- public void encode(ValueInSingleWindow<T> windowedElem, OutputStream outStream, Context context)
- throws IOException {
- Context nestedContext = context.nested();
- InstantCoder.of().encode(windowedElem.getTimestamp(), outStream, nestedContext);
- windowCoder.encode(windowedElem.getWindow(), outStream, nestedContext);
- PaneInfo.PaneInfoCoder.INSTANCE.encode(windowedElem.getPane(), outStream, nestedContext);
- valueCoder.encode(windowedElem.getValue(), outStream, context);
- }
-
- @Override
- public ValueInSingleWindow<T> decode(InputStream inStream, Context context) throws IOException {
- Context nestedContext = context.nested();
- Instant timestamp = InstantCoder.of().decode(inStream, nestedContext);
- BoundedWindow window = windowCoder.decode(inStream, nestedContext);
- PaneInfo pane = PaneInfo.PaneInfoCoder.INSTANCE.decode(inStream, nestedContext);
- T value = valueCoder.decode(inStream, context);
- return new AutoValue_ValueInSingleWindow<>(value, timestamp, window, pane);
- }
-
- @Override
- public List<? extends org.apache.beam.sdk.coders.Coder<?>> getCoderArguments() {
- // Coder arguments are coders for the type parameters of the coder - i.e. only T.
- return ImmutableList.of(valueCoder);
- }
-
- @Override
- public List<? extends org.apache.beam.sdk.coders.Coder<?>> getComponents() {
- // Coder components are all inner coders that it uses - i.e. both T and BoundedWindow.
- return ImmutableList.of(valueCoder, windowCoder);
- }
-
- @Override
- public void verifyDeterministic() throws NonDeterministicException {
- valueCoder.verifyDeterministic();
- windowCoder.verifyDeterministic();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/beam/blob/b793fe0c/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
index cc5281c..01c639a 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
@@ -33,7 +33,6 @@ import java.util.Map;
import org.apache.beam.sdk.annotations.Experimental;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.apache.beam.sdk.testing.ValueInSingleWindow;
import org.apache.beam.sdk.transforms.Combine.CombineFn;
import org.apache.beam.sdk.transforms.DoFn.OnTimerContext;
import org.apache.beam.sdk.transforms.reflect.DoFnInvoker;
@@ -51,6 +50,7 @@ import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TimestampedValue;
import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
import org.joda.time.Instant;
/**
http://git-wip-us.apache.org/repos/asf/beam/blob/b793fe0c/sdks/java/core/src/main/java/org/apache/beam/sdk/values/ValueInSingleWindow.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/ValueInSingleWindow.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/ValueInSingleWindow.java
new file mode 100644
index 0000000..c94190f
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/ValueInSingleWindow.java
@@ -0,0 +1,134 @@
+/*
+ * 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.beam.sdk.values;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.auto.value.AutoValue;
+import com.google.common.collect.ImmutableList;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.InstantCoder;
+import org.apache.beam.sdk.coders.StandardCoder;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.util.PropertyNames;
+import org.joda.time.Instant;
+
+/**
+ * An immutable tuple of value, timestamp, window, and pane.
+ *
+ * @param <T> the type of the value
+ */
+@AutoValue
+public abstract class ValueInSingleWindow<T> {
+ /** Returns the value of this {@code ValueInSingleWindow}. */
+ @Nullable
+ public abstract T getValue();
+
+ /** Returns the timestamp of this {@code ValueInSingleWindow}. */
+ public abstract Instant getTimestamp();
+
+ /** Returns the window of this {@code ValueInSingleWindow}. */
+ public abstract BoundedWindow getWindow();
+
+ /** Returns the pane of this {@code ValueInSingleWindow} in its window. */
+ public abstract PaneInfo getPane();
+
+ public static <T> ValueInSingleWindow<T> of(
+ T value, Instant timestamp, BoundedWindow window, PaneInfo paneInfo) {
+ return new AutoValue_ValueInSingleWindow<>(value, timestamp, window, paneInfo);
+ }
+
+ /** A coder for {@link ValueInSingleWindow}. */
+ public static class Coder<T> extends StandardCoder<ValueInSingleWindow<T>> {
+ private final org.apache.beam.sdk.coders.Coder<T> valueCoder;
+ private final org.apache.beam.sdk.coders.Coder<BoundedWindow> windowCoder;
+
+ public static <T> Coder<T> of(
+ org.apache.beam.sdk.coders.Coder<T> valueCoder,
+ org.apache.beam.sdk.coders.Coder<? extends BoundedWindow> windowCoder) {
+ return new Coder<>(valueCoder, windowCoder);
+ }
+
+ @JsonCreator
+ public static <T> Coder<T> of(
+ @JsonProperty(PropertyNames.COMPONENT_ENCODINGS)
+ List<org.apache.beam.sdk.coders.Coder<?>> components) {
+ checkArgument(components.size() == 2, "Expecting 2 components, got %s", components.size());
+ @SuppressWarnings("unchecked")
+ org.apache.beam.sdk.coders.Coder<T> valueCoder =
+ (org.apache.beam.sdk.coders.Coder<T>) components.get(0);
+ @SuppressWarnings("unchecked")
+ org.apache.beam.sdk.coders.Coder<BoundedWindow> windowCoder =
+ (org.apache.beam.sdk.coders.Coder<BoundedWindow>) components.get(1);
+ return new Coder<>(valueCoder, windowCoder);
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ Coder(
+ org.apache.beam.sdk.coders.Coder<T> valueCoder,
+ org.apache.beam.sdk.coders.Coder<? extends BoundedWindow> windowCoder) {
+ this.valueCoder = valueCoder;
+ this.windowCoder = (org.apache.beam.sdk.coders.Coder) windowCoder;
+ }
+
+ @Override
+ public void encode(ValueInSingleWindow<T> windowedElem, OutputStream outStream, Context context)
+ throws IOException {
+ Context nestedContext = context.nested();
+ InstantCoder.of().encode(windowedElem.getTimestamp(), outStream, nestedContext);
+ windowCoder.encode(windowedElem.getWindow(), outStream, nestedContext);
+ PaneInfo.PaneInfoCoder.INSTANCE.encode(windowedElem.getPane(), outStream, nestedContext);
+ valueCoder.encode(windowedElem.getValue(), outStream, context);
+ }
+
+ @Override
+ public ValueInSingleWindow<T> decode(InputStream inStream, Context context) throws IOException {
+ Context nestedContext = context.nested();
+ Instant timestamp = InstantCoder.of().decode(inStream, nestedContext);
+ BoundedWindow window = windowCoder.decode(inStream, nestedContext);
+ PaneInfo pane = PaneInfo.PaneInfoCoder.INSTANCE.decode(inStream, nestedContext);
+ T value = valueCoder.decode(inStream, context);
+ return new AutoValue_ValueInSingleWindow<>(value, timestamp, window, pane);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getCoderArguments() {
+ // Coder arguments are coders for the type parameters of the coder - i.e. only T.
+ return ImmutableList.of(valueCoder);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getComponents() {
+ // Coder components are all inner coders that it uses - i.e. both T and BoundedWindow.
+ return ImmutableList.of(valueCoder, windowCoder);
+ }
+
+ @Override
+ public void verifyDeterministic() throws NonDeterministicException {
+ valueCoder.verifyDeterministic();
+ windowCoder.verifyDeterministic();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/b793fe0c/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/GatherAllPanesTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/GatherAllPanesTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/GatherAllPanesTest.java
index a96e3f8..969bbc4 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/GatherAllPanesTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/GatherAllPanesTest.java
@@ -35,6 +35,7 @@ import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionList;
import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.junit.Rule;
http://git-wip-us.apache.org/repos/asf/beam/blob/b793fe0c/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PaneExtractorsTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PaneExtractorsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PaneExtractorsTest.java
index 7df2f89..8801bde 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PaneExtractorsTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PaneExtractorsTest.java
@@ -27,6 +27,7 @@ import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
import org.apache.beam.sdk.transforms.windowing.PaneInfo;
import org.apache.beam.sdk.transforms.windowing.PaneInfo.Timing;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
import org.joda.time.Instant;
import org.junit.Rule;
import org.junit.Test;
http://git-wip-us.apache.org/repos/asf/beam/blob/b793fe0c/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/ValueInSingleWindowCoderTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/ValueInSingleWindowCoderTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/ValueInSingleWindowCoderTest.java
index 3cc016e..9ce18d7 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/ValueInSingleWindowCoderTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/ValueInSingleWindowCoderTest.java
@@ -21,6 +21,7 @@ import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.junit.Test;
http://git-wip-us.apache.org/repos/asf/beam/blob/b793fe0c/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java
----------------------------------------------------------------------
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 d4ca23c..e9ea0e0 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
@@ -93,7 +93,6 @@ import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider;
import org.apache.beam.sdk.runners.PipelineRunner;
-import org.apache.beam.sdk.testing.ValueInSingleWindow;
import org.apache.beam.sdk.transforms.Aggregator;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
@@ -132,6 +131,7 @@ import org.apache.beam.sdk.values.PDone;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.TupleTagList;
import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
http://git-wip-us.apache.org/repos/asf/beam/blob/b793fe0c/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
index cf96acc..dcc3800 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
@@ -108,7 +108,6 @@ import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.PassThroughThenCleanup;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.PassThroughThenCleanup.CleanupOperation;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Status;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TransformingSource;
-import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.TableRowWriter;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
@@ -132,7 +131,6 @@ import org.apache.beam.sdk.testing.RunnableOnService;
import org.apache.beam.sdk.testing.SourceTestUtils;
import org.apache.beam.sdk.testing.SourceTestUtils.ExpectedSplitOutcome;
import org.apache.beam.sdk.testing.TestPipeline;
-import org.apache.beam.sdk.testing.ValueInSingleWindow;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnTester;
@@ -161,6 +159,7 @@ import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
import org.hamcrest.CoreMatchers;
import org.hamcrest.Matchers;
import org.joda.time.Instant;
[3/3] beam git commit: This closes #2238
Posted by tg...@apache.org.
This closes #2238
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/6816d987
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/6816d987
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/6816d987
Branch: refs/heads/master
Commit: 6816d987013b8046f571b74763aad21637157896
Parents: d47ae87 b793fe0
Author: Thomas Groh <tg...@google.com>
Authored: Mon Mar 20 09:20:58 2017 -0700
Committer: Thomas Groh <tg...@google.com>
Committed: Mon Mar 20 09:20:58 2017 -0700
----------------------------------------------------------------------
.../beam/examples/complete/AutoComplete.java | 2 +-
.../examples/complete/StreamingWordExtract.java | 2 +-
.../examples/complete/TrafficMaxLaneFlow.java | 2 +-
.../beam/examples/complete/TrafficRoutes.java | 2 +-
.../examples/cookbook/BigQueryTornadoes.java | 10 +-
.../cookbook/CombinePerKeyExamples.java | 2 +-
.../beam/examples/cookbook/FilterExamples.java | 2 +-
.../examples/cookbook/MaxPerKeyExamples.java | 2 +-
.../beam/examples/cookbook/TriggerExample.java | 4 +-
.../complete/game/utils/WriteToBigQuery.java | 10 +-
.../game/utils/WriteWindowedToBigQuery.java | 6 +-
.../beam/runners/core/SplittableParDoTest.java | 2 +-
.../apache/beam/sdk/testing/GatherAllPanes.java | 1 +
.../org/apache/beam/sdk/testing/PAssert.java | 1 +
.../apache/beam/sdk/testing/PaneExtractors.java | 1 +
.../beam/sdk/testing/ValueInSingleWindow.java | 134 ------------
.../apache/beam/sdk/transforms/DoFnTester.java | 2 +-
.../beam/sdk/values/ValueInSingleWindow.java | 134 ++++++++++++
.../beam/sdk/testing/GatherAllPanesTest.java | 1 +
.../beam/sdk/testing/PaneExtractorsTest.java | 1 +
.../testing/ValueInSingleWindowCoderTest.java | 1 +
.../beam/sdk/io/gcp/bigquery/BigQueryIO.java | 215 ++++++++++++-------
.../sdk/io/gcp/bigquery/BigQueryIOTest.java | 117 +++++-----
23 files changed, 362 insertions(+), 292 deletions(-)
----------------------------------------------------------------------