flink读写hudi的demo
ckckck2025年1月10日4 浏览Flink DataStream API 读取 Hudi 数据
1. POM 依赖
xml
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-common</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-flink1.15-bundle</artifactId>
<version>${hudi.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java-bridge_2.11</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner_2.11</artifactId>
<version>${flink.version}</version>
</dependency>
</dependencies>2. DataStream API 读取 Hudi 数据示例
java
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.data.RowData;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.util.HoodiePipeline;
import java.util.HashMap;
import java.util.Map;
public class FlinkDataStreamReadFromHudiTest {
public static void main(String[] args) throws Exception {
// 1. 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 2. 创建 Hudi 数据流
String studentHudiTable = "ods_student_table";
String studentHudiTablePath = "hdfs://hw-cdh-test02:8020/user/hive/warehouse/lake/" studentHudiTable;
Map<String, String> studentOptions = new HashMap<>();
studentOptions.put(FlinkOptions.PATH.key(), studentHudiTablePath);
studentOptions.put(FlinkOptions.TABLE_TYPE.key(), HoodieTableType.MERGE_ON_READ.name());
studentOptions.put(FlinkOptions.READ_AS_STREAMING.key(), "true"); // 启用流式读取
studentOptions.put(FlinkOptions.READ_START_COMMIT.key(), "16811748000000"); // 指定起始提交时间
studentOptions.put(FlinkOptions.READ_STREAMING_CHECK_INTERVAL.key(), "4");
studentOptions.put(FlinkOptions.CHANGELOG_ENABLED.key(), "true");
HoodiePipeline.Builder studentBuilder = HoodiePipeline.builder(studentHudiTable)
.column("id BIGINT")
.column("name STRING")
.column("age BIGINT")
.column("hobby STRING")
.pk("id")
.options(studentOptions);
DataStream<RowData> studentRowDataDS = studentBuilder.source(env);
// 3. 数据转换与输出
DataStream<Student> studentDS = studentRowDataDS.map(new MapFunction<RowData, Student>() {
@Override
public Student map(RowData value) throws Exception {
try {
String rowKind = value.getRowKind().name();
Long id = value.getLong(0);
String name = value.getString(1).toString();
Long age = value.getLong(2);
String hobby = value.getString(3).toString();
return new Student(id, name, age, hobby, rowKind);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
});
studentDS.print();
env.execute("FlinkDataStreamReadFromHudiTest");
}
public static class Student {
public Long id;
public String name;
public Long age;
public String hobby;
public String rowKind;
public Student() {}
public Student(Long id, String name, Long age, String hobby, String rowKind) {
this.id = id;
this.name = name;
this.age = age;
this.hobby = hobby;
this.rowKind = rowKind;
}
@Override
public String toString() {
return "Student{"
"id=" id
", name=\'" name \'\\'\'
", age=" age
", hobby=\'" hobby \'\\'\'
", rowKind=\'" rowKind \'\\'\'
\'}\';
}
}
}3. DataStream API 写入 Hudi 数据示例
java
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.util.HoodiePipeline;
import java.util.HashMap;
import java.util.Map;
public class FlinkDataStreamWrite2HudiTest {
public static void main(String[] args) throws Exception {
// 1. 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 2. 开启 Checkpoint
String checkPointPath = "hdfs://hw-cdh-test02:8020/flinkinfo/meta/savepoints/FlinkDataStreamWrite2HudiTest";
StateBackend backend = new EmbeddedRocksDBStateBackend(true);
env.setStateBackend(backend);
CheckpointConfig conf = env.getCheckpointConfig();
conf.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
conf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
conf.setCheckpointInterval(1000); // 毫秒
conf.setCheckpointTimeout(10 * 60 * 1000); // 毫秒
conf.setMinPauseBetweenCheckpoints(2 * 1000); // 毫秒
conf.setCheckpointStorage(checkPointPath);
// 3. 准备数据
DataStreamSource<Student> studentDS = env.fromElements(
new Student(101L, "Johnson", 17L, "swimming"),
new Student(102L, "Lin", 15L, "shopping"),
new Student(103L, "Tom", 5L, "play")
);
// 4. 创建 Hudi 数据流
String studentHudiTable = "ods_student_table";
String studentHudiTablePath = "hdfs://hw-cdh-test02:8020/user/hive/warehouse/lake/" studentHudiTable;
Map<String, String> studentOptions = new HashMap<>();
studentOptions.put(FlinkOptions.PATH.key(), studentHudiTablePath);
studentOptions.put(FlinkOptions.TABLE_TYPE.key(), HoodieTableType.MERGE_ON_READ.name());
HoodiePipeline.Builder studentBuilder = HoodiePipeline.builder(studentHudiTable)
.column("id BIGINT")
.column("name STRING")
.column("age BIGINT")
.column("hobby STRING")
.pk("id")
.options(studentOptions);
// 5. 转成 RowData 流
DataStream<RowData> studentRowDataDS = studentDS.map(new MapFunction<Student, RowData>() {
@Override
public RowData map(Student value) throws Exception {
try {
Long id = value.id;
String name = value.name;
Long age = value.age;
String hobby = value.hobby;
GenericRowData row = new GenericRowData(4);
row.setField(0, Long.valueOf(id));
row.setField(1, StringData.fromString(name));
row.setField(2, Long.valueOf(age));
row.setField(3, StringData.fromString(hobby));
return row;
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
});
studentBuilder.sink(studentRowDataDS, false);
env.execute("FlinkDataStreamWrite2HudiTest");
}
public static class Student {
public Long id;
public String name;
public Long age;
public String hobby;
public Student() {}
public Student(Long id, String name, Long age, String hobby) {
this.id = id;
this.name = name;
this.age = age;
this.hobby = hobby;
}
@Override
public String toString() {
return "Student{"
"id=" id
", name=\'" name \'\\'\'
", age=" age
", hobby=\'" hobby \'\\'\'
\'}\';
}
}
}Flink Table API 读取 Hudi 数据
java
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
public class FlinkSqlReadFromHudiTest {
public static void main(String[] args) throws Exception {
// 1. 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);
// 2. 准备 Hudi 表的数据流，并从 Hudi 表读取数据
tabEnv.executeSql(""
"CREATE TABLE out_ods_student_table(\n"
" id BIGINT COMMENT \'学号\',\n"
" name STRING COMMENT \'姓名\',\n"
" age BIGINT COMMENT \'年龄\',\n"
" hobby STRING COMMENT \'爱好\',\n"
" PRIMARY KEY (id) NOT ENFORCED\n"
")\n"
"WITH(\n"
" \'connector\' = \'hudi\',\n"
" \'path\' = \'hdfs://hw-cdh-test02:8020/user/hive/warehouse/lake/ods_student_table\',\n"
" \'table.type\' = \'MERGE_ON_READ\',\n"
" \'compaction.async.enabled\' = \'true\',\n"
" \'compaction.tasks\' = \'1\',\n"
" \'compaction.trigger.strategy\' = \'num_commits\',\n"
" \'compaction.delta_commits\' = \'3\',\n"
" \'hoodie.cleaner.policy\'=\'KEEP_LATEST_COMMITS\',\n"
" \'hoodie.cleaner.commits.retained\'=\'30\',\n"
" \'hoodie.keep.min.commits\'=\'35\',\n"
" \'hoodie.keep.max.commits\'=\'40\'\n"
")");
tabEnv.executeSql("select id, name, age, hobby from out_ods_student_table").print();
env.execute("FlinkSqlReadFromHudiTest");
}
}Flink Table API 写入 Hudi 数据（数据来源于 DataStream）
java
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
public class FlinkDataStreamSqlWrite2HudiTest {
public static void main(String[] args) throws Exception {
// 1. 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);
// 2. 开启 Checkpoint
String checkPointPath = "hdfs://hw-cdh-test02:8020/flinkinfo/meta/savepoints/FlinkDataStreamWrite2HudiTest";
StateBackend backend = new EmbeddedRocksDBStateBackend(true);
env.setStateBackend(backend);
CheckpointConfig conf = env.getCheckpointConfig();
conf.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
conf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
conf.setCheckpointInterval(1000); // 毫秒
conf.setCheckpointTimeout(10 * 60 * 1000); // 毫秒
conf.setMinPauseBetweenCheckpoints(2 * 1000); // 毫秒
conf.setCheckpointStorage(checkPointPath);
// 3. 准备数据
DataStreamSource<Student> studentDS = env.fromElements(
new Student(201L, "zhangsan", 117L, "eat"),
new Student(202L, "lisi", 115L, "drink"),
new Student(203L, "wangwu", 105L, "sleep")
);
// 4. 通过 DataStream 创建表
tabEnv.registerDataStream("tmp_student_table", studentDS, "id, name, age, hobby");
// 5. 准备 Hudi 表的数据流，并将数据写入 Hudi 表
tabEnv.executeSql(""
"CREATE TABLE out_ods_student_table(\n"
" id BIGINT COMMENT \'学号\',\n"
" name STRING COMMENT \'姓名\',\n"
" age BIGINT COMMENT \'年龄\',\n"
" hobby STRING COMMENT \'爱好\',\n"
" PRIMARY KEY (id) NOT ENFORCED\n"
")\n"
"WITH(\n"
" \'connector\' = \'hudi\',\n"
" \'path\' = \'hdfs://hw-cdh-test02:8020/user/hive/warehouse/lake/ods_student_table\',\n"
" \'table.type\' = \'MERGE_ON_READ\',\n"
" \'compaction.async.enabled\' = \'true\',\n"
" \'compaction.tasks\' = \'1\',\n"
" \'compaction.trigger.strategy\' = \'num_commits\',\n"
" \'compaction.delta_commits\' = \'3\',\n"
" \'hoodie.cleaner.policy\'=\'KEEP_LATEST_COMMITS\',\n"
" \'hoodie.cleaner.commits.retained\'=\'30\',\n"
" \'hoodie.keep.min.commits\'=\'35\',\n"
" \'hoodie.keep.max.commits\'=\'40\'\n"
")");
tabEnv.executeSql("insert into out_ods_student_table select id, name, age, hobby from tmp_student_table");
env.execute("FlinkDataStreamSqlWrite2HudiTest");
}
public static class Student {
public Long id;
public String name;
public Long age;
public String hobby;
public Student() {}
public Student(Long id, String name, Long age, String hobby) {
this.id = id;
this.name = name;
this.age = age;
this.hobby = hobby;
}
@Override
public String toString() {
return "Student{"
"id=" id
", name=\'" name \'\\'\'
", age=" age
", hobby=\'" hobby \'\\'\'
\'}\';
}
}
}Flink Table API 写入 Hudi 数据（数据来源于 Table 表）
java
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
public class FlinkValuesSqlWrite2HudiTest {
public static void main(String[] args) throws Exception {
// 1. 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);
// 2. 开启 Checkpoint
String checkPointPath = "hdfs://hw-cdh-test02:8020/flinkinfo/meta/savepoints/FlinkDataStreamWrite2HudiTest";
StateBackend backend = new EmbeddedRocksDBStateBackend(true);
env.setStateBackend(backend);
CheckpointConfig conf = env.getCheckpointConfig();
conf.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
conf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
conf.setCheckpointInterval(1000); // 毫秒
conf.setCheckpointTimeout(10 * 60 * 1000); // 毫秒
conf.setMinPauseBetweenCheckpoints(2 * 1000); // 毫秒
conf.setCheckpointStorage(checkPointPath);
// 3. 准备 Hudi 表的数据流，并将数据写入 Hudi 表
tabEnv.executeSql(""
"CREATE TABLE out_ods_student_table(\n"
" id BIGINT COMMENT \'学号\',\n"
" name STRING COMMENT \'姓名\',\n"
" age BIGINT COMMENT \'年龄\',\n"
" hobby STRING COMMENT \'爱好\',\n"
" PRIMARY KEY (id) NOT ENFORCED\n"
")\n"
"WITH(\n"
" \'connector\' = \'hudi\',\n"
" \'path\' = \'hdfs://hw-cdh-test02:8020/user/hive/warehouse/lake/ods_student_table\',\n"
" \'table.type\' = \'MERGE_ON_READ\',\n"
" \'compaction.async.enabled\' = \'true\',\n"
" \'compaction.tasks\' = \'1\',\n"
" \'compaction.trigger.strategy\' = \'num_commits\',\n"
" \'compaction.delta_commits\' = \'3\',\n"
" \'hoodie.cleaner.policy\'=\'KEEP_LATEST_COMMITS\',\n"
" \'hoodie.cleaner.commits.retained\'=\'30\',\n"
" \'hoodie.keep.min.commits\'=\'35\',\n"
" \'hoodie.keep.max.commits\'=\'40\'\n"
")");
tabEnv.executeSql(""
"insert into out_ods_student_table values\n"
" (301, \'xiaoming\', 201, \'read\'),\n"
" (302, \'xiaohong\', 202, \'write\'),\n"
" (303, \'xiaogang\', 203, \'sing\')");
env.execute("FlinkValuesSqlWrite2HudiTest");
}
}
end