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

Flink SQL+Hudi数据入湖与分析实践

news 2025/7/7 1:04:16

数据湖构建（Hudi）可以结合实时计算Flink版，以及Flink CDC相关技术，实现灵活定制化的数据入湖。并利用DLF统一元数据管理、权限管理等能力，实现数据湖多引擎分析、数据湖管理等功能。本文为您介绍Flink+DLF数据湖方案具体步骤。

背景信息

操作流程

步骤一：准备MySQL数据

CREATE DATABASE testdb;
CREATE TABLE testdb.student (`id` bigint(20) NOT NULL,`name` varchar(256) DEFAULT NULL,`age` bigint(20) DEFAULT NULL,PRIMARY KEY (`id`)
);INSERT INTO testdb.student VALUES (1,'name1',10);
INSERT INTO testdb.student VALUES (2,'name2',20);

步骤二：创建Flink入湖作业

-- 创建数据源表
CREATE TABLE IF NOT EXISTS student_source (id INT,name VARCHAR (256),age INT,PRIMARY KEY (id) NOT ENFORCED
)
WITH ('connector' = 'mysql-cdc',-- hostname替换为RDS的连接地址'hostname' = 'rm-xxxxxxxx.mysql.rds.aliyuncs.com','port' = '3306','username' = '<RDS user name>','password' = '<RDS password>','database-name' = '<RDS database>',-- table-name为数据源表，本例中填步骤二创建的student表'table-name' = 'student'
);-- catalog名为步骤二创建的dlf catalog name，本例中填dlf
CREATE DATABASE IF NOT EXISTS dlf.dlf_testdb;-- 创建目标表
CREATE TABLE IF NOT EXISTS dlf.dlf_testdb.student_hudi (id    BIGINT PRIMARY KEY NOT ENFORCED,name  STRING,age    BIGINT
) WITH('connector' = 'hudi'
);-- 创建流SQL作业
INSERT INTO dlf.dlf_testdb.student_hudi SELECT * FROM student_source  /*+ OPTIONS('server-id'='123456') */;