[FLINK-9555]Support table api in scala shell #7121

shuiqiangchen · 2018-11-16T07:48:24Z

What is the purpose of the change

make table api available in scala shell so that user can experience table api in interactive way.

Brief change log

add flink-table dependency to flink-scala-shell module.
The scala shell provides a BatchTableEnvironment and a StreamTableEnvironment initially when starting

Verifying this change

This change is already covered by existing tests

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no) yes
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no) no
The serializers: (yes / no / don't know) no
The runtime per-record code paths (performance sensitive): (yes / no / don't know) no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / no / don't know) no
The S3 file system connector: (yes / no / don't know) no

Documentation

Does this pull request introduce a new feature? (yes / no) yes
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented) not documented

sunjincheng121

Thanks for the PR @shuiqiangchen !
In addition to a few comments I left, I have two suggestions as follows：

Add test case for table/sql in ScalaShellITCase.
Add the welcome print about sql/table examples in printWelcome method.
Best,
Jincheng

sunjincheng121 · 2018-11-19T02:17:59Z

flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkILoop.scala

@@ -96,6 +97,12 @@ class FlinkILoop(
    (scalaBenv,scalaSenv)
  }

+  val (scalaBTEnv: BatchTableEnvironment, scalaSTEnv: StreamTableEnvironment) = {


I suggest that merge two local environment init logic as follows:

// local environment val ( scalaBenv: ExecutionEnvironment, scalaSenv: StreamExecutionEnvironment, scalaBTabEnv: BatchTableEnvironment, scalaSTabEnv: StreamTableEnvironment) = { val scalaBenv = new ExecutionEnvironment(remoteBenv) val scalaSenv = new StreamExecutionEnvironment(remoteSenv) val scalaBTabEnv = TableEnvironment.getTableEnvironment(scalaBenv) val scalaSTabEnv = TableEnvironment.getTableEnvironment(scalaSenv) (scalaBenv, scalaSenv, scalaBTabEnv, scalaSTabEnv) }```

Thanks! It seems much more well organized.

sunjincheng121 · 2018-11-19T02:19:31Z

flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkILoop.scala

@@ -139,7 +146,9 @@ class FlinkILoop(
    "org.apache.flink.api.scala._",
    "org.apache.flink.api.scala.utils._",
    "org.apache.flink.streaming.api.scala._",
-    "org.apache.flink.streaming.api.windowing.time._"
+    "org.apache.flink.streaming.api.windowing.time._",


Why we should import windowing.time?

This is important to define windows or interval joins in DataStream API.

sunjincheng121 · 2018-11-19T02:46:54Z

flink-dist/pom.xml

@@ -319,7 +319,6 @@ under the License.
 			<groupId>org.apache.flink</groupId>
 			<artifactId>flink-table_${scala.binary.version}</artifactId>
 			<version>${project.version}</version>
-			<scope>provided</scope>


I suggest that revert this change, because flink-dist is the whole release strategy of flink. At present, we package the flink-table module to the opt directory. The minimum change I suggest is to modify the scala-shell CLASSPATH and include opt/opt/flink-tableXXX.jar in the CLASSPATH， to support tables in scala-shell, What do you think?

I agree that this change should be reverted but flink-table should remain in /opt. If a user wants to use flink-table (or any connector for that matter) the required jars should be copied manually to /lib.

Thank you for your valuable suggestions! I have modified the scala shell start script to add the flink-table_*.jar to FLINK_CLASSPATH. In this way, we have no need to change the flink-dist module.

zjffdu · 2018-11-27T02:55:55Z

flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkILoop.scala


    * val dataSet = benv.readTextFile("/path/to/data")
    * dataSet.writeAsText("/path/to/output")
    * benv.execute("My batch program")
-
+    * val batchTable = btenv.sqlQuery("SELECT * FROM tableName")


Add example of registering table via benv then query it via btenv.

zjffdu · 2018-11-27T02:56:12Z

flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkILoop.scala


    * val dataStream = senv.fromElements(1, 2, 3, 4)
    * dataStream.countWindowAll(2).sum(0).print()
    * senv.execute("My streaming program")
-
+    * val streamTable = stenv.sqlQuery("SELECT * FROM tableName")


Add example of registering table via senv then query it via stenv.

zjffdu · 2018-11-27T02:56:48Z

flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkILoop.scala


    * val dataSet = benv.readTextFile("/path/to/data")
    * dataSet.writeAsText("/path/to/output")
    * benv.execute("My batch program")
-
+    * val batchTable = btenv.sqlQuery("SELECT * FROM tableName")
    HINT: You can use print() on a DataSet to print the contents to the shell.


Add hint of how to see the table output in scala-shell

zjffdu · 2018-11-27T02:56:54Z

flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkILoop.scala


    * val dataStream = senv.fromElements(1, 2, 3, 4)
    * dataStream.countWindowAll(2).sum(0).print()
    * senv.execute("My streaming program")
-
+    * val streamTable = stenv.sqlQuery("SELECT * FROM tableName")
    HINT: You can only print a DataStream to the shell in local mode.


Add hint of how to see the table output in scala-shell

zjffdu · 2018-11-27T02:58:41Z

Thanks @shuiqiangchen , I left a few comments. And agree with @sunjincheng121 , we need to add test

zjffdu · 2018-11-27T03:54:48Z

flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkILoop.scala

@@ -139,7 +147,8 @@ class FlinkILoop(
    "org.apache.flink.api.scala._",
    "org.apache.flink.api.scala.utils._",
    "org.apache.flink.streaming.api.scala._",
-    "org.apache.flink.streaming.api.windowing.time._"
+    "org.apache.flink.table.api._",
+    "org.apache.flink.table.api.scala._"


Could we also import org.apache.flink.types.Row which is used very frequently I believe ?

Agree +1，I will add it.

shuiqiangchen · 2018-11-28T12:09:28Z

Hi All, Thank you for your attention! I have pushed new codes according to your suggestions. Looking forward to your guidance.

zjffdu · 2018-11-30T03:23:58Z

flink-scala-shell/src/test/scala/org/apache/flink/api/scala/ScalaShellITCase.scala

+        |    data.+=((1, 1L, "Hi"))
+        |    data.+=((2, 2L, "Hello"))
+        |    data.+=((3, 2L, "Hello world"))
+        |val t = benv.fromCollection(data).toTable(btenv, 'a, 'b, 'c).select("*").where('a % 2 === 1 )


MutableList is not necessary here. Use this instead

val data = Seq( (1, 1L, "Hi), (2, 2L, "Hello"), (3, 2L, "Hello world") )

zjffdu · 2018-11-30T03:26:12Z

flink-scala-shell/src/test/scala/org/apache/flink/api/scala/ScalaShellITCase.scala

+    val output = processInShell(input)
+    Assert.assertFalse(output.contains("failed"))
+    Assert.assertFalse(output.contains("error"))
+    Assert.assertFalse(output.contains("Exception"))


We also need to verify the output contains "Hi", "Hello world" but not "Hello".

zjffdu · 2018-11-30T03:26:40Z

flink-scala-shell/src/test/scala/org/apache/flink/api/scala/ScalaShellITCase.scala

+    val output = processInShell(input)
+    Assert.assertFalse(output.contains("failed"))
+    Assert.assertFalse(output.contains("error"))
+    Assert.assertFalse(output.contains("Exception"))


We also need to verify the output contains the correct word count result.

twalthr · 2018-11-30T13:27:39Z

flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkILoop.scala

@@ -139,7 +148,9 @@ class FlinkILoop(
    "org.apache.flink.api.scala._",
    "org.apache.flink.api.scala.utils._",
    "org.apache.flink.streaming.api.scala._",
-    "org.apache.flink.streaming.api.windowing.time._"


Why did you remove this line?

Sorry, I misunderstood, it should be import initially to support window operations.

sunjincheng121 · 2018-12-03T23:12:12Z

flink-scala-shell/start-script/start-scala-shell.sh

@@ -52,6 +52,13 @@ bin=`cd "$bin"; pwd`

 FLINK_CLASSPATH=`constructFlinkClassPath`

+#adding flink table jar into class path


#adding -> # adding
# Append flink-table jar into class path ?

zjffdu · 2018-12-04T08:15:32Z

@shuiqiangchen Travis is failed, could you check that ? Thanks

shuiqiangchen · 2018-12-04T08:36:32Z

@shuiqiangchen Travis is failed, could you check that ? Thanks

Thanks! It's the matter of checkStyle. I have fixed it.

shuiqiangchen · 2018-12-05T04:00:07Z

Hi all, I have updated some codes following your suggestions. Please review it in your convenience.

sunjincheng121

Hi @shuiqiangchen Thanks for the update!
From the points of my view the PR is good to merge after you address the last comment.
+1 to merged
Bests,
Jincheng

sunjincheng121 · 2018-12-05T05:28:13Z

flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkILoop.scala

+    * btenv.registerTable("tableName", batchTable)
+    * val result = btenv.sqlQuery("SELECT * FROM tableName").collect
+    HINT: You can use print() on a DataSet to print the contents and collect()
+    a sql query result back to the shell.


contents and collect() -> contents or collect() ?

sunjincheng121 · 2018-12-06T02:29:55Z

Thanks for the update！ @shuiqiangchen
Merging...

This closes apache#7121

This closes #7121

This closes apache#7121

[FLINK-9555]Support table api in scala shell

bb2f3df

sunjincheng121 reviewed Nov 19, 2018

View reviewed changes

[FLINK-9555]Support table api in scala shell

0648aff

zjffdu reviewed Nov 27, 2018

View reviewed changes

[FLINK-9555]Support table api in scala shell

66beade

zjffdu reviewed Nov 30, 2018

View reviewed changes

twalthr self-assigned this Nov 30, 2018

twalthr reviewed Nov 30, 2018

View reviewed changes

sunjincheng121 reviewed Dec 3, 2018

View reviewed changes

[FLINK-9555]Support table api in scala shell

cb885c7

[FLINK-9555]Support table api in scala shell

0d913e0

sunjincheng121 reviewed Dec 5, 2018

View reviewed changes

[FLINK-9555]Support table api in scala shell

9a77391

sunjincheng121 pushed a commit to sunjincheng121/flink that referenced this pull request Dec 6, 2018

[FLINK-9555][scala-shell] Support table api in scala shell.

543da53

This closes apache#7121

asfgit closed this in b3a378a Dec 6, 2018

asfgit pushed a commit that referenced this pull request Dec 6, 2018

[FLINK-9555][scala-shell] Support table api in scala shell.

efc73a8

This closes #7121

sunjincheng121 self-assigned this Dec 6, 2018

zentol pushed a commit to zentol/flink that referenced this pull request Dec 6, 2018

[FLINK-9555][scala-shell] Support table api in scala shell.

124ef53

This closes apache#7121

tisonkun pushed a commit to tisonkun/flink that referenced this pull request Jan 17, 2019

[FLINK-9555][scala-shell] Support table api in scala shell.

979fd58

This closes apache#7121

rmetzger added the component=ScalaShell label Mar 18, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-9555]Support table api in scala shell #7121

[FLINK-9555]Support table api in scala shell #7121

shuiqiangchen commented Nov 16, 2018

sunjincheng121 left a comment

sunjincheng121 Nov 19, 2018

shuiqiangchen Nov 27, 2018

sunjincheng121 Nov 19, 2018

twalthr Nov 30, 2018

sunjincheng121 Nov 19, 2018

zentol Nov 19, 2018

shuiqiangchen Nov 27, 2018

zjffdu Nov 27, 2018

zjffdu Nov 27, 2018

zjffdu Nov 27, 2018

zjffdu Nov 27, 2018

zjffdu commented Nov 27, 2018

zjffdu Nov 27, 2018

shuiqiangchen Nov 27, 2018

shuiqiangchen commented Nov 28, 2018

zjffdu Nov 30, 2018 •

edited

zjffdu Nov 30, 2018

zjffdu Nov 30, 2018

sunjincheng121 Dec 3, 2018

twalthr Nov 30, 2018

shuiqiangchen Dec 3, 2018

sunjincheng121 Dec 3, 2018

zjffdu commented Dec 4, 2018

shuiqiangchen commented Dec 4, 2018

shuiqiangchen commented Dec 5, 2018

sunjincheng121 left a comment

sunjincheng121 Dec 5, 2018

sunjincheng121 commented Dec 6, 2018

		@@ -52,6 +52,13 @@ bin=`cd "$bin"; pwd`

		FLINK_CLASSPATH=`constructFlinkClassPath`

		#adding flink table jar into class path

[FLINK-9555]Support table api in scala shell #7121

[FLINK-9555]Support table api in scala shell #7121

Conversation

shuiqiangchen commented Nov 16, 2018

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

sunjincheng121 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zjffdu commented Nov 27, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shuiqiangchen commented Nov 28, 2018

zjffdu Nov 30, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zjffdu commented Dec 4, 2018

shuiqiangchen commented Dec 4, 2018

shuiqiangchen commented Dec 5, 2018

sunjincheng121 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sunjincheng121 commented Dec 6, 2018

zjffdu Nov 30, 2018 •

edited