Flink Table API/SQL 多分支sink
背景
在某个场景中,需要从Kafka中获取数据,经过转换处理后,需要同时sink到多个输出源中(kafka、mysql、hologres)等。两次调用execute, 阿里云Flink vvr引擎报错:
public static void main(String[] args) {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);StreamStatementSet streamStatementSet = tEnv.createStatementSet();String s = LocalDateTimeUtils.getDateTime(System.currentTimeMillis());DataStream<String> dataStream = env.fromElements(s, LocalDateTimeUtils.getDateTime(System.currentTimeMillis()));tEnv.executeSql(KAFKA_TABLE_SQL);tEnv.executeSql(KAFKA_TABLE_SQL_1);Table table = tEnv.fromDataStream(dataStream);table.insertInto("kafka_sink").execute();table.insertInto("kafka_sink_1").execute();streamStatementSet.execute();}
Caused by: org.apache.flink.util.FlinkRuntimeException: Cannot have more than one execute() or executeAsync() call in a single environment.at org.apache.flink.client.program.StreamContextEnvironment.validateAllowedExecution(StreamContextEnvironment.java:199) ~[flink-dist-1.15-vvr-6.0.7-1-SNAPSHOT.jar:1.15-vvr-6.0.7-1-SNAPSHOT]at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:187) ~[flink-dist-1.15-vvr-6.0.7-1-SNAPSHOT.jar:1.15-vvr-6.0.7-1-SNAPSHOT]at org.apache.flink.table.planner.delegation.DefaultExecutor.executeAsync(DefaultExecutor.java:110) ~[?:?]at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:877) ~[flink-table-api-java-uber-1.15-vvr-6.0.7-1-SNAPSHOT.jar:1.15-vvr-6.0.7-1-SNAPSHOT]at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:756) ~[flink-table-api-java-uber-1.15-vvr-6.0.7-1-SNAPSHOT.jar:1.15-vvr-6.0.7-1-SNAPSHOT]at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:955) ~[flink-table-api-java-uber-1.15-vvr-6.0.7-1-SNAPSHOT.jar:1.15-vvr-6.0.7-1-SNAPSHOT]at org.apache.flink.table.api.internal.TablePipelineImpl.execute(TablePipelineImpl.java:57) ~[flink-table-api-java-uber-1.15-vvr-6.0.7-1-SNAPSHOT.jar:1.15-vvr-6.0.7-1-SNAPSHOT]
解决
使用 StreamStatementSet. 具体参考官网:
https://nightlies.apache.org/flink/flink-docs-release-1.15/zh/docs/dev/table/data_stream_api/#converting-between-datastream-and-table
改良后的代码:
public static void main(String[] args) {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);StreamStatementSet streamStatementSet = tEnv.createStatementSet();String s = LocalDateTimeUtils.getDateTime(System.currentTimeMillis());DataStream<String> dataStream = env.fromElements(s, LocalDateTimeUtils.getDateTime(System.currentTimeMillis()));tEnv.executeSql(KAFKA_TABLE_SQL);tEnv.executeSql(KAFKA_TABLE_SQL_1);Table table = tEnv.fromDataStream(dataStream);streamStatementSet.addInsert("kafka_sink", table);streamStatementSet.addInsert("kafka_sink_1", table);streamStatementSet.execute();}
相关文章:
Flink Table API/SQL 多分支sink
背景 在某个场景中,需要从Kafka中获取数据,经过转换处理后,需要同时sink到多个输出源中(kafka、mysql、hologres)等。两次调用execute, 阿里云Flink vvr引擎报错: public static void main(String[] args) {final StreamExecuti…...
Vue3 中 导航守卫 的使用
在Vue 3中,导航守卫(Navigation Guards)用于在路由切换前后执行一些操作,例如验证用户权限、取消路由导航等。Vue 3中的导航守卫与Vue 2中的导航守卫略有不同。下面是Vue 3中导航守卫的使用方式: 全局前置守卫…...
云原生概论
云原生是一种新兴的技术趋势,它旨在将应用程序设计和部署方式从传统的基础设施转向云端。云原生应用程序是一种针对云环境进行优化的应用程序,能够充分利用云端提供的弹性和可扩展性。本文将探讨云原生的概念、优势、应用场景以及未来发展方向。 一、云…...
hive-sql
hive-常用SQL汇总 查看数据库 -- 查看所有的数据库 show databases; 使用默认的库 -- 下面的语句可以查看默认的库 use default ;查看某个库下的表 -- 查看所有的表 show tables ; -- 查看包含 stu的表 ,这种是通配的方法来查看 show tables like *stu*; 查…...
Rspack 创建 vue2/3 项目接入 antdv(rspack.config.js 配置 less 主题)
一、简介 Rspack CLI 官方文档。 rspack.config.js 官方文档。 二、创建 vue 项目 创建项目(文档中还提供了 Rspack 内置 monorepo 框架 Nx 的创建方式,根据需求进行选择) # npm 方式 $ npm create rspacklatest# yarn 方式 $ yarn create…...
基于centos7完成docker服务的一些基础操作
目录 要求完成 具体操作 1.安装docker服务,配置镜像加速器 2.下载系统镜像(Ubuntu、 centos) 3.基于下载的镜像创建两个容器 (容器名一个为自己名字全拼,一个为首名字字母) 4.容器的启动、 停止及重启…...
Microsoft Visual Studio + Qt插件编程出现错误error MSB4184问题
文章目录 报错解决 报错 C:\Users\Administrator\AppData\Local\QtMsBuild\qt_globals.targets(786,7): error MSB4184: 无法计算表达式“[System.IO.File]::ReadAllText(C:\Users\Administrator\AppData\Local\QtMsBuild\qt.natvis.xml)”。 未能找到文件“C:\Users\Administ…...
QT Quick之quick与C++混合编程
Qt quick能够生成非常绚丽界面,但有其局限性的,对于一些业务逻辑和复杂算法,比如低阶的网络编程如 QTcpSocket ,多线程,又如 XML 文档处理类库 QXmlStreamReader / QXmlStreamWriter 等等,在 QML 中要么不可…...
Ros noetic Move_base 相关状态位置的获取 实战使用教程
前言: 有一段时间没有更新,这篇文章是为了后续MPC路径跟踪算法开设的帖子用于更新我自己的思路,由于MPC算法,要镶嵌到整个导航任务中去,就绕不开这个move_base包中相关的参数设置和其中相关状态位置的获取和解读等等。 因为最近遇到小车在其他的环境中有些时候,不需要自己…...
【SpringBoot】SpringBoot项目与Vue对接接口的步骤
下面是SpringBoot项目与Vue对接接口的步骤: 创建SpringBoot项目,在项目中添加依赖,如Spring MVC、MyBatis等框架。 在SpringBoot项目中编写接口方法,使用注解标识请求方式,如GetMapping、PostMapping等,并…...
Glog安装与使用
安装 脚本 #!/bin/bash git clone https://github.com/google/glog.git cd glog git checkout v0.4.0 mkdir build && cd build cmake .. make -j4 echo "your password" | sudo -S make install使用 main.cc #include <glog/logging.h>int main(i…...
windows开发环境搭建
下载msys2,官网下载即可: MSYS2 安装其他的编译工具(貌似不需要把中间的命令全部执行): MSYS2使用教程——win10系统64位安装msys2最新版(msys2-x86_xxxx.exe)_msys64_Dreamhai的博客-CSDN博…...
8月17日上课内容 第三章 LVS+Keepalived群集
本章结构 Keepalived概述 keepalived 概述 1.服务功能 故障自动切换 健康检查 节点服务器高可用 HA keepalived工作原理 Keepalived 是一个基于VRRP协议来实现的LVS服务高可用方案,可以解决静态路由出现的单点故障问题 在一个LVS服务集群中通常有主服务器 (MAST…...
Threejs学习05——球缓冲几何体背景贴图和环境贴图
实现随机多个三角形随机位置随机颜色展示效果 这是一个非常简单基础的threejs的学习应用!本节主要学习的是球面缓冲几何体的贴图部分,这里有环境贴图以及背景贴图,这样可以有一种身临其境的效果!这里环境贴图用的是一个.hdr的文件…...
LVS+Keepalived群集实验
目录 Keepalived 是什么 Keepalived 功能 Keepalived 模块 工作原理 脑裂现象及解决方案 脑裂 形成脑裂的原因 解决脑裂的几种方法: 为了减少或避免HA集群中出现脑裂现象,我们可以采取以下措施: Keepalived服务主要功能࿰…...
软考高级之系统架构师之系统开发基础
架构 场景 场景(scenarios)在进行体系结构评估时,一般首先要精确地得出具体的质量目标,并以之作为判定该体系结构优劣的标准。为得出这些目标而采用的机制做场景。场景是从风险承担者的角度对与系统的交互的简短描述。在体系结构…...
Web 3.0 安全风险,您需要了解这些内容
随着技术的不断发展,Web 3.0 正在逐渐成为现实,为我们带来了许多新的机遇和挑战。然而,与任何新技术一样,Web 3.0 也伴随着一系列安全风险,这些风险需要被认真对待。在这篇文章中,我们将探讨一些与Web 3.0 …...
万宾科技22款产品入选《城市生命线安全工程监测技术产品名录》
2023年8月17日-18日,由北京市地下管线协会主办的2023首届城市生命线安全与发展大会在北京召开,本次大会汇聚中央及地方政府主管领导、院士专家、行业领袖、龙头代表、产业精英等。 大会聚焦安全监管智慧平台和燃气爆炸、城市内涝、地下管线交互风险、第三…...
MFC 隐藏窗口
亲测能用 改变主窗体的创建方式 将 C***App::InitInstance() 函数中的代码 CMFCApplication1Dlg dlg; m_pMainWnd &dlg; INT_PTR nResponse dlg.DoModal(); 替换为 CMFCApplication1Dlg dlg; m_pMainWnd &dlg; //INT_PTR nResponse dlg.DoModal(); INT_PTR nRe…...
Java数据库连接池原理及spring boot使用数据库连接池(HikariCP、Druid)
和线程池类似,数据库连接池的作用是建立一些和数据库的连接供需要连接数据库的业务使用,避免了每次和数据库建立、销毁连接的性能消耗,通过设置连接池参数可以防止建立连接过多导致服务宕机等,以下介绍Java中主要使用的几种数据库…...
在软件开发中正确使用MySQL日期时间类型的深度解析
在日常软件开发场景中,时间信息的存储是底层且核心的需求。从金融交易的精确记账时间、用户操作的行为日志,到供应链系统的物流节点时间戳,时间数据的准确性直接决定业务逻辑的可靠性。MySQL作为主流关系型数据库,其日期时间类型的…...
树莓派超全系列教程文档--(62)使用rpicam-app通过网络流式传输视频
使用rpicam-app通过网络流式传输视频 使用 rpicam-app 通过网络流式传输视频UDPTCPRTSPlibavGStreamerRTPlibcamerasrc GStreamer 元素 文章来源: http://raspberry.dns8844.cn/documentation 原文网址 使用 rpicam-app 通过网络流式传输视频 本节介绍来自 rpica…...
Appium+python自动化(十六)- ADB命令
简介 Android 调试桥(adb)是多种用途的工具,该工具可以帮助你你管理设备或模拟器 的状态。 adb ( Android Debug Bridge)是一个通用命令行工具,其允许您与模拟器实例或连接的 Android 设备进行通信。它可为各种设备操作提供便利,如安装和调试…...
使用分级同态加密防御梯度泄漏
抽象 联邦学习 (FL) 支持跨分布式客户端进行协作模型训练,而无需共享原始数据,这使其成为在互联和自动驾驶汽车 (CAV) 等领域保护隐私的机器学习的一种很有前途的方法。然而,最近的研究表明&…...
P3 QT项目----记事本(3.8)
3.8 记事本项目总结 项目源码 1.main.cpp #include "widget.h" #include <QApplication> int main(int argc, char *argv[]) {QApplication a(argc, argv);Widget w;w.show();return a.exec(); } 2.widget.cpp #include "widget.h" #include &q…...
跨链模式:多链互操作架构与性能扩展方案
跨链模式:多链互操作架构与性能扩展方案 ——构建下一代区块链互联网的技术基石 一、跨链架构的核心范式演进 1. 分层协议栈:模块化解耦设计 现代跨链系统采用分层协议栈实现灵活扩展(H2Cross架构): 适配层…...
【生成模型】视频生成论文调研
工作清单 上游应用方向:控制、速度、时长、高动态、多主体驱动 类型工作基础模型WAN / WAN-VACE / HunyuanVideo控制条件轨迹控制ATI~镜头控制ReCamMaster~多主体驱动Phantom~音频驱动Let Them Talk: Audio-Driven Multi-Person Conversational Video Generation速…...
【VLNs篇】07:NavRL—在动态环境中学习安全飞行
项目内容论文标题NavRL: 在动态环境中学习安全飞行 (NavRL: Learning Safe Flight in Dynamic Environments)核心问题解决无人机在包含静态和动态障碍物的复杂环境中进行安全、高效自主导航的挑战,克服传统方法和现有强化学习方法的局限性。核心算法基于近端策略优化…...
CSS | transition 和 transform的用处和区别
省流总结: transform用于变换/变形,transition是动画控制器 transform 用来对元素进行变形,常见的操作如下,它是立即生效的样式变形属性。 旋转 rotate(角度deg)、平移 translateX(像素px)、缩放 scale(倍数)、倾斜 skewX(角度…...
iview框架主题色的应用
1.下载 less要使用3.0.0以下的版本 npm install less2.7.3 npm install less-loader4.0.52./src/config/theme.js文件 module.exports {yellow: {theme-color: #FDCE04},blue: {theme-color: #547CE7} }在sass中使用theme配置的颜色主题,无需引入,直接可…...
