You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2016/07/26 09:11:39 UTC
[1/2] flink git commit: [FLINK-4103] [table] Modify CsvTableSource to
implement StreamTableSource
Repository: flink
Updated Branches:
refs/heads/master 0360fb968 -> 110bba386
[FLINK-4103] [table] Modify CsvTableSource to implement StreamTableSource
This closes #2162.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7e309eeb
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7e309eeb
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7e309eeb
Branch: refs/heads/master
Commit: 7e309eeb5f0c6977c5c62028bfd15dd8927c62aa
Parents: 0360fb9
Author: smarthi <sm...@apache.org>
Authored: Fri Jun 24 23:55:14 2016 -0400
Committer: twalthr <tw...@apache.org>
Committed: Tue Jul 26 10:47:57 2016 +0200
----------------------------------------------------------------------
.../api/table/sources/CsvTableSource.scala | 41 ++++++++-----
.../api/table/validate/FunctionCatalog.scala | 5 +-
.../api/scala/stream/TableSourceITCase.scala | 60 +++++++++++++++++++-
3 files changed, 88 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7e309eeb/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/sources/CsvTableSource.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/sources/CsvTableSource.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/sources/CsvTableSource.scala
index 54d7718..fc3cf7e 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/sources/CsvTableSource.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/sources/CsvTableSource.scala
@@ -25,9 +25,12 @@ import org.apache.flink.api.table.{Row, TableException}
import org.apache.flink.api.table.runtime.io.RowCsvInputFormat
import org.apache.flink.api.table.typeutils.RowTypeInfo
import org.apache.flink.core.fs.Path
+import org.apache.flink.streaming.api.datastream.DataStream
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
/**
- * A [[TableSource]] for simple CSV files with a (logically) unlimited number of fields.
+ * A [[BatchTableSource]] and [[StreamTableSource]] for simple CSV files with a
+ * (logically) unlimited number of fields.
*
* @param path The path to the CSV file.
* @param fieldNames The names of the table fields.
@@ -49,7 +52,8 @@ class CsvTableSource(
ignoreFirstLine: Boolean = false,
ignoreComments: String = null,
lenient: Boolean = false)
- extends BatchTableSource[Row] {
+ extends BatchTableSource[Row]
+ with StreamTableSource[Row] {
if (fieldNames.length != fieldTypes.length) {
throw TableException("Number of field names and field types must be equal.")
@@ -59,18 +63,7 @@ class CsvTableSource(
/** Returns the data of the table as a [[DataSet]] of [[Row]]. */
override def getDataSet(execEnv: ExecutionEnvironment): DataSet[Row] = {
- val inputFormat = new RowCsvInputFormat(new Path(path), returnType, rowDelim, fieldDelim)
-
- inputFormat.setSkipFirstLineAsHeader(ignoreFirstLine)
- inputFormat.setLenient(lenient)
- if (quoteCharacter != null) {
- inputFormat.enableQuotedStringParsing(quoteCharacter)
- }
- if (ignoreComments != null) {
- inputFormat.setCommentPrefix(ignoreComments)
- }
-
- execEnv.createInput(inputFormat, returnType)
+ execEnv.createInput(createCsvInput(), returnType)
}
/** Returns the types of the table fields. */
@@ -84,4 +77,24 @@ class CsvTableSource(
/** Returns the [[RowTypeInfo]] for the return type of the [[CsvTableSource]]. */
override def getReturnType: RowTypeInfo = returnType
+
+ /** Returns the data of the table as a [[DataStream]] of [[Row]]. */
+ override def getDataStream(streamExecEnv: StreamExecutionEnvironment): DataStream[Row] = {
+ streamExecEnv.createInput(createCsvInput(), returnType)
+ }
+
+ private def createCsvInput(): RowCsvInputFormat = {
+ val inputFormat = new RowCsvInputFormat(new Path(path), returnType, rowDelim, fieldDelim)
+
+ inputFormat.setSkipFirstLineAsHeader(ignoreFirstLine)
+ inputFormat.setLenient(lenient)
+ if (quoteCharacter != null) {
+ inputFormat.enableQuotedStringParsing(quoteCharacter)
+ }
+ if (ignoreComments != null) {
+ inputFormat.setCommentPrefix(ignoreComments)
+ }
+
+ inputFormat
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7e309eeb/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala
index 726917e..c988474 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/validate/FunctionCatalog.scala
@@ -38,9 +38,8 @@ class FunctionCatalog {
* Lookup and create an expression if we find a match.
*/
def lookupFunction(name: String, children: Seq[Expression]): Expression = {
- val funcClass = functionBuilders.get(name.toLowerCase).getOrElse {
- throw new ValidationException(s"undefined function $name")
- }
+ val funcClass = functionBuilders.getOrElse(name.toLowerCase,
+ throw new ValidationException(s"undefined function $name"))
withChildren(funcClass, children)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7e309eeb/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/TableSourceITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/TableSourceITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/TableSourceITCase.scala
index 50d9a42..c82e6df 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/TableSourceITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/TableSourceITCase.scala
@@ -18,11 +18,13 @@
package org.apache.flink.api.scala.stream
+import java.io.{File, FileOutputStream, OutputStreamWriter}
+
import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
import org.apache.flink.api.scala._
import org.apache.flink.api.scala.stream.utils.StreamITCase
import org.apache.flink.api.scala.table._
-import org.apache.flink.api.table.sources.StreamTableSource
+import org.apache.flink.api.table.sources.{CsvTableSource, StreamTableSource}
import org.apache.flink.api.table.typeutils.RowTypeInfo
import org.apache.flink.api.table.{Row, TableEnvironment}
import org.apache.flink.streaming.api.datastream.DataStream
@@ -83,8 +85,64 @@ class TableSourceITCase extends StreamingMultipleProgramsTestBase {
assertEquals(expected.sorted, StreamITCase.testResults.sorted)
}
+ @Test
+ def testCsvTableSource(): Unit = {
+
+ val csvRecords = Seq(
+ "First#Id#Score#Last",
+ "Mike#1#12.3#Smith",
+ "Bob#2#45.6#Taylor",
+ "Sam#3#7.89#Miller",
+ "Peter#4#0.12#Smith",
+ "% Just a comment",
+ "Liz#5#34.5#Williams",
+ "Sally#6#6.78#Miller",
+ "Alice#7#90.1#Smith",
+ "Kelly#8#2.34#Williams"
+ )
+
+ val tempFile = File.createTempFile("csv-test", "tmp")
+ tempFile.deleteOnExit()
+ val tmpWriter = new OutputStreamWriter(new FileOutputStream(tempFile), "UTF-8")
+ tmpWriter.write(csvRecords.mkString("$"))
+ tmpWriter.close()
+
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val tEnv = TableEnvironment.getTableEnvironment(env)
+ StreamITCase.testResults = mutable.MutableList()
+
+ val csvTable = new CsvTableSource(
+ tempFile.getAbsolutePath,
+ Array("first", "id", "score", "last"),
+ Array(
+ BasicTypeInfo.STRING_TYPE_INFO,
+ BasicTypeInfo.INT_TYPE_INFO,
+ BasicTypeInfo.DOUBLE_TYPE_INFO,
+ BasicTypeInfo.STRING_TYPE_INFO
+ ),
+ fieldDelim = "#",
+ rowDelim = "$",
+ ignoreFirstLine = true,
+ ignoreComments = "%"
+ )
+
+ tEnv.registerTableSource("csvTable", csvTable)
+ tEnv.sql(
+ "SELECT STREAM last, score, id FROM csvTable WHERE id < 4 ")
+ .toDataStream[Row]
+ .addSink(new StreamITCase.StringSink)
+
+ env.execute()
+
+ val expected = mutable.MutableList(
+ "Smith,12.3,1",
+ "Taylor,45.6,2",
+ "Miller,7.89,3")
+ assertEquals(expected.sorted, StreamITCase.testResults.sorted)
+ }
}
+
class TestStreamTableSource(val numRecords: Int) extends StreamTableSource[Row] {
val fieldTypes: Array[TypeInformation[_]] = Array(
[2/2] flink git commit: [FLINK-4103] [table] Add CsvTableSource docs
and Java accessibility
Posted by tw...@apache.org.
[FLINK-4103] [table] Add CsvTableSource docs and Java accessibility
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/110bba38
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/110bba38
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/110bba38
Branch: refs/heads/master
Commit: 110bba386a3ac89a8f312cd4d1405571eee9d4b5
Parents: 7e309ee
Author: twalthr <tw...@apache.org>
Authored: Tue Jul 26 10:44:24 2016 +0200
Committer: twalthr <tw...@apache.org>
Committed: Tue Jul 26 11:03:29 2016 +0200
----------------------------------------------------------------------
docs/apis/table.md | 83 ++++++++++++++++++++
.../api/table/sources/CsvTableSource.scala | 20 ++++-
2 files changed, 99 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/110bba38/docs/apis/table.md
----------------------------------------------------------------------
diff --git a/docs/apis/table.md b/docs/apis/table.md
index cf9133c..14439ed 100644
--- a/docs/apis/table.md
+++ b/docs/apis/table.md
@@ -252,6 +252,89 @@ tableEnvironment.registerTableSource("kafka-source", kafkaTableSource);
Table result = tableEnvironment.ingest("kafka-source");
```
+#### CsvTableSource
+
+The `CsvTableSource` is already included in `flink-table` without additional dependecies.
+
+It can be configured with the following properties:
+
+ - `path` The path to the CSV file, required.
+ - `fieldNames` The names of the table fields, required.
+ - `fieldTypes` The types of the table fields, required.
+ - `fieldDelim` The field delimiter, `","` by default.
+ - `rowDelim` The row delimiter, `"\n"` by default.
+ - `quoteCharacter` An optional quote character for String values, `null` by default.
+ - `ignoreFirstLine` Flag to ignore the first line, `false` by default.
+ - `ignoreComments` An optional prefix to indicate comments, `null` by default.
+ - `lenient` Flag to skip records with parse error instead to fail, `false` by default.
+
+You can create the source as follows:
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+CsvTableSource csvTableSource = new CsvTableSource(
+ "/path/to/your/file.csv",
+ new String[] { "name", "id", "score", "comments" },
+ new TypeInformation<?>[] {
+ Types.STRING(),
+ Types.INT(),
+ Types.DOUBLE(),
+ Types.STRING()
+ },
+ "#", // fieldDelim
+ "$", // rowDelim
+ null, // quoteCharacter
+ true, // ignoreFirstLine
+ "%", // ignoreComments
+ false); // lenient
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+val csvTableSource = new CsvTableSource(
+ "/path/to/your/file.csv",
+ Array("name", "id", "score", "comments"),
+ Array(
+ Types.STRING,
+ Types.INT,
+ Types.DOUBLE,
+ Types.STRING
+ ),
+ fieldDelim = "#",
+ rowDelim = "$",
+ ignoreFirstLine = true,
+ ignoreComments = "%")
+{% endhighlight %}
+</div>
+</div>
+
+You can work with the Table as explained in the rest of the Table API guide in both stream and batch `TableEnvironment`s:
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+tableEnvironment.registerTableSource("mycsv", csvTableSource);
+
+Table streamTable = streamTableEnvironment.ingest("mycsv");
+
+Table batchTable = batchTableEnvironment.scan("mycsv");
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+tableEnvironment.registerTableSource("mycsv", csvTableSource)
+
+val streamTable = streamTableEnvironment.ingest("mycsv")
+
+val batchTable = batchTableEnvironment.scan("mycsv")
+{% endhighlight %}
+</div>
+</div>
+
+
Table API
----------
The Table API provides methods to apply relational operations on DataSets and Datastreams both in Scala and Java.
http://git-wip-us.apache.org/repos/asf/flink/blob/110bba38/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/sources/CsvTableSource.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/sources/CsvTableSource.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/sources/CsvTableSource.scala
index fc3cf7e..40fdf82 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/sources/CsvTableSource.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/sources/CsvTableSource.scala
@@ -35,11 +35,11 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
* @param path The path to the CSV file.
* @param fieldNames The names of the table fields.
* @param fieldTypes The types of the table fields.
- * @param fieldDelim The field delimiter, ',' by default.
- * @param rowDelim The row delimiter, '\n' by default.
- * @param quoteCharacter An optional quote character for String values, disabled by default.
+ * @param fieldDelim The field delimiter, "," by default.
+ * @param rowDelim The row delimiter, "\n" by default.
+ * @param quoteCharacter An optional quote character for String values, null by default.
* @param ignoreFirstLine Flag to ignore the first line, false by default.
- * @param ignoreComments An optional prefix to indicate comments, disabled by default.
+ * @param ignoreComments An optional prefix to indicate comments, null by default.
* @param lenient Flag to skip records with parse error instead to fail, false by default.
*/
class CsvTableSource(
@@ -55,6 +55,18 @@ class CsvTableSource(
extends BatchTableSource[Row]
with StreamTableSource[Row] {
+ /**
+ * A [[BatchTableSource]] and [[StreamTableSource]] for simple CSV files with a
+ * (logically) unlimited number of fields.
+ *
+ * @param path The path to the CSV file.
+ * @param fieldNames The names of the table fields.
+ * @param fieldTypes The types of the table fields.
+ */
+ def this(path: String, fieldNames: Array[String], fieldTypes: Array[TypeInformation[_]]) =
+ this(path, fieldNames, fieldTypes, CsvInputFormat.DEFAULT_FIELD_DELIMITER,
+ CsvInputFormat.DEFAULT_LINE_DELIMITER, null, false, null, false)
+
if (fieldNames.length != fieldTypes.length) {
throw TableException("Number of field names and field types must be equal.")
}