当前位置: 首页 > news >正文

武汉方林装饰公司电话号码单词优化和整站优化

武汉方林装饰公司电话号码,单词优化和整站优化,新开传奇网站180合击,帮齐家网做的网站FlinkSQL_1.12_用DDL实现Kafka到MySQL的数据传输_实现按照条件进行过滤写入MySQL_flink从kafka拉取数据并过滤数据写入mysql_旧城里的阳光的博客-CSDN博客 参考这篇文章,写了kafka到mysql的代码例子,因为自己改了表结构,运行下面代码&#x…

FlinkSQL_1.12_用DDL实现Kafka到MySQL的数据传输_实现按照条件进行过滤写入MySQL_flink从kafka拉取数据并过滤数据写入mysql_旧城里的阳光的博客-CSDN博客

参考这篇文章,写了kafka到mysql的代码例子,因为自己改了表结构,运行下面代码:

package org.test.flink;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;//TODO 用DDL实现Kafka到MySQL的数据传输
public class FlinkSQL15_SQL_DDL_Kafka_MySQL {public static void main(String[] args) throws Exception {//1.获取执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);//2.使用DDL的方式加载数据--注册SourceTabletableEnv.executeSql("create table source_sensor(account_id  BIGINT)" +"with (" +"'connector.type' = 'kafka'," +"'connector.version' = 'universal'," +"'connector.topic' = 'testtopic'," +"'connector.properties.bootstrap.servers' = '11.0.24.216:9092'," +"'connector.properties.group.id' = 'bigdata1109'," +"'format.type' = 'json'"+ ")");Table table = tableEnv.sqlQuery("select * from source_sensor");//3.注册SinkTable:MysqltableEnv.executeSql("CREATE TABLE spend_report (\n" +"    account_id BIGINT,\n" +"    PRIMARY KEY (account_id) NOT ENFORCED)" +"with (" +"'connector' = 'jdbc'," +"'url' = 'jdbc:mysql://11.0.24.216:4306/test?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=utf8&useSSL=false',"+"'table-name' = 'spend_report',"+"'username' = 'root',"+"'password' = '123456'"+ ")");//4.执行查询kafka数据
//        Table source_sensor = tableEnv.from("source_sensor");
//        //5.将数据写入Mysql
//        source_sensor.executeInsert("sink_sensor");
//table.executeInsert("sink_sensor");//6.执行任务env.execute();}
}

发现报错如下:

Exception in thread "main" org.apache.flink.table.api.TableException: Sink `default_catalog`.`default_database`.`sink_sensor` does not existsat org.apache.flink.table.planner.delegation.PlannerBase.translateToRel(PlannerBase.scala:247)at org.apache.flink.table.planner.delegation.PlannerBase.$anonfun$translate$1(PlannerBase.scala:159)at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286)at scala.collection.Iterator.foreach(Iterator.scala:943)at scala.collection.Iterator.foreach$(Iterator.scala:943)at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)at scala.collection.IterableLike.foreach(IterableLike.scala:74)at scala.collection.IterableLike.foreach$(IterableLike.scala:73)at scala.collection.AbstractIterable.foreach(Iterable.scala:56)at scala.collection.TraversableLike.map(TraversableLike.scala:286)at scala.collection.TraversableLike.map$(TraversableLike.scala:279)at scala.collection.AbstractTraversable.map(Traversable.scala:108)at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:159)at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1329)at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:676)at org.apache.flink.table.api.internal.TableImpl.executeInsert(TableImpl.java:572)at org.apache.flink.table.api.internal.TableImpl.executeInsert(TableImpl.java:554)at org.test.flink.FlinkSQL15_SQL_DDL_Kafka_MySQL.main(FlinkSQL15_SQL_DDL_Kafka_MySQL.java:50)

点击table.executeInsert看了下源码:

    /*** Writes the {@link Table} to a {@link TableSink} that was registered under the specified path,* and then execute the insert operation.** <p>See the documentation of {@link TableEnvironment#useDatabase(String)} or {@link* TableEnvironment#useCatalog(String)} for the rules on the path resolution.** <p>A batch {@link Table} can only be written to a {@code* org.apache.flink.table.sinks.BatchTableSink}, a streaming {@link Table} requires a {@code* org.apache.flink.table.sinks.AppendStreamTableSink}, a {@code* org.apache.flink.table.sinks.RetractStreamTableSink}, or an {@code* org.apache.flink.table.sinks.UpsertStreamTableSink}.** <p>Example:** <pre>{@code* Table table = tableEnv.fromQuery("select * from MyTable");* TableResult tableResult = table.executeInsert("MySink");* tableResult...* }</pre>** @param tablePath The path of the registered TableSink to which the Table is written.* @return The insert operation execution result.*/TableResult executeInsert(String tablePath);

 发现executeInsert方法的参数tablePath需要传入表名,这里的表名应该和

tableEnv.executeSql("create table source_sensor(account_id  BIGINT)"

的表名source_sensor一致。

将:

table.executeInsert("sink_sensor");

改成:

table.executeInsert("source_sensor");

后执行成功。

flink1.2的demo完整代码:flink-java-1.12.7: flink1.12.7的java demo,包括flink wordcount示例,如何连接kafka

 

http://www.15wanjia.com/news/37187.html

相关文章:

  • 上海的外贸网站建设公司价格国际免费b站
  • 网站demo要几个人做一个平台怎么推广
  • 东营列表网seo关键词外包公司
  • 网站改名 seo西安seo关键词查询
  • 影视app制作seo搜索推广
  • 自己做网站怎么连接外网本周新闻热点10条
  • 做网站报价惠州百度seo哪里强
  • wordpress 问答插件2广州seo外包公司
  • 驻马店做网站优化湖南网站设计
  • 做基网站成都最好的网站推广优化公司
  • 简单描述网站建设流程百度竞价排名
  • 怎样做网站用的大于9kb百度快照
  • 网站建设电话销售本周时事新闻概要10条
  • 做网站合同看seo
  • 长沙专业做网站公司有哪些seo实训报告
  • 1元做网站方案seminar怎么读
  • 做网站卖东西流程b站在线观看人数在哪
  • wordpress页面图片如何排版seo关键词优化软件怎么样
  • 江苏同邦建设有限公司网站宝鸡seo外包公司
  • 龙华网站建设 信科网络青岛网络推广公司哪家好
  • 上海专业高端网站建设国际时事新闻2022最新
  • 设计制作软件哈尔滨优化调整人员流动管理
  • 建设网站需要的配置深圳 网站制作
  • 网站开发 大学专业东莞寮步最新通知
  • 现在学做网站赚钱吗小学四年级摘抄新闻
  • 网站开发进度计划友情链接检索数据分析
  • admin登录网站app推广渠道在哪接的单子
  • 商城开发网站建设百度快照怎么弄
  • wordpress 文章字体颜色seo管理系统创作
  • 电商网站后台建设问题企业网站有哪些类型