[
https://issues.apache.org/jira/browse/FLINK-6355?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kaibo Zhou updated FLINK-6355:
------------------------------
Description:
Currently {{TableEnvironment}} only supports register ScalarFunction, the Java
users cannot use {{TableEnvironment}} to register {{TableFunction}}.
This is the test code:
{code:title=TableFunc0.java|borderStyle=solid}
public class TableFunc0 extends TableFunction<String> {
public void eval(String str) {
for (String s : str.split(" ")) {
collect(s);
}
}
}
{code}
{code:title=WordCountExample.java|borderStyle=solid}
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.Types;
import org.apache.flink.table.sinks.CsvTableSink;
import org.apache.flink.table.sources.CsvTableSource;
import org.apache.flink.table.sources.TableSource;
import org.apache.flink.types.Row;
public class WordCountExample {
public void build(TableEnvironment tableEnv, java.util.Properties
properties) {
TableSource<Row> tableSource = new CsvTableSource("/path/to/csv/", new
String[]{"a", "b"}, new TypeInformation[]{Types.STRING(), Types.STRING()});
tableEnv.registerTableSource("csv_source", tableSource);
TableFunc0 tableFunc0 = new TableFunc0();
// expected
tableEnv.registerFunction("func0", tableFunc0); // will compile error
here
// not expected
((StreamTableEnvironment) tableEnv).registerFunction("func0",
tableFunc0); // also can work, but not what we wanted
tableEnv.scan("csv_source")
.select("a, b")
.join("func0(a) as c")
.select("b, c")
.writeToSink(new CsvTableSink("/path/to/result", ","));
}
}
{code}
There will be compilation errors in tableEnv.registerFunction： "Found
xx.xxx.TableFunc0，required org.apache.flink.table.functions.ScalarFunction"。
I did some testing, only Java users have this problem.
----
Why the user interface only has {{TableEnvironment}}, but not
{{StreamTableEnvironment}} or {{BatchStreamTableEnvironment}} ？
We want to eliminate most of the differences between the batch and the stream
execution environment, and support the use of the {{TableEnvironment}} to write
programs only. So that users only need to write a code, and can use the command
parameters to determine run in the batch or stream processing environment.This
is also the advantage of flink which can unify batch and stream processing.
was:
Currently {{TableEnvironment}} only supports register ScalarFunction, the Java
users cannot use {{TableEnvironment}} to register {{TableFunction}}.
This is the test code:
TableFunc0.java
{quote}
public class TableFunc0 extends TableFunction<String> {
public void eval(String str) {
for (String s : str.split(" ")) {
collect(s);
}
}
}
{quote}
WordCountExample.java
{quote}
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.Types;
import org.apache.flink.table.sinks.CsvTableSink;
import org.apache.flink.table.sources.CsvTableSource;
import org.apache.flink.table.sources.TableSource;
import org.apache.flink.types.Row;
public class WordCountExample {
public void build(TableEnvironment tableEnv, java.util.Properties
properties) {
TableSource<Row> tableSource = new CsvTableSource("/path/to/csv/", new
String[]{"a", "b"}, new TypeInformation[]{Types.STRING(), Types.STRING()});
tableEnv.registerTableSource("csv_source", tableSource);
TableFunc0 tableFunc0 = new TableFunc0();
tableEnv.registerFunction("func0", tableFunc0); // compile error here
((StreamTableEnvironment) tableEnv).registerFunction("func0",
tableFunc0); // also can work, but not what we want
tableEnv.scan("csv_source")
.select("a, b")
.join("func0(a) as c")
.select("b, c")
.writeToSink(new CsvTableSink("/path/to/result", ","));
}
}
{quote}
There will be compilation errors in tableEnv.registerFunction： "Found
xx.xxx.TableFunc0，required org.apache.flink.table.functions.ScalarFunction"。
I did some testing, only Java users have this problem.
> TableEnvironment support register TableFunction
> -----------------------------------------------
>
> Key: FLINK-6355
> URL: https://issues.apache.org/jira/browse/FLINK-6355
> Project: Flink
> Issue Type: Improvement
> Components: Table API & SQL
> Reporter: Kaibo Zhou
> Assignee: Kaibo Zhou
>
> Currently {{TableEnvironment}} only supports register ScalarFunction, the
> Java users cannot use {{TableEnvironment}} to register {{TableFunction}}.
> This is the test code:
> {code:title=TableFunc0.java|borderStyle=solid}
> public class TableFunc0 extends TableFunction<String> {
> public void eval(String str) {
> for (String s : str.split(" ")) {
> collect(s);
> }
> }
> }
> {code}
> {code:title=WordCountExample.java|borderStyle=solid}
> import org.apache.flink.api.common.typeinfo.TypeInformation;
> import org.apache.flink.table.api.java.StreamTableEnvironment;
> import org.apache.flink.table.api.TableEnvironment;
> import org.apache.flink.table.api.Types;
> import org.apache.flink.table.sinks.CsvTableSink;
> import org.apache.flink.table.sources.CsvTableSource;
> import org.apache.flink.table.sources.TableSource;
> import org.apache.flink.types.Row;
> public class WordCountExample {
> public void build(TableEnvironment tableEnv, java.util.Properties
> properties) {
> TableSource<Row> tableSource = new CsvTableSource("/path/to/csv/",
> new String[]{"a", "b"}, new TypeInformation[]{Types.STRING(),
> Types.STRING()});
> tableEnv.registerTableSource("csv_source", tableSource);
> TableFunc0 tableFunc0 = new TableFunc0();
> // expected
> tableEnv.registerFunction("func0", tableFunc0); // will compile error
> here
> // not expected
> ((StreamTableEnvironment) tableEnv).registerFunction("func0",
> tableFunc0); // also can work, but not what we wanted
> tableEnv.scan("csv_source")
> .select("a, b")
> .join("func0(a) as c")
> .select("b, c")
> .writeToSink(new CsvTableSink("/path/to/result", ","));
> }
> }
> {code}
> There will be compilation errors in tableEnv.registerFunction： "Found
> xx.xxx.TableFunc0，required org.apache.flink.table.functions.ScalarFunction"。
> I did some testing, only Java users have this problem.
> ----
> Why the user interface only has {{TableEnvironment}}, but not
> {{StreamTableEnvironment}} or {{BatchStreamTableEnvironment}} ？
> We want to eliminate most of the differences between the batch and the stream
> execution environment, and support the use of the {{TableEnvironment}} to
> write programs only. So that users only need to write a code, and can use the
> command parameters to determine run in the batch or stream processing
> environment.This is also the advantage of flink which can unify batch and
> stream processing.
>
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
