大数据-148 Apache Kudu 从 Flink 下沉数据到 Kudu

络腮胡菲菲  金牌会员 | 2024-10-7 20:30:01 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 835|帖子 835|积分 2505

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：



  • Hadoop（已更完）
  • HDFS（已更完）
  • MapReduce（已更完）
  • Hive（已更完）
  • Flume（已更完）
  • Sqoop（已更完）
  • Zookeeper（已更完）
  • HBase（已更完）
  • Redis （已更完）
  • Kafka（已更完）
  • Spark（已更完）
  • Flink（已更完）
  • ClickHouse（已更完）
  • Kudu（正在更新…）
章节内容

上节我们完成了如下的内容：


  • Kudu Java API
  • 增删改查 编写案例测试

实现思路

将数据从 Flink 下沉到 Kudu 的基本思路如下：


  • 环境准备：确保 Flink 和 Kudu 环境正常运行，并设置好相干依赖。
  • 创建 Kudu 表：在 Kudu 中界说要存储的数据表，包括主键和列类型。
  • 数据流计划：使用 Flink 的 DataStream API 读取输入数据流，进行必要的数据处置惩罚和转换。
  • 写入 Kudu：通过 Kudu 的毗连器将处置惩罚后的数据写入 Kudu 表。需要设置 Kudu 客户端和表的相干信息。
  • 执行作业：启动 Flink 作业，实时将数据流中的数据写入 Kudu，便于后续查询和分析。
添加依赖

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  7.     <groupId>org.example</groupId>
  8.     <artifactId>flink-test</artifactId>
  9.     <version>1.0-SNAPSHOT</version>
  11.     <properties>
  12.         <maven.compiler.source>11</maven.compiler.source>
  13.         <maven.compiler.target>11</maven.compiler.target>
  14.         <flink.version>1.11.1</flink.version>
  15.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  16.     </properties>
  18.     <dependencies>
  19.         <dependency>
  20.             <groupId>org.apache.flink</groupId>
  21.             <artifactId>flink-java</artifactId>
  22.             <version>${flink.version}</version>
  23.         </dependency>
  24.         <dependency>
  25.             <groupId>org.apache.flink</groupId>
  26.             <artifactId>flink-streaming-java_2.12</artifactId>
  27.             <version>${flink.version}</version>
  28.         </dependency>
  29.         <dependency>
  30.             <groupId>org.apache.flink</groupId>
  31.             <artifactId>flink-clients_2.12</artifactId>
  32.             <version>${flink.version}</version>
  33.         </dependency>
  34.         <dependency>
  35.             <groupId>org.apache.kudu</groupId>
  36.             <artifactId>kudu-client</artifactId>
  37.             <version>1.17.0</version>
  38.         </dependency>
  40.     </dependencies>
  41. </project>
复制代码
数据源

  1. new UserInfo("001", "Jack", 18),
  2. new UserInfo("002", "Rose", 20),
  3. new UserInfo("003", "Cris", 22),
  4. new UserInfo("004", "Lily", 19),
  5. new UserInfo("005", "Lucy", 21),
  6. new UserInfo("006", "Json", 24),
复制代码
自界说下沉器

  1. package icu.wzk.kudu;
  3. import org.apache.flink.configuration.Configuration;
  4. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
  5. import org.apache.kudu.Schema;
  6. import org.apache.kudu.Type;
  7. import org.apache.kudu.client.*;
  8. import org.apache.log4j.Logger;
  10. import java.io.ByteArrayOutputStream;
  11. import java.io.ObjectOutputStream;
  12. import java.util.Map;
  15. public class MyFlinkSinkToKudu extends RichSinkFunction<Map<String, Object>> {
  17.     private final static Logger logger = Logger.getLogger("MyFlinkSinkToKudu");
  19.     private KuduClient kuduClient;
  20.     private KuduTable kuduTable;
  22.     private String kuduMasterAddr;
  23.     private String tableName;
  24.     private Schema schema;
  25.     private KuduSession kuduSession;
  26.     private ByteArrayOutputStream out;
  27.     private ObjectOutputStream os;
  29.     public MyFlinkSinkToKudu(String kuduMasterAddr, String tableName) {
  30.         this.kuduMasterAddr = kuduMasterAddr;
  31.         this.tableName = tableName;
  32.     }
  34.     @Override
  35.     public void open(Configuration parameters) throws Exception {
  36.         out = new ByteArrayOutputStream();
  37.         os = new ObjectOutputStream(out);
  38.         kuduClient = new KuduClient.KuduClientBuilder(kuduMasterAddr).build();
  39.         kuduTable = kuduClient.openTable(tableName);
  40.         schema = kuduTable.getSchema();
  41.         kuduSession = kuduClient.newSession();
  42.         kuduSession.setFlushMode(KuduSession.FlushMode.AUTO_FLUSH_BACKGROUND);
  43.     }
  45.     @Override
  46.     public void invoke(Map<String, Object> map, Context context) throws Exception {
  47.         if (null == map) {
  48.             return;
  49.         }
  50.         try {
  51.             int columnCount = schema.getColumnCount();
  52.             Insert insert = kuduTable.newInsert();
  53.             PartialRow row = insert.getRow();
  54.             for (int i = 0; i < columnCount; i ++) {
  55.                 Object value = map.get(schema.getColumnByIndex(i).getName());
  56.                 insertData(row, schema.getColumnByIndex(i).getType(), schema.getColumnByIndex(i).getName(), value);
  57.                 OperationResponse response = kuduSession.apply(insert);
  58.                 if (null != response) {
  59.                     logger.error(response.getRowError().toString());
  60.                 }
  61.             }
  62.         } catch (Exception e) {
  63.             logger.error(e);
  64.         }
  65.     }
  67.     @Override
  68.     public void close() throws Exception {
  69.         try {
  70.             kuduSession.close();
  71.             kuduClient.close();
  72.             os.close();
  73.             out.close();
  74.         } catch (Exception e) {
  75.             logger.error(e);
  76.         }
  77.     }
  79.     private void insertData(PartialRow row, Type type, String columnName, Object value) {
  80.         try {
  81.             switch (type) {
  82.                 case STRING:
  83.                     row.addString(columnName, value.toString());
  84.                     return;
  85.                 case INT32:
  86.                     row.addInt(columnName, Integer.valueOf(value.toString()));
  87.                     return;
  88.                 case INT64:
  89.                     row.addLong(columnName, Long.valueOf(value.toString()));
  90.                     return;
  91.                 case DOUBLE:
  92.                     row.addDouble(columnName, Double.valueOf(value.toString()));
  93.                     return;
  94.                 case BOOL:
  95.                     row.addBoolean(columnName, Boolean.valueOf(value.toString()));
  96.                     return;
  97.                 case BINARY:
  98.                     os.writeObject(value);
  99.                     row.addBinary(columnName, out.toByteArray());
  100.                     return;
  101.                 case FLOAT:
  102.                     row.addFloat(columnName, Float.valueOf(value.toString()));
  103.                 default:
  104.                     throw new UnsupportedOperationException("Unknown Type: " + type);
  105.             }
  107.         } catch (Exception e) {
  108.             logger.error("插入数据异常: " + e);
  109.         }
  110.     }
  111. }
复制代码
编写实体

  1. package icu.wzk.kudu;
  3. public class UserInfo {
  5.     private String id;
  7.     private String name;
  9.     private Integer age;
  11.     public UserInfo(String id, String name, Integer age) {
  12.         this.id = id;
  13.         this.name = name;
  14.         this.age = age;
  15.     }
  17.     public String getId() {
  18.         return id;
  19.     }
  21.     public void setId(String id) {
  22.         this.id = id;
  23.     }
  25.     public String getName() {
  26.         return name;
  27.     }
  29.     public void setName(String name) {
  30.         this.name = name;
  31.     }
  33.     public Integer getAge() {
  34.         return age;
  35.     }
  37.     public void setAge(Integer age) {
  38.         this.age = age;
  39.     }
  40. }
复制代码
执行建表

  1. package icu.wzk.kudu;
  3. import org.apache.kudu.ColumnSchema;
  4. import org.apache.kudu.Schema;
  5. import org.apache.kudu.Type;
  6. import org.apache.kudu.client.CreateTableOptions;
  7. import org.apache.kudu.client.KuduClient;
  8. import org.apache.kudu.client.KuduException;
  10. import java.util.ArrayList;
  11. import java.util.List;
  13. public class KuduCreateTable {
  15.     public static void main(String[] args) throws KuduException {
  16.         String masterAddress = "localhost:7051,localhost:7151,localhost:7251";
  17.         KuduClient.KuduClientBuilder kuduClientBuilder = new KuduClient.KuduClientBuilder(masterAddress);
  18.         KuduClient kuduClient = kuduClientBuilder.build();
  20.         String tableName = "user";
  21.         List<ColumnSchema> columnSchemas = new ArrayList<>();
  22.         ColumnSchema id = new ColumnSchema
  23.                 .ColumnSchemaBuilder("id", Type.INT32)
  24.                 .key(true)
  25.                 .build();
  26.         columnSchemas.add(id);
  27.         ColumnSchema name = new ColumnSchema
  28.                 .ColumnSchemaBuilder("name", Type.STRING)
  29.                 .key(false)
  30.                 .build();
  31.         columnSchemas.add(name);
  32.         ColumnSchema age = new ColumnSchema
  33.                 .ColumnSchemaBuilder("age", Type.INT32)
  34.                 .key(false)
  35.                 .build();
  36.         columnSchemas.add(age);
  38.         Schema schema = new Schema(columnSchemas);
  39.         CreateTableOptions options = new CreateTableOptions();
  40.         // 副本数量为1
  41.         options.setNumReplicas(1);
  42.         List<String> colrule = new ArrayList<>();
  43.         colrule.add("id");
  44.         options.addHashPartitions(colrule, 3);
  46.         kuduClient.createTable(tableName, schema, options);
  47.         kuduClient.close();
  48.     }
  50. }
复制代码
主逻辑代码

  1. package icu.wzk.kudu;
  3. import org.apache.flink.api.common.functions.MapFunction;
  4. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  5. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. import java.util.HashMap;
  9. import java.util.Map;
  10. import java.util.stream.Stream;
  12. public class SinkToKuduTest {
  14.     public static void main(String[] args) throws Exception {
  15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  16.         DataStreamSource<UserInfo> dataSource = env.fromElements(
  17.                 new UserInfo("001", "Jack", 18),
  18.                 new UserInfo("002", "Rose", 20),
  19.                 new UserInfo("003", "Cris", 22),
  20.                 new UserInfo("004", "Lily", 19),
  21.                 new UserInfo("005", "Lucy", 21),
  22.                 new UserInfo("006", "Json", 24)
  23.         );
  24.         SingleOutputStreamOperator<Map<String, Object>> mapSource = dataSource
  25.                 .map(new MapFunction<UserInfo, Map<String, Object>>() {
  26.                     @Override
  27.                     public Map<String, Object> map(UserInfo value) throws Exception {
  28.                         Map<String, Object> map = new HashMap<>();
  29.                         map.put("id", value.getId());
  30.                         map.put("name", value.getName());
  31.                         map.put("age", value.getAge());
  32.                         return map;
  33.                     }
  34.                 });
  36.         String kuduMasterAddr = "localhost:7051,localhost:7151,localhost:7251";
  37.         String tableInfo = "user";
  38.         mapSource.addSink(new MyFlinkSinkToKudu(kuduMasterAddr, tableInfo));
  40.         env.execute("SinkToKuduTest");
  41.     }
  43. }
复制代码
解释分析

环境设置

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();：初始化 Flink 的执行环境，这是 Flink 应用的入口。
数据源创建

DataStreamSource dataSource = env.fromElements(…)：创建了一个包罗多个 UserInfo 对象的数据源，模拟了一个输入流。
数据转换

SingleOutputStreamOperator<Map<String, Object>> mapSource = dataSource.map(…)：使用 map 函数将 UserInfo 对象转换为 Map<String, Object>，便于后续处置惩罚和写入 Kudu。每个 UserInfo 的属性都被放入一个 HashMap 中。
Kudu 设置信息

String kuduMasterAddr = “localhost:7051,localhost:7151,localhost:7251”; 和 String tableInfo = “user”;：界说 Kudu 的主节点地点和目标表的信息。
数据下沉

mapSource.addSink(new MyFlinkSinkToKudu(kuduMasterAddr, tableInfo));：将转换后的数据流添加到 Kudu 的自界说 Sink 中。MyFlinkSinkToKudu 类应该实现了将数据写入 Kudu 的逻辑。
执行作业

env.execute(“SinkToKuduTest”);：启动 Flink 作业，执行整个数据流处置惩罚流程。
测试运行



  • 先运行建表
  • 再运行主逻辑
我们建表之后，确认user表存在。然后我们运行Flink步伐，将数据写入Kudu。

确认有表后，执行 Flink 步伐：

注意事项



  • 并发性：根据 Kudu 集群的规模和设置，可以调解 Flink 作业的并发性，以进步写入性能。
  • 批量写入：Kudu 支持批量插入，可以通过适当设置 Flink 的 sink 来进步性能。
  • 故障处置惩罚：确保在作业中处置惩罚异常和重试逻辑，以确保数据不会丢失。
  • 监控与调试：使用 Flink 的监控工具和 Kudu 的工具（如 Kudu UI）来监控数据流和性能。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

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

本版积分规则

络腮胡菲菲

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

楼主热帖

标签云

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