From 998ba7ba369a34654b9e93f662f7b1d9b7d83a58 Mon Sep 17 00:00:00 2001 From: Scott Wegner Date: Wed, 11 May 2016 14:17:10 -0700 Subject: [PATCH 1/2] Add labels for SDK display data --- .../DataflowPipelineTranslatorTest.java | 2 + .../java/org/apache/beam/sdk/io/AvroIO.java | 28 ++++++++---- .../org/apache/beam/sdk/io/BigQueryIO.java | 43 +++++++++++++------ .../io/BoundedReadFromUnboundedSource.java | 9 ++-- .../apache/beam/sdk/io/CompressedSource.java | 9 ++-- .../org/apache/beam/sdk/io/CountingInput.java | 12 ++++-- .../org/apache/beam/sdk/io/DatastoreIO.java | 18 +++++--- .../org/apache/beam/sdk/io/FileBasedSink.java | 3 +- .../apache/beam/sdk/io/FileBasedSource.java | 3 +- .../apache/beam/sdk/io/OffsetBasedSource.java | 9 ++-- .../java/org/apache/beam/sdk/io/PubsubIO.java | 27 ++++++++---- .../java/org/apache/beam/sdk/io/Read.java | 6 ++- .../java/org/apache/beam/sdk/io/TextIO.java | 25 +++++++---- .../java/org/apache/beam/sdk/io/Write.java | 3 +- .../java/org/apache/beam/sdk/io/XmlSink.java | 6 ++- .../org/apache/beam/sdk/io/XmlSource.java | 9 ++-- .../sdk/transforms/ApproximateQuantiles.java | 6 ++- .../sdk/transforms/ApproximateUnique.java | 6 ++- .../apache/beam/sdk/transforms/Combine.java | 15 ++++--- .../beam/sdk/transforms/CombineFns.java | 3 +- .../apache/beam/sdk/transforms/Filter.java | 3 +- .../beam/sdk/transforms/GroupByKey.java | 3 +- .../IntraBundleParallelization.java | 6 ++- .../beam/sdk/transforms/MapElements.java | 3 +- .../org/apache/beam/sdk/transforms/Max.java | 3 +- .../org/apache/beam/sdk/transforms/Min.java | 3 +- .../org/apache/beam/sdk/transforms/ParDo.java | 3 +- .../apache/beam/sdk/transforms/Partition.java | 6 ++- .../apache/beam/sdk/transforms/Sample.java | 6 ++- .../org/apache/beam/sdk/transforms/Top.java | 6 ++- .../transforms/windowing/CalendarWindows.java | 18 +++++--- .../transforms/windowing/FixedWindows.java | 6 ++- .../sdk/transforms/windowing/Sessions.java | 3 +- .../transforms/windowing/SlidingWindows.java | 9 ++-- .../beam/sdk/transforms/windowing/Window.java | 18 +++++--- .../beam/sdk/io/gcp/bigtable/BigtableIO.java | 15 ++++--- 36 files changed, 236 insertions(+), 117 deletions(-) diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java index ed7e67d49f8e..58c6f75cbdf6 100644 --- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java +++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java @@ -936,6 +936,7 @@ public void populateDisplayData(DisplayData.Builder builder) { .build(), ImmutableMap.builder() .put("key", "fn") + .put("label", "Transform Function") .put("type", "JAVA_CLASS") .put("value", fn1.getClass().getName()) .put("shortValue", fn1.getClass().getSimpleName()) @@ -955,6 +956,7 @@ public void populateDisplayData(DisplayData.Builder builder) { ImmutableSet> expectedFn2DisplayData = ImmutableSet.of( ImmutableMap.builder() .put("key", "fn") + .put("label", "Transform Function") .put("type", "JAVA_CLASS") .put("value", fn2.getClass().getName()) .put("shortValue", fn2.getClass().getSimpleName()) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java index 6b9f0100f034..15ea85f61a99 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java @@ -331,8 +331,10 @@ public PCollection apply(PInput input) { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .addIfNotNull(DisplayData.item("filePattern", filepattern)) - .addIfNotDefault(DisplayData.item("validation", validate), true); + .addIfNotNull(DisplayData.item("filePattern", filepattern) + .withLabel("Input File Pattern")) + .addIfNotDefault(DisplayData.item("validation", validate) + .withLabel("Validate Paths"), true); } @Override @@ -694,14 +696,22 @@ public PDone apply(PCollection input) { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .add(DisplayData.item("schema", type)) - .addIfNotNull(DisplayData.item("filePrefix", filenamePrefix)) - .addIfNotDefault( - DisplayData.item("shardNameTemplate", shardTemplate), + .add(DisplayData.item("schema", type) + .withLabel("Record Schema")) + .addIfNotNull(DisplayData.item("filePrefix", filenamePrefix) + .withLabel("Output File Prefix")) + .addIfNotDefault(DisplayData.item("shardNameTemplate", shardTemplate) + .withLabel("Output Shard Name Template"), DEFAULT_SHARD_TEMPLATE) - .addIfNotDefault(DisplayData.item("fileSuffix", filenameSuffix), "") - .addIfNotDefault(DisplayData.item("numShards", numShards), 0) - .addIfNotDefault(DisplayData.item("validation", validate), true); + .addIfNotDefault(DisplayData.item("fileSuffix", filenameSuffix) + .withLabel("Output File Suffix"), + "") + .addIfNotDefault(DisplayData.item("numShards", numShards) + .withLabel("Maximum Shard Count"), + 0) + .addIfNotDefault(DisplayData.item("validation", validate) + .withLabel("Validate Paths"), + true); } /** diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BigQueryIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BigQueryIO.java index 3f22648bdc60..fc20090045f6 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BigQueryIO.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BigQueryIO.java @@ -631,13 +631,18 @@ public void populateDisplayData(DisplayData.Builder builder) { TableReference table = getTable(); if (table != null) { - builder.add(DisplayData.item("table", toTableSpec(table))); + builder.add(DisplayData.item("table", toTableSpec(table)) + .withLabel("Table")); } builder - .addIfNotNull(DisplayData.item("query", query)) - .addIfNotNull(DisplayData.item("flattenResults", flattenResults)) - .addIfNotDefault(DisplayData.item("validation", validate), true); + .addIfNotNull(DisplayData.item("query", query) + .withLabel("Query")) + .addIfNotNull(DisplayData.item("flattenResults", flattenResults) + .withLabel("Flatten Query Results")) + .addIfNotDefault(DisplayData.item("validation", validate) + .withLabel("Validate Paths"), + true); } /** @@ -1753,17 +1758,23 @@ public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .addIfNotNull(DisplayData.item("table", jsonTableRef)) - .addIfNotNull(DisplayData.item("schema", jsonSchema)); + .addIfNotNull(DisplayData.item("table", jsonTableRef) + .withLabel("Table Reference")) + .addIfNotNull(DisplayData.item("schema", jsonSchema) + .withLabel("Table Schema")); if (tableRefFunction != null) { - builder.add(DisplayData.item("tableFn", tableRefFunction.getClass())); + builder.add(DisplayData.item("tableFn", tableRefFunction.getClass()) + .withLabel("Table Reference Function")); } builder - .add(DisplayData.item("createDisposition", createDisposition.toString())) - .add(DisplayData.item("writeDisposition", writeDisposition.toString())) - .addIfNotDefault(DisplayData.item("validation", validate), true); + .add(DisplayData.item("createDisposition", createDisposition.toString()) + .withLabel("Table CreateDisposition")) + .add(DisplayData.item("writeDisposition", writeDisposition.toString()) + .withLabel("Table WriteDisposition")) + .addIfNotDefault(DisplayData.item("validation", validate) + .withLabel("Validate Paths"), true); } /** Returns the create disposition. */ @@ -1855,8 +1866,10 @@ public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .addIfNotNull(DisplayData.item("schema", jsonSchema)) - .addIfNotNull(DisplayData.item("tableSpec", jsonTable)); + .addIfNotNull(DisplayData.item("schema", jsonSchema) + .withLabel("Table Schema")) + .addIfNotNull(DisplayData.item("tableSpec", jsonTable) + .withLabel("Table Specification")); } private static class BigQueryWriteOperation extends FileBasedWriteOperation { @@ -2096,7 +2109,8 @@ public void finishBundle(Context context) throws Exception { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); - builder.addIfNotNull(DisplayData.item("schema", jsonTableSchema)); + builder.addIfNotNull(DisplayData.item("schema", jsonTableSchema) + .withLabel("Table Schema")); } public TableReference getOrCreateTable(BigQueryOptions options, String tableSpec) @@ -2312,7 +2326,8 @@ public void populateDisplayData(DisplayData.Builder builder) { builder.addIfNotNull(DisplayData.item("tableSpec", tableSpec)); if (tableRefFunction != null) { - builder.add(DisplayData.item("tableFn", tableRefFunction.getClass())); + builder.add(DisplayData.item("tableFn", tableRefFunction.getClass()) + .withLabel("Table Reference Function")); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java index cf4f02ddd60c..49b2ad4fab34 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.java @@ -113,9 +113,12 @@ public String getKindString() { public void populateDisplayData(DisplayData.Builder builder) { // We explicitly do not register base-class data, instead we use the delegate inner source. builder - .add(DisplayData.item("source", source.getClass())) - .addIfNotDefault(DisplayData.item("maxRecords", maxNumRecords), Long.MAX_VALUE) - .addIfNotNull(DisplayData.item("maxReadTime", maxReadTime)) + .add(DisplayData.item("source", source.getClass()) + .withLabel("Read Source")) + .addIfNotDefault(DisplayData.item("maxRecords", maxNumRecords) + .withLabel("Maximum Read Records"), Long.MAX_VALUE) + .addIfNotNull(DisplayData.item("maxReadTime", maxReadTime) + .withLabel("Maximum Read Time")) .include(source); } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CompressedSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CompressedSource.java index 370c1e22c2ec..5cb0684bb3ab 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CompressedSource.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CompressedSource.java @@ -326,14 +326,17 @@ public void populateDisplayData(DisplayData.Builder builder) { // We explicitly do not register base-class data, instead we use the delegate inner source. builder .include(sourceDelegate) - .add(DisplayData.item("source", sourceDelegate.getClass())); + .add(DisplayData.item("source", sourceDelegate.getClass()) + .withLabel("Read Source")); if (channelFactory instanceof Enum) { // GZIP and BZIP are implemented as enums; Enum classes are anonymous, so use the .name() // value instead - builder.add(DisplayData.item("compressionMode", ((Enum) channelFactory).name())); + builder.add(DisplayData.item("compressionMode", ((Enum) channelFactory).name()) + .withLabel("Compression Mode")); } else { - builder.add(DisplayData.item("compressionMode", channelFactory.getClass())); + builder.add(DisplayData.item("compressionMode", channelFactory.getClass()) + .withLabel("Compression Mode")); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingInput.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingInput.java index 5a53e4d990dd..a21bc2418fa3 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingInput.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingInput.java @@ -119,7 +119,8 @@ public PCollection apply(PBegin begin) { @Override public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); - builder.add(DisplayData.item("upTo", numElements)); + builder.add(DisplayData.item("upTo", numElements) + .withLabel("Count Up To")); } } @@ -233,14 +234,17 @@ public PCollection apply(PBegin begin) { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); - builder.add(DisplayData.item("timestampFn", timestampFn.getClass())); + builder.add(DisplayData.item("timestampFn", timestampFn.getClass()) + .withLabel("Timestamp Function")); if (maxReadTime.isPresent()) { - builder.add(DisplayData.item("maxReadTime", maxReadTime.get())); + builder.add(DisplayData.item("maxReadTime", maxReadTime.get()) + .withLabel("Maximum Read Time")); } if (maxNumRecords.isPresent()) { - builder.add(DisplayData.item("maxRecords", maxNumRecords.get())); + builder.add(DisplayData.item("maxRecords", maxNumRecords.get()) + .withLabel("Maximum Read Records")); } } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/DatastoreIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/DatastoreIO.java index 81c3a395c156..cc8e9230d097 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/DatastoreIO.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/DatastoreIO.java @@ -391,12 +391,16 @@ public long getEstimatedSizeBytes(PipelineOptions options) throws Exception { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .addIfNotDefault(DisplayData.item("host", host), DEFAULT_HOST) - .addIfNotNull(DisplayData.item("dataset", datasetId)) - .addIfNotNull(DisplayData.item("namespace", namespace)); + .addIfNotDefault(DisplayData.item("host", host) + .withLabel("Datastore Service"), DEFAULT_HOST) + .addIfNotNull(DisplayData.item("dataset", datasetId) + .withLabel("Input Dataset")) + .addIfNotNull(DisplayData.item("namespace", namespace) + .withLabel("App Engine Namespace")); if (query != null) { - builder.add(DisplayData.item("query", query.toString())); + builder.add(DisplayData.item("query", query.toString()) + .withLabel("Query")); } } @@ -606,8 +610,10 @@ public DatastoreWriteOperation createWriteOperation(PipelineOptions options) { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .addIfNotDefault(DisplayData.item("host", host), DEFAULT_HOST) - .addIfNotNull(DisplayData.item("dataset", datasetId)); + .addIfNotDefault(DisplayData.item("host", host) + .withLabel("Datastore Service"), DEFAULT_HOST) + .addIfNotNull(DisplayData.item("dataset", datasetId) + .withLabel("Output Dataset")); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java index 23ccdd0142ad..904838003105 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java @@ -129,7 +129,8 @@ public void populateDisplayData(DisplayData.Builder builder) { String fileNamePattern = String.format("%s%s%s", baseOutputFilename, fileNamingTemplate, getFileExtension(extension)); - builder.add(DisplayData.item("fileNamePattern", fileNamePattern)); + builder.add(DisplayData.item("fileNamePattern", fileNamePattern) + .withLabel("File Name Pattern")); } /** diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSource.java index 954877f6ec73..e42c92c89849 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSource.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSource.java @@ -277,7 +277,8 @@ private static long getEstimatedSizeOfFilesBySampling( @Override public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); - builder.add(DisplayData.item("filePattern", getFileOrPatternSpec())); + builder.add(DisplayData.item("filePattern", getFileOrPatternSpec()) + .withLabel("File Pattern")); } private ListenableFuture>> createFutureForFileSplit( diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/OffsetBasedSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/OffsetBasedSource.java index 1c033f8b4eaa..9ee89a2bf799 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/OffsetBasedSource.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/OffsetBasedSource.java @@ -207,9 +207,12 @@ public boolean allowsDynamicSplitting() { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .addIfNotDefault(DisplayData.item("minBundleSize", minBundleSize), 1L) - .addIfNotDefault(DisplayData.item("startOffset", startOffset), 0L) - .addIfNotDefault(DisplayData.item("endOffset", endOffset), Long.MAX_VALUE); + .addIfNotDefault(DisplayData.item("minBundleSize", minBundleSize) + .withLabel("Minimum Bundle Size"), 1L) + .addIfNotDefault(DisplayData.item("startOffset", startOffset) + .withLabel("Start Read Offset"), 0L) + .addIfNotDefault(DisplayData.item("endOffset", endOffset) + .withLabel("End Read Offset"), Long.MAX_VALUE); } /** diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java index 6a1447775f68..e8350ffdddb1 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java @@ -643,17 +643,23 @@ public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .addIfNotNull(DisplayData.item("timestampLabel", timestampLabel)) - .addIfNotNull(DisplayData.item("idLabel", idLabel)) - .addIfNotNull(DisplayData.item("maxReadTime", maxReadTime)) - .addIfNotDefault(DisplayData.item("maxNumRecords", maxNumRecords), 0); + .addIfNotNull(DisplayData.item("timestampLabel", timestampLabel) + .withLabel("Timestamp Label Attribute")) + .addIfNotNull(DisplayData.item("idLabel", idLabel) + .withLabel("ID Label Attribute")) + .addIfNotNull(DisplayData.item("maxReadTime", maxReadTime) + .withLabel("Maximum Read Time")) + .addIfNotDefault(DisplayData.item("maxNumRecords", maxNumRecords) + .withLabel("Maximum Read Records"), 0); if (topic != null) { - builder.add(DisplayData.item("topic", topic.asPath())); + builder.add(DisplayData.item("topic", topic.asPath()) + .withLabel("Pubsub Topic")); } if (subscription != null) { - builder.add(DisplayData.item("subscription", subscription.asPath())); + builder.add(DisplayData.item("subscription", subscription.asPath()) + .withLabel("Pubsub Subscription")); } } @@ -955,11 +961,14 @@ public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .addIfNotNull(DisplayData.item("timestampLabel", timestampLabel)) - .addIfNotNull(DisplayData.item("idLabel", idLabel)); + .addIfNotNull(DisplayData.item("timestampLabel", timestampLabel) + .withLabel("Timestamp Label Attribute")) + .addIfNotNull(DisplayData.item("idLabel", idLabel) + .withLabel("ID Label Attribute")); if (topic != null) { - builder.add(DisplayData.item("topic", topic.asPath())); + builder.add(DisplayData.item("topic", topic.asPath()) + .withLabel("Pubsub Topic")); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Read.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Read.java index 965b14f07baf..fb4006383011 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Read.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Read.java @@ -149,7 +149,8 @@ public String getKindString() { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .add(DisplayData.item("source", source.getClass())) + .add(DisplayData.item("source", source.getClass()) + .withLabel("Read Source")) .include(source); } @@ -264,7 +265,8 @@ public String getKindString() { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .add(DisplayData.item("source", source.getClass())) + .add(DisplayData.item("source", source.getClass()) + .withLabel("Read Source")) .include(source); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java index 7f69c0a2131a..e4d1cf04fa12 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java @@ -344,9 +344,12 @@ public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .add(DisplayData.item("compressionType", compressionType.toString())) - .addIfNotDefault(DisplayData.item("validation", validate), true) - .addIfNotNull(DisplayData.item("filePattern", filepattern)); + .add(DisplayData.item("compressionType", compressionType.toString()) + .withLabel("Compression Type")) + .addIfNotDefault(DisplayData.item("validation", validate) + .withLabel("Validate Paths"), true) + .addIfNotNull(DisplayData.item("filePattern", filepattern) + .withLabel("File Pattern")); } @Override @@ -649,13 +652,17 @@ public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .addIfNotNull(DisplayData.item("filePrefix", filenamePrefix)) - .addIfNotDefault(DisplayData.item("fileSuffix", filenameSuffix), "") - .addIfNotDefault( - DisplayData.item("shardNameTemplate", shardTemplate), + .addIfNotNull(DisplayData.item("filePrefix", filenamePrefix) + .withLabel("Output File Prefix")) + .addIfNotDefault(DisplayData.item("fileSuffix", filenameSuffix) + .withLabel("Output Fix Suffix"), "") + .addIfNotDefault(DisplayData.item("shardNameTemplate", shardTemplate) + .withLabel("Output Shard Name Template"), DEFAULT_SHARD_TEMPLATE) - .addIfNotDefault(DisplayData.item("validation", validate), true) - .addIfNotDefault(DisplayData.item("numShards", numShards), 0); + .addIfNotDefault(DisplayData.item("validation", validate) + .withLabel("Validate Paths"), true) + .addIfNotDefault(DisplayData.item("numShards", numShards) + .withLabel("Maximum Shard Count"), 0); } /** diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Write.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Write.java index 0f2dbf87174c..66fef842e13c 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Write.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Write.java @@ -86,7 +86,8 @@ public PDone apply(PCollection input) { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .add(DisplayData.item("sink", sink.getClass())) + .add(DisplayData.item("sink", sink.getClass()) + .withLabel("Write Sink")) .include(sink); } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/XmlSink.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/XmlSink.java index eb7ba38dee03..bbe68d9ab70a 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/XmlSink.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/XmlSink.java @@ -227,8 +227,10 @@ public XmlWriteOperation createWriteOperation(PipelineOptions options) { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .addIfNotNull(DisplayData.item("rootElement", rootElementName)) - .addIfNotNull(DisplayData.item("recordClass", classToBind)); + .addIfNotNull(DisplayData.item("rootElement", rootElementName) + .withLabel("XML Root Element")) + .addIfNotNull(DisplayData.item("recordClass", classToBind) + .withLabel("XML Record Class")); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/XmlSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/XmlSource.java index 0a418c883ac9..56651e85c2ea 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/XmlSource.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/XmlSource.java @@ -222,9 +222,12 @@ public void validate() { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .addIfNotNull(DisplayData.item("rootElement", rootElement)) - .addIfNotNull(DisplayData.item("recordElement", recordElement)) - .addIfNotNull(DisplayData.item("recordClass", recordClass)); + .addIfNotNull(DisplayData.item("rootElement", rootElement) + .withLabel("XML Root Element")) + .addIfNotNull(DisplayData.item("recordElement", recordElement) + .withLabel("XML Record Element")) + .addIfNotNull(DisplayData.item("recordClass", recordClass) + .withLabel("XML Record Class")); } @Override diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ApproximateQuantiles.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ApproximateQuantiles.java index 4eb06d22b87a..b056830f4219 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ApproximateQuantiles.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ApproximateQuantiles.java @@ -365,8 +365,10 @@ public Coder> getAccumulatorCoder( public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .add(DisplayData.item("numQuantiles", numQuantiles)) - .add(DisplayData.item("comparer", compareFn.getClass())); + .add(DisplayData.item("numQuantiles", numQuantiles) + .withLabel("Quantile Count")) + .add(DisplayData.item("comparer", compareFn.getClass()) + .withLabel("Record Comparer")); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ApproximateUnique.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ApproximateUnique.java index e4559d472a76..2fa24277661c 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ApproximateUnique.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ApproximateUnique.java @@ -461,7 +461,9 @@ static long sampleSizeFromEstimationError(double estimationError) { private static void populateDisplayData( DisplayData.Builder builder, long sampleSize, Double maxEstimationError) { builder - .add(DisplayData.item("sampleSize", sampleSize)) - .addIfNotNull(DisplayData.item("maximumEstimationError", maxEstimationError)); + .add(DisplayData.item("sampleSize", sampleSize) + .withLabel("Sample Size")) + .addIfNotNull(DisplayData.item("maximumEstimationError", maxEstimationError) + .withLabel("Maximum Estimation Error")); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java index ffbaafa2533a..190c413ca0a2 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java @@ -127,7 +127,8 @@ public static Globally globally( } private static DisplayData.Item> displayDataForFn(T fn) { - return DisplayData.item("combineFn", fn.getClass()); + return DisplayData.item("combineFn", fn.getClass()) + .withLabel("Combiner"); } private static Globally globally( @@ -1428,8 +1429,10 @@ private static void populateDisplayData( private static void populateGlobalDisplayData( DisplayData.Builder builder, int fanout, boolean insertDefault) { builder - .addIfNotDefault(DisplayData.item("fanout", fanout), 0) - .add(DisplayData.item("emitDefaultOnEmptyInput", insertDefault)); + .addIfNotDefault(DisplayData.item("fanout", fanout) + .withLabel("Key Fanout Size"), 0) + .add(DisplayData.item("emitDefaultOnEmptyInput", insertDefault) + .withLabel("Emit Default On Empty Input")); } /** @@ -1604,7 +1607,8 @@ public List compact(List accumulator) { @Override public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); - builder.add(DisplayData.item("combineFn", combiner.getClass())); + builder.add(DisplayData.item("combineFn", combiner.getClass()) + .withLabel("Combiner")); } private List mergeToSingleton(Iterable values) { @@ -2095,7 +2099,8 @@ public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); Combine.populateDisplayData(builder, fn, fnDisplayData); - builder.add(DisplayData.item("fanoutFn", hotKeyFanout.getClass())); + builder.add(DisplayData.item("fanoutFn", hotKeyFanout.getClass()) + .withLabel("Fanout Function")); } /** diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/CombineFns.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/CombineFns.java index 1bf305fb9e87..d0a8909138b5 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/CombineFns.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/CombineFns.java @@ -1053,7 +1053,8 @@ private static void populateDisplayData( for (int i = 0; i < combineFns.size(); i++) { HasDisplayData combineFn = combineFns.get(i); - builder.add(DisplayData.item("combineFn" + (i + 1), combineFn.getClass())); + builder.add(DisplayData.item("combineFn" + (i + 1), combineFn.getClass()) + .withLabel("Combine Function")); combineFnMap.put(combineFn.getClass(), combineFn); } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Filter.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Filter.java index da692b2ce19d..57796b89564f 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Filter.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Filter.java @@ -261,6 +261,7 @@ protected Coder getDefaultOutputCoder(PCollection input) { private static void populateDisplayData( DisplayData.Builder builder, String predicateDescription) { - builder.add(DisplayData.item("predicate", predicateDescription)); + builder.add(DisplayData.item("predicate", predicateDescription) + .withLabel("Filter Predicate")); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java index 21e6ecfb816f..8ad57d2bb3b2 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java @@ -279,7 +279,8 @@ public static KvCoder> getOutputKvCoder(Coder> in public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); if (fewKeys) { - builder.add(DisplayData.item("fewKeys", true)); + builder.add(DisplayData.item("fewKeys", true) + .withLabel("Has Few Keys")); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/IntraBundleParallelization.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/IntraBundleParallelization.java index 62c09c2158ac..3cb84159f0c9 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/IntraBundleParallelization.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/IntraBundleParallelization.java @@ -178,8 +178,10 @@ public PCollection apply(PCollection input) { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .add(DisplayData.item("maxParallelism", maxParallelism)) - .add(DisplayData.item("fn", doFn.getClass())) + .add(DisplayData.item("maxParallelism", maxParallelism) + .withLabel("Maximum Parallelism")) + .add(DisplayData.item("fn", doFn.getClass()) + .withLabel("Function")) .include(doFn); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/MapElements.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/MapElements.java index 29d1dde8e2fc..d64bad1ff775 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/MapElements.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/MapElements.java @@ -120,6 +120,7 @@ public void populateDisplayData(DisplayData.Builder builder) { @Override public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); - builder.add(DisplayData.item("mapFn", fn.getClass())); + builder.add(DisplayData.item("mapFn", fn.getClass()) + .withLabel("Map Function")); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Max.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Max.java index 01a190f234ac..c05bd1744367 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Max.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Max.java @@ -209,7 +209,8 @@ public T apply(T left, T right) { @Override public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); - builder.add(DisplayData.item("comparer", comparator.getClass())); + builder.add(DisplayData.item("comparer", comparator.getClass()) + .withLabel("Record Comparer")); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Min.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Min.java index f20d48bbe4ee..7a6a8a2ae953 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Min.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Min.java @@ -209,7 +209,8 @@ public T apply(T left, T right) { @Override public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); - builder.add(DisplayData.item("comparer", comparator.getClass())); + builder.add(DisplayData.item("comparer", comparator.getClass()) + .withLabel("Record Comparer")); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java index 88945c43ecb6..834f60d212bb 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java @@ -1266,7 +1266,8 @@ private static void populateDisplayData( DisplayData.Builder builder, DoFn fn, Class fnClass) { builder .include(fn) - .add(DisplayData.item("fn", fnClass)); + .add(DisplayData.item("fn", fnClass) + .withLabel("Transform Function")); } /** diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Partition.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Partition.java index 44ba2f30b111..c5aa82c94230 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Partition.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Partition.java @@ -182,8 +182,10 @@ public void processElement(ProcessContext c) { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .add(DisplayData.item("numPartitions", numPartitions)) - .add(DisplayData.item("partitionFn", partitionFn.getClass())); + .add(DisplayData.item("numPartitions", numPartitions) + .withLabel("Partition Count")) + .add(DisplayData.item("partitionFn", partitionFn.getClass()) + .withLabel("Partition Function")); } } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Sample.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Sample.java index 58188def4226..1d98668d4a5e 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Sample.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Sample.java @@ -158,7 +158,8 @@ public PCollection apply(PCollection in) { @Override public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); - builder.add(DisplayData.item("sampleSize", limit)); + builder.add(DisplayData.item("sampleSize", limit) + .withLabel("Sample Size")); } } @@ -258,7 +259,8 @@ public Coder> getDefaultOutputCoder( @Override public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); - builder.add(DisplayData.item("sampleSize", sampleSize)); + builder.add(DisplayData.item("sampleSize", sampleSize) + .withLabel("Sample Size")); } } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Top.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Top.java index 71af081278a8..257f77b1ed48 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Top.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Top.java @@ -397,8 +397,10 @@ public Coder> getAccumulatorCoder( public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .add(DisplayData.item("count", count)) - .add(DisplayData.item("comparer", compareFn.getClass())); + .add(DisplayData.item("count", count) + .withLabel("Top Count")) + .add(DisplayData.item("comparer", compareFn.getClass()) + .withLabel("Record Comparer")); } @Override diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/CalendarWindows.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/CalendarWindows.java index 490fbd1a41ec..9653ae84ea94 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/CalendarWindows.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/CalendarWindows.java @@ -150,9 +150,11 @@ public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .add(DisplayData.item("numDays", number)) + .add(DisplayData.item("numDays", number) + .withLabel("Windows Days")) .addIfNotDefault( - DisplayData.item("startDate", new DateTime(startDate, timeZone).toInstant()), + DisplayData.item("startDate", new DateTime(startDate, timeZone).toInstant()) + .withLabel("Window Start Date"), new DateTime(DEFAULT_START_DATE, DateTimeZone.UTC).toInstant()); } @@ -248,9 +250,11 @@ public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .add(DisplayData.item("numMonths", number)) + .add(DisplayData.item("numMonths", number) + .withLabel("Window Months")) .addIfNotDefault( - DisplayData.item("startDate", new DateTime(startDate, timeZone).toInstant()), + DisplayData.item("startDate", new DateTime(startDate, timeZone).toInstant()) + .withLabel("Window Start Date"), new DateTime(DEFAULT_START_DATE, DateTimeZone.UTC).toInstant()); } @@ -355,9 +359,11 @@ public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .add(DisplayData.item("numYears", number)) + .add(DisplayData.item("numYears", number) + .withLabel("Window Years")) .addIfNotDefault( - DisplayData.item("startDate", new DateTime(startDate, timeZone).toInstant()), + DisplayData.item("startDate", new DateTime(startDate, timeZone).toInstant()) + .withLabel("Window Start Date"), new DateTime(DEFAULT_START_DATE, DateTimeZone.UTC).toInstant()); } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/FixedWindows.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/FixedWindows.java index 322259c990d0..1e439ff389b3 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/FixedWindows.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/FixedWindows.java @@ -87,8 +87,10 @@ public IntervalWindow assignWindow(Instant timestamp) { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .add(DisplayData.item("size", size)) - .addIfNotDefault(DisplayData.item("offset", offset), Duration.ZERO); + .add(DisplayData.item("size", size) + .withLabel("Window Duration")) + .addIfNotDefault(DisplayData.item("offset", offset) + .withLabel("Window Start Offset"), Duration.ZERO); } @Override diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Sessions.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Sessions.java index 788566eed3b0..875db12918ae 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Sessions.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Sessions.java @@ -93,7 +93,8 @@ public Duration getGapDuration() { @Override public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); - builder.add(DisplayData.item("gapDuration", gapDuration)); + builder.add(DisplayData.item("gapDuration", gapDuration) + .withLabel("Session Gap Duration")); } @Override diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/SlidingWindows.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/SlidingWindows.java index 62c2738d96ca..99367f5557f4 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/SlidingWindows.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/SlidingWindows.java @@ -144,9 +144,12 @@ public boolean isCompatible(WindowFn other) { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); builder - .add(DisplayData.item("size", size)) - .add(DisplayData.item("period", period)) - .add(DisplayData.item("offset", offset)); + .add(DisplayData.item("size", size) + .withLabel("Window Size")) + .add(DisplayData.item("period", period) + .withLabel("Window Period")) + .add(DisplayData.item("offset", offset) + .withLabel("Window Start Offset")); } /** diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java index ab6b7f0cb4d5..324b4d5c34b8 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Window.java @@ -602,29 +602,35 @@ public void populateDisplayData(DisplayData.Builder builder) { if (windowFn != null) { builder - .add(DisplayData.item("windowFn", windowFn.getClass())) + .add(DisplayData.item("windowFn", windowFn.getClass()) + .withLabel("Windowing Function")) .include(windowFn); } if (allowedLateness != null) { - builder.addIfNotDefault(DisplayData.item("allowedLateness", allowedLateness), + builder.addIfNotDefault(DisplayData.item("allowedLateness", allowedLateness) + .withLabel("Allowed Lateness"), Duration.millis(BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis())); } if (trigger != null && !(trigger instanceof DefaultTrigger)) { - builder.add(DisplayData.item("trigger", trigger.toString())); + builder.add(DisplayData.item("trigger", trigger.toString()) + .withLabel("Trigger")); } if (mode != null) { - builder.add(DisplayData.item("accumulationMode", mode.toString())); + builder.add(DisplayData.item("accumulationMode", mode.toString()) + .withLabel("Accumulation Mode")); } if (closingBehavior != null) { - builder.add(DisplayData.item("closingBehavior", closingBehavior.toString())); + builder.add(DisplayData.item("closingBehavior", closingBehavior.toString()) + .withLabel("Window Closing Behavior")); } if (outputTimeFn != null) { - builder.add(DisplayData.item("outputTimeFn", outputTimeFn.getClass())); + builder.add(DisplayData.item("outputTimeFn", outputTimeFn.getClass()) + .withLabel("Output Time Function")); } } diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java index bf85dea6d330..965649417409 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java @@ -266,14 +266,17 @@ public void validate(PBegin input) { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); - builder.add(DisplayData.item("tableId", tableId)); + builder.add(DisplayData.item("tableId", tableId) + .withLinkUrl("Table ID")); if (options != null) { - builder.add(DisplayData.item("bigtableOptions", options.toString())); + builder.add(DisplayData.item("bigtableOptions", options.toString()) + .withLabel("Bigtable Options")); } if (filter != null) { - builder.add(DisplayData.item("rowFilter", filter.toString())); + builder.add(DisplayData.item("rowFilter", filter.toString()) + .withLabel("Table Row Filter")); } } @@ -448,10 +451,12 @@ Write withBigtableService(BigtableService bigtableService) { public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); - builder.add(DisplayData.item("tableId", tableId)); + builder.add(DisplayData.item("tableId", tableId) + .withLabel("Table ID")); if (options != null) { - builder.add(DisplayData.item("bigtableOptions", options.toString())); + builder.add(DisplayData.item("bigtableOptions", options.toString()) + .withLabel("Bigtable Options")); } } From 472dd41967c857becda2266fa234dcf99d3d43db Mon Sep 17 00:00:00 2001 From: Scott Wegner Date: Mon, 16 May 2016 11:19:08 -0700 Subject: [PATCH 2/2] fixup! Minor changes based on feedback --- .../java/org/apache/beam/sdk/io/AvroIO.java | 6 +-- .../org/apache/beam/sdk/io/BigQueryIO.java | 4 +- .../java/org/apache/beam/sdk/io/PubsubIO.java | 39 +++++++++---------- .../java/org/apache/beam/sdk/io/TextIO.java | 6 +-- 4 files changed, 27 insertions(+), 28 deletions(-) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java index 15ea85f61a99..4b40c01eedfc 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java @@ -334,7 +334,7 @@ public void populateDisplayData(DisplayData.Builder builder) { .addIfNotNull(DisplayData.item("filePattern", filepattern) .withLabel("Input File Pattern")) .addIfNotDefault(DisplayData.item("validation", validate) - .withLabel("Validate Paths"), true); + .withLabel("Validation Enabled"), true); } @Override @@ -707,10 +707,10 @@ public void populateDisplayData(DisplayData.Builder builder) { .withLabel("Output File Suffix"), "") .addIfNotDefault(DisplayData.item("numShards", numShards) - .withLabel("Maximum Shard Count"), + .withLabel("Maximum Output Shards"), 0) .addIfNotDefault(DisplayData.item("validation", validate) - .withLabel("Validate Paths"), + .withLabel("Validation Enabled"), true); } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BigQueryIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BigQueryIO.java index fc20090045f6..f8a05e90b8df 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BigQueryIO.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/BigQueryIO.java @@ -641,7 +641,7 @@ public void populateDisplayData(DisplayData.Builder builder) { .addIfNotNull(DisplayData.item("flattenResults", flattenResults) .withLabel("Flatten Query Results")) .addIfNotDefault(DisplayData.item("validation", validate) - .withLabel("Validate Paths"), + .withLabel("Validation Enabled"), true); } @@ -1774,7 +1774,7 @@ public void populateDisplayData(DisplayData.Builder builder) { .add(DisplayData.item("writeDisposition", writeDisposition.toString()) .withLabel("Table WriteDisposition")) .addIfNotDefault(DisplayData.item("validation", validate) - .withLabel("Validate Paths"), true); + .withLabel("Validation Enabled"), true); } /** Returns the create disposition. */ diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java index e8350ffdddb1..78fec852c666 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java @@ -134,6 +134,23 @@ private static void validatePubsubName(String name) { } } + /** + * Populate common {@link DisplayData} between Pubsub source and sink. + */ + private static void populateCommonDisplayData(DisplayData.Builder builder, + String timestampLabel, String idLabel, PubsubTopic topic) { + builder + .addIfNotNull(DisplayData.item("timestampLabel", timestampLabel) + .withLabel("Timestamp Label Attribute")) + .addIfNotNull(DisplayData.item("idLabel", idLabel) + .withLabel("ID Label Attribute")); + + if (topic != null) { + builder.add(DisplayData.item("topic", topic.asPath()) + .withLabel("Pubsub Topic")); + } + } + /** * Class representing a Cloud Pub/Sub Subscription. */ @@ -641,22 +658,14 @@ public PCollection apply(PInput input) { @Override public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); + populateCommonDisplayData(builder, timestampLabel, idLabel, topic); builder - .addIfNotNull(DisplayData.item("timestampLabel", timestampLabel) - .withLabel("Timestamp Label Attribute")) - .addIfNotNull(DisplayData.item("idLabel", idLabel) - .withLabel("ID Label Attribute")) .addIfNotNull(DisplayData.item("maxReadTime", maxReadTime) .withLabel("Maximum Read Time")) .addIfNotDefault(DisplayData.item("maxNumRecords", maxNumRecords) .withLabel("Maximum Read Records"), 0); - if (topic != null) { - builder.add(DisplayData.item("topic", topic.asPath()) - .withLabel("Pubsub Topic")); - } - if (subscription != null) { builder.add(DisplayData.item("subscription", subscription.asPath()) .withLabel("Pubsub Subscription")); @@ -959,17 +968,7 @@ public PDone apply(PCollection input) { @Override public void populateDisplayData(DisplayData.Builder builder) { super.populateDisplayData(builder); - - builder - .addIfNotNull(DisplayData.item("timestampLabel", timestampLabel) - .withLabel("Timestamp Label Attribute")) - .addIfNotNull(DisplayData.item("idLabel", idLabel) - .withLabel("ID Label Attribute")); - - if (topic != null) { - builder.add(DisplayData.item("topic", topic.asPath()) - .withLabel("Pubsub Topic")); - } + populateCommonDisplayData(builder, timestampLabel, idLabel, topic); } @Override diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java index e4d1cf04fa12..79eeb081671b 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java @@ -347,7 +347,7 @@ public void populateDisplayData(DisplayData.Builder builder) { .add(DisplayData.item("compressionType", compressionType.toString()) .withLabel("Compression Type")) .addIfNotDefault(DisplayData.item("validation", validate) - .withLabel("Validate Paths"), true) + .withLabel("Validation Enabled"), true) .addIfNotNull(DisplayData.item("filePattern", filepattern) .withLabel("File Pattern")); } @@ -660,9 +660,9 @@ public void populateDisplayData(DisplayData.Builder builder) { .withLabel("Output Shard Name Template"), DEFAULT_SHARD_TEMPLATE) .addIfNotDefault(DisplayData.item("validation", validate) - .withLabel("Validate Paths"), true) + .withLabel("Validation Enabled"), true) .addIfNotDefault(DisplayData.item("numShards", numShards) - .withLabel("Maximum Shard Count"), 0); + .withLabel("Maximum Output Shards"), 0); } /**