You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Leonard Xu (Jira)" <ji...@apache.org> on 2020/10/20 12:54:00 UTC

[jira] [Closed] (FLINK-19731) 使用了批量执行SQL的api,job执行的时候,DataStream 不触发

     [ https://issues.apache.org/jira/browse/FLINK-19731?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Leonard Xu closed FLINK-19731.
------------------------------
    Resolution: Duplicate

> 使用了批量执行SQL的api,job执行的时候,DataStream 不触发
> --------------------------------------
>
>                 Key: FLINK-19731
>                 URL: https://issues.apache.org/jira/browse/FLINK-19731
>             Project: Flink
>          Issue Type: Bug
>          Components: API / DataStream, Table SQL / API
>    Affects Versions: 1.11.2
>            Reporter: little-tomato
>            Priority: Minor
>
> 使用了批量执行SQL的api,job执行的时候,DataStream 不触发。
> 如果使用env.execute,则批量执行SQL的不触发。
> 能帮忙看看是不是哪边用得不对?非常感谢!
> final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
> EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
> StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, bsSettings);
> //读取kafka消息
> DataStream <String> kafkaDataStream...
> //进行业务过滤
> SingleOutputStreamOperator<MessageInfo> sourceStream = kafkaDataStream.map...
> //构建table
> DataStream <Row> dataStreamRow = sourceStream.map...
> Table table = tEnv.fromDataStream(dataStreamRow, fieldStr);
> StatementSet ss =tEnv.createStatementSet();
> ss.addInsertSql("insert into xx ...");
> ss.addInsertSql("insert into yy ...");
> TableResult result = ss.execute();



--
This message was sent by Atlassian Jira
(v8.3.4#803005)