flinksql source doris案例

亦凉 2023-10-06 22:39 39阅读 0赞

上一篇文章使用doris作为sink，  
为了简便，直接采用上一篇文章的worker表作为source  
[上一篇传送门][Link 1]

首先创建环境

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

然后创建flink doris table 映射表  
需要注意的是：  
**doris的数据类型和flink支持的类型，不完全一致**

tEnv.executeSql("create  table doris_worker(\n" +
                    " `startTime` String  ,\n" +
                    " `id` int ,\n" +
                    " `name` String,\n" +
                    "  `age` int ,\n" +
                    "  `city` String ,\n" +
                    "  `salary` int \n" +
                    ")with(\n" +
                    "'connector' = 'doris',\n" +
                    "'fenodes' = 'xxx.xxx.xxx.xxx.xxx:8030',\n" +
                    " 'table.identifier'='db.worker',\n" +
                    "   'sink.batch.size' = '2',\n" +
                    "    'username' = 'username',\n" +
                    "    'password' = 'password'\n" +
                    ")");

最后执行sql语句即可

tEnv.executeSql("select * from doris_worker where id >2").print();

这里需要注意的是  
tEnv.executeSql() 执行了sql之后，不需要再去执行 env.execute(“”);  
若添加会提示：

No operators defined in streaming topology. Cannot execute.

实践之后，才知道这么简单。

[Link 1]: https://blog.csdn.net/weixin_45399602/article/details/124614020

发表评论取消回复

表情：

评论列表（有 0 条评论，39人围观）

还没有评论，来说两句吧...

相关阅读

相关 Doris

Doris数据仓库总结目录一、Doris简介二、Doris的定位三、产品定位四、Doris的整体架构五、Doris的数据分布六、Doris的关

灰太狼/ 2024年03月25日 23:14/ 0 赞/ 107 阅读

相关 flinksql source doris案例

上一篇文章使用doris作为sink，为了简便，直接采用上一篇文章的worker表作为source [上一篇传送门][Link 1] 首先创建环境 Str

亦凉/ 2023年10月06日 22:39/ 0 赞/ 40 阅读

相关 Flink sink doris案例

添加 flink-doris-connector 和必要的 Flink Maven 依赖此处参考官网的配置 Flink 1.13.\ 及以前的版本 <de

Dear 丶/ 2023年10月06日 22:39/ 0 赞/ 42 阅读

相关 Flink案例——kafka、MySQL source

Flink案例——kafka、MySQL source 一、kafka source flink和kafka的连接是十分友好的，毕竟是做流式处理的吧。首先依赖

川长思鸟来/ 2023年10月06日 21:28/ 0 赞/ 27 阅读

相关 flinksql去掉换行符

可以使用REPLACE函数来去除flinksql中的换行符。示例语句如下: SELECT REPLACE(column_name, '\n', '') FROM ta

你的名字/ 2023年09月25日 15:16/ 0 赞/ 108 阅读

相关 FlinkSQL快速入门

实时数仓来势凶猛，FlinkSQL已成必杀技。五千字长文带大家快速入门FlinkSQL，占据技术栈优势，升职加薪加特！ [![图片][bf9eb562bc3227917604

蔚落/ 2022年10月29日 00:46/ 0 赞/ 40 阅读

相关 95-910-335-源码-FlinkSQL-Calcite-FlinkSQL解析

文章目录 1.美图 2.config.mpp 3.Parser.tdd 4.packge,class,imports 5.keywords

淡淡的烟草味﹌/ 2022年09月10日 11:11/ 0 赞/ 182 阅读

相关 95-910-140-源码-FlinkSQL-FlinkSQL简介

文章目录 1.世界 2.概述 3.FlinkSQL执行计划 4.Table与DataStream和DataSet API集成 5.\\Sca

我会带着你远行/ 2022年09月10日 11:09/ 0 赞/ 230 阅读

相关 Doris分区

1、创建分区表（1）默认分区当不使用 Partition 建表时，系统会自动生成一个和表名同名的，全值范围的 Partition。该 Partition 对用户不可

一时失言乱红尘/ 2022年09月09日 13:50/ 0 赞/ 242 阅读

相关 FlinkSQL MySQL CDC

[flink-cdc-connectors][] package com.sm.job import com.sm.function.udf.

雨点打透心脏的1/2处/ 2021年09月24日 21:10/ 0 赞/ 412 阅读