HuaweiCloud-Hudi-plugins:基于华为云OBS和DIS的数据湖存储格式插件

Hudi

Current English DOC | 查阅中文文档

项目背景

Hudi是一种数据湖的存储格式，在Hadoop文件系统之上提供了更新数据和删除数据的能力以及消费变化数据的能力。支持多种计算引擎，提供IUD接口，在 HDFS的数据集上提供了插入更新和增量拉取的流原语。

参考资料：https://support.huaweicloud.com/productdesc-mrs/mrs_08_0083.html

使用文档

1.使用前准备

(1) 在华为云OBS控制台，创建一个桶，用于存放Hudi写出的数据 https://console.huaweicloud.com/console/#/obs/manager/buckets

(2) 在华为云DIS控制台，创建一个通道，用于配置OBS事件通知 https://console.huaweicloud.com/dis/

注意：在创建通道时，数据源类型需要选择 JSON

2.参数配置方式

在 core-site.xml 中，配置上链接OBS所需的信息

<configuration>
    <property>
        <name>fs.defaultFS</name>
        <value>obs://bucketname/</value>
    </property>
    <property>
        <name>fs.obs.access.key</name>
        <value>Huawei Cloud access key</value>
    </property>
    <property>
        <name>fs.obs.secret.key</name>
        <value>Huawei Cloud secret key</value>
    </property>
    <property>
        <name>fs.obs.endpoint</name>
        <value>Huawei Cloud OBS endpoint to connect to</value>
    </property>
    <property>
        <name>fs.obs.impl</name>
        <value>org.apache.hadoop.fs.obs.OBSFileSystem</value>
    </property>
</configuration>

在 dis.properties 中，配置上链接DIS事件通知所需的信息

endpoint=Huawei Cloud DIS endpoint to connect to
region=Huawei Cloud DIS region
ak=Huawei Cloud access key
sk=Huawei Cloud secret key
projectId=The projectId corresponding to the Huawei Cloud DIS region

注意：endpoint 需要在开头加上 https://，例如 https://dis.cn-north-4.myhuaweicloud.com

3.运行方式

参考 com.xnx3.obs.sources.TestDISEventSource，使用 HoodieWriterConfig 和 DISReaderConfig 对运行所需的参数进行配置

HoodieWriterConfig config = new HoodieWriterConfig()
        .basePath("obs://hudi-test-target/hudi_dis_cow")
        .tableName("hudi_dis_cow")
        .saveMode(SaveMode.Append)
        .keyGenerator(NonpartitionedKeyGenerator.class.getName())
        .recordkeyFieldOptKey("partitionKey")
        .precombineFieldOptKey("timestamp");

DISReaderConfig dicConfig = new DISReaderConfig()
        .streamName("hudi-dis-test")
        .partitionId("0")
        .startingSequenceNumber("0")
        .cursorType(PartitionCursorTypeEnum.AT_SEQUENCE_NUMBER.name());

new DISEventSource().fetchEvents(config, dicConfig);