You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by am...@apache.org on 2016/03/10 21:58:34 UTC
[09/50] [abbrv] incubator-beam git commit: Resolve some generics
warnings with some fancier footwork
Resolve some generics warnings with some fancier footwork
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/2820534a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/2820534a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/2820534a
Branch: refs/heads/master
Commit: 2820534a54fe77bf8220da9bb0cd1d4ce6c8b2b3
Parents: 5069eed
Author: Sean Owen <so...@cloudera.com>
Authored: Fri Jul 3 18:38:02 2015 +0100
Committer: Tom White <to...@cloudera.com>
Committed: Thu Mar 10 11:15:14 2016 +0000
----------------------------------------------------------------------
runners/spark/README.md | 2 +-
runners/spark/pom.xml | 2 +-
.../com/cloudera/dataflow/hadoop/HadoopIO.java | 17 +++++++++--------
.../cloudera/dataflow/spark/EvaluationContext.java | 2 +-
.../dataflow/spark/SparkPipelineRunner.java | 13 +++++++------
.../dataflow/spark/SparkRuntimeContext.java | 7 ++++---
.../dataflow/spark/TransformEvaluator.java | 2 +-
.../dataflow/spark/TransformTranslator.java | 6 +++---
.../spark/HadoopFileFormatPipelineTest.java | 15 ++++++++++-----
9 files changed, 37 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/2820534a/runners/spark/README.md
----------------------------------------------------------------------
diff --git a/runners/spark/README.md b/runners/spark/README.md
index 52f7f9b..d93f554 100644
--- a/runners/spark/README.md
+++ b/runners/spark/README.md
@@ -66,7 +66,7 @@ First download a text document to use as input:
Then run the [word count example][wc] from the SDK using a single threaded Spark instance
in local mode:
- mvn exec:exec -Dclass=com.google.cloud.dataflow.examples.WordCount \
+ mvn exec:exec -DmainClass=com.google.cloud.dataflow.examples.WordCount \
-Dinput=/tmp/kinglear.txt -Doutput=/tmp/out -Drunner=SparkPipelineRunner \
-DsparkMaster=local
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/2820534a/runners/spark/pom.xml
----------------------------------------------------------------------
diff --git a/runners/spark/pom.xml b/runners/spark/pom.xml
index 1ab37b7..0382108 100644
--- a/runners/spark/pom.xml
+++ b/runners/spark/pom.xml
@@ -194,7 +194,7 @@ License.
<arguments>
<argument>-classpath</argument>
<classpath />
- <argument>${class}</argument>
+ <argument>${mainClass}</argument>
<argument>--input=${input}</argument>
<argument>--output=${output}</argument>
<argument>--runner=${runner}</argument>
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/2820534a/runners/spark/src/main/java/com/cloudera/dataflow/hadoop/HadoopIO.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/com/cloudera/dataflow/hadoop/HadoopIO.java b/runners/spark/src/main/java/com/cloudera/dataflow/hadoop/HadoopIO.java
index 434a132..5873b9f 100644
--- a/runners/spark/src/main/java/com/cloudera/dataflow/hadoop/HadoopIO.java
+++ b/runners/spark/src/main/java/com/cloudera/dataflow/hadoop/HadoopIO.java
@@ -32,20 +32,21 @@ public final class HadoopIO {
private Read() {
}
- public static Bound from(String filepattern) {
- return new Bound().from(filepattern);
+ public static <K, V> Bound<K, V> from(String filepattern) {
+ return new Bound<K, V>().from(filepattern);
}
- public static Bound withFormatClass(Class<? extends FileInputFormat<?, ?>> format) {
- return new Bound().withFormatClass(format);
+ public static <K, V> Bound<K, V> withFormatClass(
+ Class<? extends FileInputFormat<K, V>> format) {
+ return new Bound<K, V>().withFormatClass(format);
}
- public static Bound withKeyClass(Class<?> key) {
- return new Bound().withKeyClass(key);
+ public static <K, V> Bound<K, V> withKeyClass(Class<K> key) {
+ return new Bound<K, V>().withKeyClass(key);
}
- public static Bound withValueClass(Class<?> value) {
- return new Bound().withValueClass(value);
+ public static <K, V> Bound<K, V> withValueClass(Class<V> value) {
+ return new Bound<K, V>().withValueClass(value);
}
public static class Bound<K, V> extends PTransform<PInput, PCollection<KV<K, V>>> {
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/2820534a/runners/spark/src/main/java/com/cloudera/dataflow/spark/EvaluationContext.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/com/cloudera/dataflow/spark/EvaluationContext.java b/runners/spark/src/main/java/com/cloudera/dataflow/spark/EvaluationContext.java
index 7906775..56f8521 100644
--- a/runners/spark/src/main/java/com/cloudera/dataflow/spark/EvaluationContext.java
+++ b/runners/spark/src/main/java/com/cloudera/dataflow/spark/EvaluationContext.java
@@ -127,7 +127,7 @@ public class EvaluationContext implements EvaluationResult {
leafRdds.add(rdd);
}
- JavaRDDLike<?, ?> getInputRDD(PTransform transform) {
+ JavaRDDLike<?, ?> getInputRDD(PTransform<? extends PInput, ?> transform) {
return getRDD((PValue) getInput(transform));
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/2820534a/runners/spark/src/main/java/com/cloudera/dataflow/spark/SparkPipelineRunner.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/com/cloudera/dataflow/spark/SparkPipelineRunner.java b/runners/spark/src/main/java/com/cloudera/dataflow/spark/SparkPipelineRunner.java
index 36685c3..792888d 100644
--- a/runners/spark/src/main/java/com/cloudera/dataflow/spark/SparkPipelineRunner.java
+++ b/runners/spark/src/main/java/com/cloudera/dataflow/spark/SparkPipelineRunner.java
@@ -183,15 +183,16 @@ public final class SparkPipelineRunner extends PipelineRunner<EvaluationResult>
doVisitTransform(node);
}
- private <PT extends PTransform> void doVisitTransform(TransformTreeNode node) {
+ private <PT extends PTransform<? super PInput, POutput>>
+ void doVisitTransform(TransformTreeNode node) {
+ @SuppressWarnings("unchecked")
PT transform = (PT) node.getTransform();
@SuppressWarnings("unchecked")
- TransformEvaluator<PT> evaluator = (TransformEvaluator<PT>)
- TransformTranslator.getTransformEvaluator(transform.getClass());
+ Class<PT> transformClass = (Class<PT>) (Class<?>) transform.getClass();
+ TransformEvaluator<PT> evaluator = TransformTranslator.getTransformEvaluator(transformClass);
LOG.info("Evaluating {}", transform);
- AppliedPTransform<PInput, POutput, ? extends PTransform> appliedTransform =
- AppliedPTransform.of(node.getFullName(), node.getInput(), node.getOutput(),
- (PTransform) transform);
+ AppliedPTransform<PInput, POutput, PT> appliedTransform =
+ AppliedPTransform.of(node.getFullName(), node.getInput(), node.getOutput(), transform);
ctxt.setCurrentTransform(appliedTransform);
evaluator.evaluate(transform, ctxt);
ctxt.setCurrentTransform(null);
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/2820534a/runners/spark/src/main/java/com/cloudera/dataflow/spark/SparkRuntimeContext.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/com/cloudera/dataflow/spark/SparkRuntimeContext.java b/runners/spark/src/main/java/com/cloudera/dataflow/spark/SparkRuntimeContext.java
index 51db39b..ec590a9 100644
--- a/runners/spark/src/main/java/com/cloudera/dataflow/spark/SparkRuntimeContext.java
+++ b/runners/spark/src/main/java/com/cloudera/dataflow/spark/SparkRuntimeContext.java
@@ -94,8 +94,9 @@ public class SparkRuntimeContext implements Serializable {
}
public <T> AggregatorValues<T> getAggregatorValues(Aggregator<?, T> aggregator) {
- final T aggregatorValue = (T) getAggregatorValue(aggregator.getName(),
- aggregator.getCombineFn().getOutputType().getRawType());
+ @SuppressWarnings("unchecked")
+ Class<T> aggValueClass = (Class<T>) aggregator.getCombineFn().getOutputType().getRawType();
+ final T aggregatorValue = getAggregatorValue(aggregator.getName(), aggValueClass);
return new AggregatorValues<T>() {
@Override
public Collection<T> getValues() {
@@ -150,7 +151,7 @@ public class SparkRuntimeContext implements Serializable {
return coderRegistry;
}
- private Coder getCoder(Combine.CombineFn<?, ?, ?> combiner) {
+ private Coder<?> getCoder(Combine.CombineFn<?, ?, ?> combiner) {
try {
if (combiner.getClass() == Sum.SumIntegerFn.class) {
return getCoderRegistry().getDefaultCoder(TypeDescriptor.of(Integer.class));
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/2820534a/runners/spark/src/main/java/com/cloudera/dataflow/spark/TransformEvaluator.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/com/cloudera/dataflow/spark/TransformEvaluator.java b/runners/spark/src/main/java/com/cloudera/dataflow/spark/TransformEvaluator.java
index bf45d12..52842d5 100644
--- a/runners/spark/src/main/java/com/cloudera/dataflow/spark/TransformEvaluator.java
+++ b/runners/spark/src/main/java/com/cloudera/dataflow/spark/TransformEvaluator.java
@@ -19,6 +19,6 @@ import java.io.Serializable;
import com.google.cloud.dataflow.sdk.transforms.PTransform;
-public interface TransformEvaluator<PT extends PTransform> extends Serializable {
+public interface TransformEvaluator<PT extends PTransform<?, ?>> extends Serializable {
void evaluate(PT transform, EvaluationContext context);
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/2820534a/runners/spark/src/main/java/com/cloudera/dataflow/spark/TransformTranslator.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/com/cloudera/dataflow/spark/TransformTranslator.java b/runners/spark/src/main/java/com/cloudera/dataflow/spark/TransformTranslator.java
index 2689424..b0fd4a3 100644
--- a/runners/spark/src/main/java/com/cloudera/dataflow/spark/TransformTranslator.java
+++ b/runners/spark/src/main/java/com/cloudera/dataflow/spark/TransformTranslator.java
@@ -633,12 +633,12 @@ public final class TransformTranslator {
EVALUATORS.put(Window.Bound.class, window());
}
- public static <PT extends PTransform> boolean hasTransformEvaluator(Class<PT> clazz) {
+ public static <PT extends PTransform<?, ?>> boolean hasTransformEvaluator(Class<PT> clazz) {
return EVALUATORS.containsKey(clazz);
}
- public static <PT extends PTransform> TransformEvaluator<PT> getTransformEvaluator(Class<PT>
- clazz) {
+ public static <PT extends PTransform<?, ?>> TransformEvaluator<PT> getTransformEvaluator(Class<PT>
+ clazz) {
@SuppressWarnings("unchecked")
TransformEvaluator<PT> transform = (TransformEvaluator<PT>) EVALUATORS.get(clazz);
if (transform == null) {
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/2820534a/runners/spark/src/test/java/com/cloudera/dataflow/spark/HadoopFileFormatPipelineTest.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/test/java/com/cloudera/dataflow/spark/HadoopFileFormatPipelineTest.java b/runners/spark/src/test/java/com/cloudera/dataflow/spark/HadoopFileFormatPipelineTest.java
index 9aa634e..1fd8e41 100644
--- a/runners/spark/src/test/java/com/cloudera/dataflow/spark/HadoopFileFormatPipelineTest.java
+++ b/runners/spark/src/test/java/com/cloudera/dataflow/spark/HadoopFileFormatPipelineTest.java
@@ -35,6 +35,7 @@ import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.SequenceFile.Writer;
import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.junit.Before;
import org.junit.Rule;
@@ -63,11 +64,15 @@ public class HadoopFileFormatPipelineTest {
populateFile();
Pipeline p = Pipeline.create(PipelineOptionsFactory.create());
- PCollection<KV<IntWritable, Text>> input = (PCollection<KV<IntWritable, Text>>)
- p.apply(HadoopIO.Read.from(inputFile.getAbsolutePath())
- .withFormatClass(SequenceFileInputFormat.class)
- .withKeyClass(IntWritable.class)
- .withValueClass(Text.class));
+ @SuppressWarnings("unchecked")
+ Class<? extends FileInputFormat<IntWritable, Text>> inputFormatClass =
+ (Class<? extends FileInputFormat<IntWritable, Text>>) (Class<?>) SequenceFileInputFormat.class;
+ HadoopIO.Read.Bound<IntWritable,Text> bound =
+ HadoopIO.Read.<IntWritable,Text>from(inputFile.getAbsolutePath())
+ .withKeyClass(IntWritable.class)
+ .withValueClass(Text.class)
+ .withFormatClass(inputFormatClass);
+ PCollection<KV<IntWritable, Text>> input = p.apply(bound);
input.apply(ParDo.of(new TabSeparatedString()))
.apply(TextIO.Write.to(outputFile.getAbsolutePath()).withoutSharding());
EvaluationResult res = SparkPipelineRunner.create().run(p);