Add 3 examples in playground SQL transform, schema transform and Comp…

…osed Combine. All applying the same combiners (Min,Max,Sum) on the input.

examples/java/src/main/java/org/apache/beam/examples/CoCombineTransformExample.java

@@ -0,0 +1,141 @@
+/*
+ * 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.examples;
+
+// beam-playground:
+//   name: CombineFns.ComposedCombineFn
+//   description: Demonstration of Composed Combine transform usage.
+//   multifile: false
+//   default_example: false
+//   context_line: 64
+//   categories:
+//     - Schemas
+//     - Combiners
+//   complexity: MEDIUM
+//   tags:
+//     - transforms
+//     - numbers
+
+// gradle clean execute -DmainClass=org.apache.beam.examples.CoCombineTransformExample --args="--runner=DirectRunner" -Pdirect-runner
+
+import java.util.ArrayList;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.CombineFns;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.Max;
+import org.apache.beam.sdk.transforms.Min;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.SimpleFunction;
+import org.apache.beam.sdk.transforms.Sum;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TupleTag;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An example that uses Composed combiners to apply multiple combiners (Sum, Min, Max) on the 
+ * input PCollection.
+ * 
+ * <p>For a detailed documentation of Composed Combines, see <a
+ * href="https://beam.apache.org/releases/javadoc/2.1.0/org/apache/beam/sdk/transforms/CombineFns.html">
+ * https://beam.apache.org/releases/javadoc/2.1.0/org/apache/beam/sdk/transforms/CombineFns.html </a>
+ */
+public class CoCombineTransformExample {
+  public static void main(String[] args) {
+    PipelineOptions options = PipelineOptionsFactory.create();
+    Pipeline pipeline = Pipeline.create(options);
+    // [START main_section]
+    // Create input
+    PCollection<KV<Long, Long>> inputKV =
+        pipeline.apply(Create.of(
+            KV.of(1L, 1L),
+            KV.of(1L, 5L),
+            KV.of(2L, 10L),
+            KV.of(2L, 20L),
+            KV.of(3L, 1L)
+          ));
+    /** 
+     * Define the function used to filter elements before sending them to the Combiner. 
+     * With identityFn all elements (here perKey) will be combined.
+     */
+    SimpleFunction<Long, Long> identityFn =
+      new SimpleFunction<Long, Long>() {
+       @Override
+        public Long apply(Long input) {
+            return input;
+        }};
+
+    // tuple tags to identify the outputs of the Composed Combine
+    TupleTag<Long> sumTag = new TupleTag<Long>("sum_n");
+    TupleTag<Long> minTag = new TupleTag<Long>("min_n");
+    TupleTag<Long> maxTag = new TupleTag<Long>("max_n");
+
+    CombineFns.ComposedCombineFn<Long> composedCombine = 
+      CombineFns.compose()
+        .with(identityFn, Sum.ofLongs(), sumTag) //elements filtered by the identityFn, will be combined in a Sum and the output will be tagged 
+        .with(identityFn, Min.ofLongs(), minTag)
+        .with(identityFn, Max.ofLongs(), maxTag)
+        ;
+
+    PCollection<KV<Long,CombineFns.CoCombineResult>> combinedData = 
+      inputKV
+        .apply("Combine all", Combine.perKey(composedCombine));
+
+    // transform the CoCombineResult output into a KV format, simpler to use for printing
+    PCollection<KV<Long,Iterable<KV<String,Long>>>> result = combinedData
+        .apply(ParDo.of(
+         new DoFn<KV<Long,CombineFns.CoCombineResult>, KV<Long,Iterable<KV<String,Long>>>>() {
+          @ProcessElement
+           public void processElement(ProcessContext c) throws Exception {
+             CombineFns.CoCombineResult e = c.element().getValue();
+             c.output(KV.of(c.element().getKey(),
+                new ArrayList<KV<String,Long>>() {{
+                  add(KV.of(minTag.getId(), e.get(minTag)));
+                  add(KV.of(maxTag.getId(), e.get(maxTag)));
+                  add(KV.of(sumTag.getId(), e.get(sumTag)));
+                }}
+             ));
+           }
+         }));
+
+    // [END main_section]
+    // Log values
+    result.apply(ParDo.of(new LogOutput<>("PCollection values after CoCombine transform: ")));
+    pipeline.run();
+  }
+
+  static class LogOutput<T> extends DoFn<T, T> {
+    private static final Logger LOG = LoggerFactory.getLogger(LogOutput.class);
+    private final String prefix;
+
+    public LogOutput(String prefix) {
+      this.prefix = prefix;
+    }
+
+    @ProcessElement
+    public void processElement(ProcessContext c) throws Exception {
+      LOG.info(prefix + c.element());
+      c.output(c.element());
+    }
+  }
+}

examples/java/src/main/java/org/apache/beam/examples/SchemaTransformExample.java

@@ -0,0 +1,110 @@
+/*
+ * 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.examples;
+
+// beam-playground:
+//   name: Group.ByFields
+//   description: Demonstration of Schema transform usage.
+//   multifile: false
+//   default_example: false
+//   context_line: 60
+//   categories:
+//     - Schemas
+//     - Combiners
+//   complexity: BASIC
+//   tags:
+//     - transforms
+//     - numbers
+
+//gradle clean execute -DmainClass=org.apache.beam.examples.SchemaTransformExample --args="--runner=DirectRunner" -Pdirect-runner
+
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Sum;
+import org.apache.beam.sdk.values.PCollection;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.transforms.Group;
+import org.apache.beam.sdk.schemas.transforms.Select;
+import org.apache.beam.sdk.transforms.Max;
+import org.apache.beam.sdk.transforms.Min;
+import org.apache.beam.sdk.values.Row;
+
+/**
+ * An example that uses Schema transforms to apply multiple combiners (Sum, Min, Max) on the 
+ * input PCollection.
+ * 
+ * <p>For a detailed documentation of Schemas, see <a
+ * href="https://beam.apache.org/documentation/programming-guide/#schemas">
+ * https://beam.apache.org/documentation/programming-guide/#schemas </a>
+ */
+public class SchemaTransformExample {
+  public static void main(String[] args) {
+    PipelineOptions options = PipelineOptionsFactory.create();
+    Pipeline pipeline = Pipeline.create(options);
+    // [START main_section]
+    // define the input row schema
+    Schema inputSchema =
+      Schema.builder()
+          .addInt32Field("k")
+          .addInt32Field("n")
+          .build();
+    // Create input
+    PCollection<Row> input = pipeline.apply(
+        Create.of(
+          Row.withSchema(inputSchema).addValues(1,1).build(),
+          Row.withSchema(inputSchema).addValues(1,5).build(),
+          Row.withSchema(inputSchema).addValues(2,10).build(),
+          Row.withSchema(inputSchema).addValues(2,20).build(),
+          Row.withSchema(inputSchema).addValues(3,1).build()
+        ))
+      .setRowSchema(inputSchema);
+
+    PCollection<Row> result = input.apply(Select.fieldNames("n",",k"))
+      .apply(Group.<Row>byFieldNames("k")
+            .aggregateField("n", Min.ofIntegers(), "min_n")
+            .aggregateField("n", Max.ofIntegers(), "max_n")
+            .aggregateField("n", Sum.ofIntegers(), "sum_n")
+        )
+    ;
+    // [END main_section]
+    // Log values
+    result.apply(ParDo.of(new LogOutput<>("PCollection values after Schema transform: ")));
+    pipeline.run();
+  }
+
+  static class LogOutput<T> extends DoFn<T, T> {
+    private static final Logger LOG = LoggerFactory.getLogger(LogOutput.class);
+    private final String prefix;
+
+    public LogOutput(String prefix) {
+      this.prefix = prefix;
+    }
+
+    @ProcessElement
+    public void processElement(ProcessContext c) throws Exception {
+      LOG.info(prefix + c.element());
+      c.output(c.element());
+    }
+  }
+}

examples/java/src/main/java/org/apache/beam/examples/SqlTransformExample.java

@@ -0,0 +1,104 @@
+/*
+ * 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.examples;
+
+// beam-playground:
+//   name: SqlTransform
+//   description: Demonstration of SQL transform usage.
+//   multifile: false
+//   default_example: false
+//   context_line: 60
+//   categories:
+//     - SQL
+//     - Combiners
+//   complexity: BASIC
+//   tags:
+//     - transforms
+//     - numbers
+
+//gradle clean execute -DmainClass=org.apache.beam.examples.SqlTransformExample --args="--runner=DirectRunner" -Pdirect-runner
+
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.beam.sdk.extensions.sql.SqlTransform;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.values.Row;
+
+/**
+ * An example that uses Beam SQL transformation to apply multiple combiners (Min, Max, Sum) on the 
+ * input PCollection.
+ * 
+ * <p> Using SQL syntax to define a transform than can be integrated in a Java pipeline.
+ *
+ * <p>For a detailed documentation of Beam SQL, see <a
+ * href="https://beam.apache.org/documentation/dsls/sql/overview/">
+ * https://beam.apache.org/documentation/dsls/sql/overview/ </a>
+ */
+public class SqlTransformExample {
+  public static void main(String[] args) {
+    PipelineOptions options = PipelineOptionsFactory.create();
+    Pipeline pipeline = Pipeline.create(options);
+    // [START main_section]
+    // define the input row format
+    Schema inputSchema =
+      Schema.builder()
+          .addInt32Field("k")
+          .addInt32Field("n")
+          .build();
+    // Create input
+    PCollection<Row> input = pipeline.apply(
+        Create.of(
+          Row.withSchema(inputSchema).addValues(1,1).build(),
+          Row.withSchema(inputSchema).addValues(1,5).build(),
+          Row.withSchema(inputSchema).addValues(2,10).build(),
+          Row.withSchema(inputSchema).addValues(2,20).build(),
+          Row.withSchema(inputSchema).addValues(3,1).build()
+        ))
+      .setRowSchema(inputSchema);
+
+
+    PCollection<Row> result = input.apply(
+      SqlTransform.query("select k, min(n) as min_n, max(n) as max_n, sum(n) as sum_n from PCOLLECTION group by k"));
+    // [END main_section]
+    // Log values
+    result.apply(ParDo.of(new LogOutput<>("PCollection values after SQL transform: ")));
+    pipeline.run();
+  }
+
+  static class LogOutput<T> extends DoFn<T, T> {
+    private static final Logger LOG = LoggerFactory.getLogger(LogOutput.class);
+    private final String prefix;
+
+    public LogOutput(String prefix) {
+      this.prefix = prefix;
+    }
+
+    @ProcessElement
+    public void processElement(ProcessContext c) throws Exception {
+      LOG.info(prefix + c.element());
+      c.output(c.element());
+    }
+  }
+}