Re: How to use Dimension table in Flink TableAPI with StreamExecutionEnvironment ?

[Fabian Hueske]

Hi,

Liu and Hequn are right.
You need to pass at least one parameter into the table function, i.e.,

select t.col_1 from test t left join lateral
table(dim_test(SOME_ATTRIBUTE)) b on true

Best, Fabian

2018-02-24 13:24 GMT+01:00 ZhenBao Ye :

> hi，i was use 1.4.0。
>
> Yezhenbao
>
> > 在 2018年2月24日，17:55，Renjie Liu  写道：
> >
> > Hi, according to flink doc, it seems that you need to pass at least one
> > argument into the table function.
> >
> >> On Fri, Feb 23, 2018 at 12:35 AM 叶振宝 <827295...@qq.com> wrote:
> >>
> >> Hey, I am new to flink and I have a question and want to see if anyone
> can
> >> help here.
> >>
> >> How to use Dimension table in Flink TableAPI with
> >> StreamExecutionEnvironment ?
> >>
> >> I use TableFuncion to deal this question, but it have some problem in
> debug
> >> like this:
> >> LogicalProject(col_1=[$0])
> >>  LogicalJoin(condition=[true], joinType=[left])
> >>LogicalTableScan(table=[[test]])
> >>LogicalTableFunctionScan(invocation=[dim_test()],
> >> rowType=[RecordType(VARCHAR(65536) col, VARCHAR(65536) name)],
> >> elementType=[class [Ljava.lang.Object;])
> >>
> >> This exception indicates that the query uses an unsupported SQL feature.
> >> Please check the documentation for the set of currently supported SQL
> >> features.
> >>at
> >> org.apache.flink.table.api.TableEnvironment.runVolcanoPlanner(
> TableEnvironment.scala:274)
> >>at
> >> org.apache.flink.table.api.StreamTableEnvironment.optimize(
> StreamTableEnvironment.scala:674)
> >>at
> >> org.apache.flink.table.api.StreamTableEnvironment.translate(
> StreamTableEnvironment.scala:730)
> >>at
> >> org.apache.flink.table.api.StreamTableEnvironment.writeToSink(
> StreamTableEnvironment.scala:216)
> >>at
> >> org.apache.flink.table.api.TableEnvironment.insertInto(
> TableEnvironment.scala:692)
> >>at
> >> com.bigdata.stream.streamsql.FlinkSqlTest.main(FlinkSqlTest.java:64)
> >>
> >> SQL : select t.col_1 from test t left join lateral table(dim_test()) b
> on
> >> true
> >>
> >> Main Code:
> >> public static void main(String[] args) throws Exception {
> >>String sql = "select t.col_1 from test t left join lateral
> >> table(dim_test()) b on true";
> >>StreamExecutionEnvironment streamEnv =
> >> StreamExecutionEnvironment.getExecutionEnvironment();
> >>StreamTableEnvironment stEnv =
> >> TableEnvironment.getTableEnvironment(streamEnv);
> >>Properties kafkaProps = new Properties();
> >>kafkaProps.setProperty("zookeeper.connect", "localhost:2181");
> >>kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
> >>kafkaProps.setProperty("group.id", "test");
> >>Kafka010JsonTableSource tableSource =
> >> Kafka010JsonTableSource.builder()
> >>.forTopic("test")
> >>.withKafkaProperties(kafkaProps)
> >>.withSchema(TableSchema.builder()
> >>.field("col_1", Types.STRING)
> >>.field("col_2",Types.STRING).build())
> >>.build();
> >>stEnv.registerTableSource("test", tableSource);
> >>String[] columns = {"col","name"};
> >>TypeInformation[] typeInformations =
> >> {TypeInformation.of(String.class),TypeInformation.of(String.class)};
> >>TableSchema tableSchema = new
> >> TableSchema(columns,typeInformations);
> >>Map context = new HashMap<>();
> >>context.put("mysql.url","jdbc:mysql://localhost:3306/test");
> >>context.put("mysql.driver","com.mysql.jdbc.Driver");
> >>context.put("mysql.user","test");
> >>context.put("mysql.password","test");
> >>context.put("mysql.table","dim_test");
> >>StreamSqlDim dim = new
> >> MySqlDimFactory().getInstance(tableSchema,new
> StreamSqlContext(context));
> >>stEnv.registerFunction("dim_test",dim);
> >>
> >>String[] outColumns = {"col"};
> >>TypeInformation[] outType = {TypeInformation.of(String.class)};
> >>TableSink tableSink = new
> >> Kafka010JsonTableSink("test_out",kafkaProps);
> >>stEnv.registerTableSink("test_out",outColumns,outType,
> tableSink);
> >>Table t = stEnv.sql(sql);
> >>stEnv.insertInto(t,"test_out",stEnv.queryConfig());
> >>streamEnv.execute();
> >>}
> >>
> >> MySqlDim is extends TableFunction ,and the method eval() is empty,like
> >> this:
> >> public void eval(){
> >>
> >> }
> >>
> >>
> >>
> >> --
> > Liu, Renjie
> > Software Engineer, MVAD
> >
>
>
>
>

Re: How to use Dimension table in Flink TableAPI with StreamExecutionEnvironment ?

[ZhenBao Ye]

hi，i was use 1.4.0。

Yezhenbao

> 在 2018年2月24日，17:55，Renjie Liu  写道：
> 
> Hi, according to flink doc, it seems that you need to pass at least one
> argument into the table function.
> 
>> On Fri, Feb 23, 2018 at 12:35 AM 叶振宝 <827295...@qq.com> wrote:
>> 
>> Hey, I am new to flink and I have a question and want to see if anyone can
>> help here.
>> 
>> How to use Dimension table in Flink TableAPI with
>> StreamExecutionEnvironment ?
>> 
>> I use TableFuncion to deal this question, but it have some problem in debug
>> like this:
>> LogicalProject(col_1=[$0])
>>  LogicalJoin(condition=[true], joinType=[left])
>>LogicalTableScan(table=[[test]])
>>LogicalTableFunctionScan(invocation=[dim_test()],
>> rowType=[RecordType(VARCHAR(65536) col, VARCHAR(65536) name)],
>> elementType=[class [Ljava.lang.Object;])
>> 
>> This exception indicates that the query uses an unsupported SQL feature.
>> Please check the documentation for the set of currently supported SQL
>> features.
>>at
>> org.apache.flink.table.api.TableEnvironment.runVolcanoPlanner(TableEnvironment.scala:274)
>>at
>> org.apache.flink.table.api.StreamTableEnvironment.optimize(StreamTableEnvironment.scala:674)
>>at
>> org.apache.flink.table.api.StreamTableEnvironment.translate(StreamTableEnvironment.scala:730)
>>at
>> org.apache.flink.table.api.StreamTableEnvironment.writeToSink(StreamTableEnvironment.scala:216)
>>at
>> org.apache.flink.table.api.TableEnvironment.insertInto(TableEnvironment.scala:692)
>>at
>> com.bigdata.stream.streamsql.FlinkSqlTest.main(FlinkSqlTest.java:64)
>> 
>> SQL : select t.col_1 from test t left join lateral table(dim_test()) b on
>> true
>> 
>> Main Code:
>> public static void main(String[] args) throws Exception {
>>String sql = "select t.col_1 from test t left join lateral
>> table(dim_test()) b on true";
>>StreamExecutionEnvironment streamEnv =
>> StreamExecutionEnvironment.getExecutionEnvironment();
>>StreamTableEnvironment stEnv =
>> TableEnvironment.getTableEnvironment(streamEnv);
>>Properties kafkaProps = new Properties();
>>kafkaProps.setProperty("zookeeper.connect", "localhost:2181");
>>kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
>>kafkaProps.setProperty("group.id", "test");
>>Kafka010JsonTableSource tableSource =
>> Kafka010JsonTableSource.builder()
>>.forTopic("test")
>>.withKafkaProperties(kafkaProps)
>>.withSchema(TableSchema.builder()
>>.field("col_1", Types.STRING)
>>.field("col_2",Types.STRING).build())
>>.build();
>>stEnv.registerTableSource("test", tableSource);
>>String[] columns = {"col","name"};
>>TypeInformation[] typeInformations =
>> {TypeInformation.of(String.class),TypeInformation.of(String.class)};
>>TableSchema tableSchema = new
>> TableSchema(columns,typeInformations);
>>Map context = new HashMap<>();
>>context.put("mysql.url","jdbc:mysql://localhost:3306/test");
>>context.put("mysql.driver","com.mysql.jdbc.Driver");
>>context.put("mysql.user","test");
>>context.put("mysql.password","test");
>>context.put("mysql.table","dim_test");
>>StreamSqlDim dim = new
>> MySqlDimFactory().getInstance(tableSchema,new StreamSqlContext(context));
>>stEnv.registerFunction("dim_test",dim);
>> 
>>String[] outColumns = {"col"};
>>TypeInformation[] outType = {TypeInformation.of(String.class)};
>>TableSink tableSink = new
>> Kafka010JsonTableSink("test_out",kafkaProps);
>>stEnv.registerTableSink("test_out",outColumns,outType,tableSink);
>>Table t = stEnv.sql(sql);
>>stEnv.insertInto(t,"test_out",stEnv.queryConfig());
>>streamEnv.execute();
>>}
>> 
>> MySqlDim is extends TableFunction ,and the method eval() is empty,like
>> this:
>> public void eval(){
>> 
>> }
>> 
>> 
>> 
>> --
> Liu, Renjie
> Software Engineer, MVAD
>