[FLINK-2692] Untangle CsvInputFormat

This closes apache#1266
VRDate · Nov 18, 2015 · bd61f2d · bd61f2d
1 parent fc6fec7
commit bd61f2d
Show file tree

Hide file tree

Showing 18 changed files with 573 additions and 599 deletions.
diff --git a/...m-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountPojo.java b/...m-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountPojo.java
@@ -20,6 +20,7 @@
 import backtype.storm.topology.IRichBolt;
 
 import org.apache.flink.api.java.io.CsvInputFormat;
+import org.apache.flink.api.java.io.PojoCsvInputFormat;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.PojoTypeInfo;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
@@ -119,7 +120,7 @@ private static DataStream<Sentence> getTextDataStream(final StreamExecutionEnvir
  // read the text file from given input path
  PojoTypeInfo<Sentence> sourceType = (PojoTypeInfo<Sentence>) TypeExtractor
  .getForObject(new Sentence(""));
- return env.createInput(new CsvInputFormat<Sentence>(new Path(
+ return env.createInput(new PojoCsvInputFormat<Sentence>(new Path(
  textPath), CsvInputFormat.DEFAULT_LINE_DELIMITER,
  CsvInputFormat.DEFAULT_LINE_DELIMITER, sourceType),
  sourceType);

diff --git a/...mples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountWithNames.java b/...mples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountWithNames.java
@@ -21,6 +21,7 @@
 import backtype.storm.tuple.Fields;
 
 import org.apache.flink.api.java.io.CsvInputFormat;
+import org.apache.flink.api.java.io.TupleCsvInputFormat;
 import org.apache.flink.api.java.tuple.Tuple;
 import org.apache.flink.api.java.tuple.Tuple1;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -122,7 +123,7 @@ private static DataStream<Tuple1<String>> getTextDataStream(final StreamExecutio
  // read the text file from given input path
  TupleTypeInfo<Tuple1<String>> sourceType = (TupleTypeInfo<Tuple1<String>>)TypeExtractor
  .getForObject(new Tuple1<String>(""));
- return env.createInput(new CsvInputFormat<Tuple1<String>>(new Path(
+ return env.createInput(new TupleCsvInputFormat<Tuple1<String>>(new Path(
  textPath), CsvInputFormat.DEFAULT_LINE_DELIMITER,
  CsvInputFormat.DEFAULT_LINE_DELIMITER, sourceType),
  sourceType);

diff --git a/flink-java/src/main/java/org/apache/flink/api/java/io/CommonCsvInputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/io/CommonCsvInputFormat.java