Flink读写Doris操作介绍

乌市泽哥  金牌会员 | 2024-6-14 22:36:48 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 844|帖子 844|积分 2532

Flink读写Doris操作介绍

  ​ Flink Doris Connector 可以支持通过 Flink 操作（读取、插入、修改、删除） Doris 中存储的数据。可以将 Doris 表映射为 DataStream 或者 Table。


  • Flink操作Doris修改和删除只支持在 Unique Key 模子上
1. 准备开发情况



  • pom.xml参加依赖
  1. <dependency>
  2.     <groupId>org.apache.doris</groupId>
  3.     <artifactId>flink-doris-connector-1.13_2.12</artifactId>
  4.     <version>1.0.3</version>
  5. </dependency>
复制代码


  • 创建测试库测试表
  1. -- 切测试库
  2. use test_db;
  4. -- 创建测试表flinktest
  5. CREATE TABLE flinktest
  6. (
  7.     siteid INT DEFAULT '10',
  8.     citycode SMALLINT,
  9.     username VARCHAR(32) DEFAULT '',
  10.     pv BIGINT SUM DEFAULT '0'
  11. )
  12. AGGREGATE KEY(siteid, citycode, username)
  13. DISTRIBUTED BY HASH(siteid) BUCKETS 10
  14. PROPERTIES("replication_num" = "1");
  16. -- 插入样例数据
  17. insert into flinktest values
  18. (1,1,'jim',2),
  19. (2,1,'grace',2),
  20. (3,2,'tom',2),
  21. (4,3,'bush',3),
  22. (5,3,'helen',3);
  24. -- 查看表数据情况
  25. select * from flinktest;
  26. +--------+----------+----------+------+
  27. | siteid | citycode | username | pv   |
  28. +--------+----------+----------+------+
  29. |      1 |        1 | jim      |    2 |
  30. |      5 |        3 | helen    |    3 |
  31. |      4 |        3 | bush     |    3 |
  32. |      3 |        2 | tom      |    2 |
  33. |      2 |        1 | grace    |    2 |
  34. +--------+----------+----------+------+
复制代码


  • Doris 和 Flink 列范例映射关系
Doris TypeFlink TypeNULL_TYPENULLBOOLEANBOOLEANTINYINTTINYINTSMALLINTSMALLINTINTINTBIGINTBIGINTFLOATFLOATDOUBLEDOUBLEDATEDATEDATETIMETIMESTAMPDECIMALDECIMALCHARSTRINGLARGEINTSTRINGVARCHARSTRINGDECIMALV2DECIMALTIMEDOUBLEHLLUnsupported datatype 2. Flink-DataStream读Doris

代码示例：
  1. package com.zenitera.bigdata.doris;
  3. import org.apache.doris.flink.cfg.DorisStreamOptions;
  4. import org.apache.doris.flink.datastream.DorisSourceFunction;
  5. import org.apache.doris.flink.deserialization.SimpleListDeserializationSchema;
  6. import org.apache.flink.configuration.Configuration;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import java.util.Properties;
  11. public class Flink_stream_read_doris {
  12.     public static void main(String[] args) {
  13.         Configuration conf = new Configuration();
  14.         conf.setInteger("rest.port", 2000);
  15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
  16.         env.setParallelism(1);
  19.         Properties props = new Properties();
  20.         props.setProperty("fenodes", "hdt-dmcp-ops01:8130");
  21.         props.setProperty("username", "root");
  22.         props.setProperty("password", "123456");
  23.         props.setProperty("table.identifier", "test_db.flinktest");
  25.         env
  26.                 .addSource(new DorisSourceFunction(new DorisStreamOptions(props), new SimpleListDeserializationSchema()))
  27.                 .print();
  30.         try {
  31.             env.execute();
  32.         } catch (Exception e) {
  33.             e.printStackTrace();
  34.         }
  36.     }
  37. }
  39. /*
  40.   代码控制台输出：
  41. [4, 3, bush, 3]
  42. [2, 1, grace, 2]
  43. [1, 1, jim, 2]
  44. [5, 3, helen, 3]
  45. [3, 2, tom, 2]
  46. */
复制代码
3. Flink写Doris

Flink 读写 Doris 数据重要有两种方式


  • DataStream
  • SQL
3.1 Flink-DataStream以 JSON 数据 写到Doris

代码示例：
  1. package com.zenitera.bigdata.doris;
  3. import org.apache.doris.flink.cfg.DorisExecutionOptions;
  4. import org.apache.doris.flink.cfg.DorisOptions;
  5. import org.apache.doris.flink.cfg.DorisSink;
  6. import org.apache.flink.configuration.Configuration;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import java.util.Properties;
  11. /**
  12. * 使用 Flink 将 JSON 数据 写到Doris数据库
  13. */
  14. public class Flink_stream_write_doris_json {
  15.     public static void main(String[] args) {
  16.         Configuration conf = new Configuration();
  17.         conf.setInteger("rest.port", 2000);
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
  19.         env.setParallelism(1);
  21.         Properties pro = new Properties();
  22.         pro.setProperty("format", "json");
  23.         pro.setProperty("strip_outer_array", "true");
  24.         env
  25.                 .fromElements("{"siteid":"10", "citycode": "1001","username": "ww","pv":"100"}")
  26.                 .addSink(DorisSink.sink(
  27.                         new DorisExecutionOptions.Builder()
  28.                                 .setBatchIntervalMs(2000L)
  29.                                 .setEnableDelete(false)
  30.                                 .setMaxRetries(3)
  31.                                 .setStreamLoadProp(pro)
  32.                                 .build(),
  33.                         new DorisOptions.Builder()
  34.                                 .setFenodes("hdt-dmcp-ops01:8130")
  35.                                 .setUsername("root")
  36.                                 .setPassword("123456")
  37.                                 .setTableIdentifier("test_db.flinktest")
  38.                                 .build())
  39.                 );
  41.         try {
  42.             env.execute();
  43.         } catch (Exception e) {
  44.             e.printStackTrace();
  45.         }
  47.     }
  48. }
  50. /*
  51.     代码执行前： 5 rows
  52. select * from flinktest;
  53. +--------+----------+----------+------+
  54. | siteid | citycode | username | pv   |
  55. +--------+----------+----------+------+
  56. |      1 |        1 | jim      |    2 |
  57. |      5 |        3 | helen    |    3 |
  58. |      4 |        3 | bush     |    3 |
  59. |      3 |        2 | tom      |    2 |
  60. |      2 |        1 | grace    |    2 |
  61. +--------+----------+----------+------+
  63.     代码执行后： 6 rows
  64. select * from flinktest;
  65. +--------+----------+----------+------+
  66. | siteid | citycode | username | pv   |
  67. +--------+----------+----------+------+
  68. |      2 |        1 | grace    |    2 |
  69. |      3 |        2 | tom      |    2 |
  70. |      5 |        3 | helen    |    3 |
  71. |      1 |        1 | jim      |    2 |
  72. |     10 |     1001 | ww       |  100 |
  73. |      4 |        3 | bush     |    3 |
  74. +--------+----------+----------+------+
  75. */
复制代码
3.2 Flink-DataStream以 RowData 数据 写Doris

代码示例：
  1. package com.zenitera.bigdata.doris;
  3. import com.alibaba.fastjson.JSON;
  4. import com.alibaba.fastjson.JSONObject;
  5. import org.apache.doris.flink.cfg.DorisExecutionOptions;
  6. import org.apache.doris.flink.cfg.DorisOptions;
  7. import org.apache.doris.flink.cfg.DorisSink;
  8. import org.apache.flink.configuration.Configuration;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  10. import org.apache.flink.table.data.GenericRowData;
  11. import org.apache.flink.table.data.StringData;
  12. import org.apache.flink.table.types.logical.*;
  15. public class Flink_stream_write_doris_rowdata {
  16.     public static void main(String[] args) {
  17.         Configuration conf = new Configuration();
  18.         conf.setInteger("rest.port", 2000);
  19.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
  20.         env.setParallelism(1);
  22.         LogicalType[] types = {new IntType(), new SmallIntType(), new VarCharType(), new BigIntType()};
  23.         String[] fields = {"siteid", "citycode", "username", "pv"};
  25.         env
  26.                 .fromElements("{"siteid":"100", "citycode": "1002","username": "wang","pv":"100"}")
  28.                 .map(json -> {
  29.                     JSONObject obj = JSON.parseObject(json);
  30.                     GenericRowData rowData = new GenericRowData(4);
  31.                     rowData.setField(0, obj.getIntValue("siteid"));
  32.                     rowData.setField(1, obj.getShortValue("citycode"));
  33.                     rowData.setField(2, StringData.fromString(obj.getString("username")));
  34.                     rowData.setField(3, obj.getLongValue("pv"));
  35.                     return rowData;
  37.                 })
  40.                 .addSink(DorisSink.sink(
  41.                         fields,
  42.                         types,
  43.                         new DorisExecutionOptions.Builder()
  44.                                 .setBatchIntervalMs(2000L)
  45.                                 .setEnableDelete(false)
  46.                                 .setMaxRetries(3)
  47.                                 .build(),
  48.                         new DorisOptions.Builder()
  49.                                 .setFenodes("hdt-dmcp-ops01:8130")
  50.                                 .setUsername("root")
  51.                                 .setPassword("123456")
  52.                                 .setTableIdentifier("test_db.flinktest")
  53.                                 .build())
  54.                 );
  56.         try {
  57.             env.execute();
  58.         } catch (Exception e) {
  59.             e.printStackTrace();
  60.         }
  62.     }
  63. }
  65. /*
  66.     代码执行前： 6 rows
  67. select * from flinktest;
  68. +--------+----------+----------+------+
  69. | siteid | citycode | username | pv   |
  70. +--------+----------+----------+------+
  71. |      2 |        1 | grace    |    2 |
  72. |      3 |        2 | tom      |    2 |
  73. |      5 |        3 | helen    |    3 |
  74. |      1 |        1 | jim      |    2 |
  75. |     10 |     1001 | ww       |  100 |
  76. |      4 |        3 | bush     |    3 |
  77. +--------+----------+----------+------+
  79.     代码执行后： 7 rows
  80. select * from flinktest;
  81. +--------+----------+----------+------+
  82. | siteid | citycode | username | pv   |
  83. +--------+----------+----------+------+
  84. |      1 |        1 | jim      |    2 |
  85. |      2 |        1 | grace    |    2 |
  86. |      3 |        2 | tom      |    2 |
  87. |      5 |        3 | helen    |    3 |
  88. |     10 |     1001 | ww       |  100 |
  89. |    100 |     1002 | wang     |  100 |
  90. |      4 |        3 | bush     |    3 |
  91. +--------+----------+----------+------+
  92. */
复制代码
3.3 Flink-SQL 方式写Doris

Doris测试表：
  1. use test_db;
  3. truncate table flinktest;
  5. insert into flinktest values
  6. (1,1,'aaa',1),
  7. (2,2,'bbb',2),
  8. (3,3,'ccc',3);
  10. select * from flinktest;
  11. +--------+----------+----------+------+
  12. | siteid | citycode | username | pv   |
  13. +--------+----------+----------+------+
  14. |      2 |        2 | bbb      |    2 |
  15. |      1 |        1 | aaa      |    1 |
  16. |      3 |        3 | ccc      |    3 |
  17. +--------+----------+----------+------+
  18. 3 rows in set (0.01 sec)
复制代码
Flink-SQL代码示例：
  1. package com.zenitera.bigdata.doris;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  6. import org.apache.flink.configuration.Configuration;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  10. public class Flink_SQL_doris {
  11.     public static void main(String[] args) {
  12.         Configuration conf = new Configuration();
  13.         conf.setInteger("rest.port", 2000);
  14.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
  15.         env.setParallelism(1);
  17.         StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
  19.         tEnv.executeSql("create table flink_0518(" +
  20.                 " siteid int, " +
  21.                 " citycode int, " +
  22.                 " username string, " +
  23.                 " pv bigint " +
  24.                 ")with(" +
  25.                 "  'connector' = 'doris', " +
  26.                 "  'fenodes' = 'hdt-dmcp-ops01:8130', " +
  27.                 "  'table.identifier' = 'test_db.flinktest', " +
  28.                 "  'username' = 'root', " +
  29.                 "  'password' = '123456' " +
  30.                 ")");
  32.         tEnv.executeSql("insert into flink_0518(siteid, citycode, username, pv) values(4, 4, 'wangting', 4) ");
  34.     }
  36.     @Data
  37.     @NoArgsConstructor
  38.     @AllArgsConstructor
  39.     public static class Flink_0518 {
  40.         private Integer siteid;
  41.         private Integer citycode;
  42.         private String username;
  43.         private Long pv;
  44.     }
  45. }
复制代码
实行代码，实行完成后查看Doris对应表数据进行验证：
  1. select * from flinktest;
  2. +--------+----------+----------+------+
  3. | siteid | citycode | username | pv   |
  4. +--------+----------+----------+------+
  5. |      3 |        3 | ccc      |    3 |
  6. |      2 |        2 | bbb      |    2 |
  7. |      1 |        1 | aaa      |    1 |
  8. |      4 |        4 | wangting |    4 |
  9. +--------+----------+----------+------+
  10. 4 rows in set (0.01 sec)
复制代码
3.4 Flink-SQL 方式读Doris

  1. package com.zenitera.bigdata.doris;
  3. import org.apache.flink.configuration.Configuration;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  7. public class Flink_SQL_doris_read {
  8.     public static void main(String[] args) {
  9.         Configuration conf = new Configuration();
  10.         conf.setInteger("rest.port", 2000);
  11.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
  12.         env.setParallelism(1);
  14.         StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
  16.         tEnv.executeSql("create table flink_0520(" +
  17.                 " siteid int, " +
  18.                 " citycode SMALLINT, " +
  19.                 " username string, " +
  20.                 " pv bigint " +
  21.                 ")with(" +
  22.                 "  'connector' = 'doris', " +
  23.                 "  'fenodes' = 'hdt-dmcp-ops01:8130', " +
  24.                 "  'table.identifier' = 'test_db.flinktest', " +
  25.                 "  'username' = 'root', " +
  26.                 "  'password' = '123456' " +
  27.                 ")");
  29.         tEnv.sqlQuery("select * from flink_0520").execute().print();
  31.     }
  32. }
  34. /*
  35.    控制台输出信息：
  36. +----+-------------+----------+---------------+---------+
  37. | op |      siteid | citycode |      username |      pv |
  38. +----+-------------+----------+---------------+---------+
  39. | +I |           1 |        1 |           aaa |       1 |
  40. | +I |           3 |        3 |           ccc |       3 |
  41. | +I |           2 |        2 |           bbb |       2 |
  42. | +I |           4 |        4 |      wangting |       4 |
  43. +----+-------------+----------+---------------+---------+
  44. 4 rows in set
  45. */
复制代码
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

乌市泽哥

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表