Flink 1.11 table案例

╰半夏微凉° 2023-02-24 11:21 3阅读 0赞

1，先来的简单点的，折腾了半天 昨天是maven下载不了flink-clients.jar ，下载之后手动导入，然后最简单的代码都运行不了

今天没办法，还是报错，缺包（org.apache.flink.optimizer.costs.CostEstimator），没法子，缺的包就是flink-clients.jar里面的，

再次尝试加入依赖，今天是成功了，代码就执行成功了。

官网中文文档地址：[https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/common.html\#%E9%80%9A%E8%BF%87-dataset-%E6%88%96-datastream-%E5%88%9B%E5%BB%BA%E8%A7%86%E5%9B%BE][https_ci.apache.org_projects_flink_flink-docs-release-1.11_zh_dev_table_common.html_E9_80_9A_E8_BF_87-dataset-_E6_88_96-datastream-_E5_88_9B_E5_BB_BA_E8_A7_86_E5_9B_BE]

>     <dependency>
>         <groupId>org.apache.flink</groupId>
>         <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
>         <version>${flink.version}</version>
>     </dependency>
>     <dependency>
>         <groupId>org.apache.flink</groupId>
>         <artifactId>flink-table-api-scala-bridge_${scala.binary.version}</artifactId>
>         <version>${flink.version}</version>
>     </dependency>
>     <dependency>
>         <groupId>org.apache.flink</groupId>
>         <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
>         <version>${flink.version}</version>
>     </dependency>
>     <dependency>
>         <groupId>org.apache.flink</groupId>
>         <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
>         <version>${flink.version}</version>
>     </dependency>
>     
>     <dependency>
>         <groupId>org.apache.flink</groupId>
>         <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
>         <version>${flink.version}</version>
>     </dependency>
>     
>     <dependency>
>         <groupId>org.apache.flink</groupId>
>         <artifactId>flink-clients_${scala.binary.version}</artifactId>
>         <version>${flink.version}</version>
>     </dependency>

>     public class SqlTest03 {
>         public static void main(String[] args) throws Exception {
>             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
>             EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
>             StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env,bsSettings);
>     
>     // ingest a DataStream from an external source
>             DataStream<Tuple3<Long, String, Integer>> ds = env.addSource(new SourceFunction<Tuple3<Long, String, Integer>>() {
>                 @Override
>                 public void run(SourceContext<Tuple3<Long, String, Integer>> out) throws Exception {
>                     while (true){
>                         out.collect(new Tuple3<>(1L,"a",11));
>                         Thread.sleep(1000L);
>                     }
>                 }
>                 @Override
>                 public void cancel() {
>     
>                 }
>             });
>             Table table = tableEnv.fromDataStream(ds, $("user"), $("product"), $("amount"));
>             DataStream<Tuple2<Boolean, Row>> dsRow = tableEnv.toRetractStream(table, Row.class);
>             dsRow.print();
>             env.execute();
>     
>         }
>     }

顺便提一句关于 最后面的执行：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

[https_ci.apache.org_projects_flink_flink-docs-release-1.11_zh_dev_table_common.html_E9_80_9A_E8_BF_87-dataset-_E6_88_96-datastream-_E5_88_9B_E5_BB_BA_E8_A7_86_E5_9B_BE]: https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/common.html#%E9%80%9A%E8%BF%87-dataset-%E6%88%96-datastream-%E5%88%9B%E5%BB%BA%E8%A7%86%E5%9B%BE
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20230209/70e9f6d0ae954ced89dc32b64a16658d.png