hudi的lakehouse

前言

starrocks提供了链接iceberg和hudi的能力，如下图所示，经过metastore元数据就可以对外部数据湖的数据进行访问，

该文档详细描述了如何配置和启动一个集成 StarRocks、Hudi、MinIO 和 Spark 的演示环境。以下是主要内容：

前提条件：
- 克隆 StarRocks 的演示仓库，确保所有操作从 demo/documentation-samples/hudi/ 目录开始。
- 配置 Docker 并分配足够的内存（至少 5 GB）和 CPU（至少 4 个）。
SQL 客户端：
- 可以选择 Docker 环境中的 SQL 客户端或其他 MySQL 兼容客户端。
Docker 服务：
- starrocks-fe: 元数据管理、查询规划和调度。
- starrocks-be: 查询计划的执行。
- metastore_db: 用于存储 Hive 元数据的 Postgres 数据库。
- hive_metastore: 提供 Apache Hive 的元数据服务。
- minio 和 mc: MinIO 对象存储及其命令行客户端。
- spark-hudi: 集成的 MinIO 配置服务。
配置文件：
- core-site.xml: 对象存储相关设置。
- spark-defaults.conf: Hive、MinIO 和 Spark SQL 的设置。
- hudi-defaults.conf: 用于在 Spark-shell 中静默警告。
- hadoop-metrics2-hbase.properties 和 hadoop-metrics2-s3a-file-system.properties: 同样用于消除警告。
启动演示集群：
- 在包含 docker-compose.yml 文件的目录中运行以下命令启动集群：
  1
  
  docker compose up --detach --wait --wait-timeout 60

注意：当有很多容器时，可以通过jq来进行处理

1

jq '{Service: .Service, State: .State, Status: .Status}'

输出效果如下：

创建bucket：

因为spark要使用huditest这个bucket，接下来就在minio上创建对应的bucket 创建bucket：

创建表格, 接下来写数据到Hive中：

打开spark-shell，命令如下：

1

docker compose exec spark-hudi spark-shell

注意输出的这些警告是正常的，不影响使用

接下来就可以把下面的代码输入到’scala>‘之后就可以了

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42


import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._
import org.apache.spark.sql.Row
import org.apache.spark.sql.SaveMode._
import org.apache.hudi.DataSourceReadOptions._
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig._
import scala.collection.JavaConversions._

val schema = StructType( Array(
                 StructField("language", StringType, true),
                 StructField("users", StringType, true),
                 StructField("id", StringType, true)
             ))

val rowData= Seq(Row("Java", "20000", "a"),
               Row("Python", "100000", "b"),
               Row("Scala", "3000", "c"))


val df = spark.createDataFrame(rowData,schema)

val databaseName = "hudi_sample"
val tableName = "hudi_coders_hive"
val basePath = "s3a://huditest/hudi_coders"

df.write.format("hudi").
  option(org.apache.hudi.config.HoodieWriteConfig.TABLE_NAME, tableName).
  option(RECORDKEY_FIELD_OPT_KEY, "id").
  option(PARTITIONPATH_FIELD_OPT_KEY, "language").
  option(PRECOMBINE_FIELD_OPT_KEY, "users").
  option("hoodie.datasource.write.hive_style_partitioning", "true").
  option("hoodie.datasource.hive_sync.enable", "true").
  option("hoodie.datasource.hive_sync.mode", "hms").
  option("hoodie.datasource.hive_sync.database", databaseName).
  option("hoodie.datasource.hive_sync.table", tableName).
  option("hoodie.datasource.hive_sync.partition_fields", "language").
  option("hoodie.datasource.hive_sync.partition_extractor_class", "org.apache.hudi.hive.MultiPartKeysValueExtractor").
  option("hoodie.datasource.hive_sync.metastore.uris", "thrift://hive-metastore:9083").
  mode(Overwrite).
  save(basePath)
System.exit(0)

这段代码实现了如下功能：

配置spark session，加载处理，写入数据
创建一个dataframe 同时写入到hudi的表中
同步数据到hivemetastore 当出现如下图展示即可退出了

接下来starRocks中操作了

连接StarRocks

1
2


docker compose exec starrocks-fe \
  mysql -P 9030 -h 127.0.0.1 -u root --prompt="StarRocks > "

连接starrocks和hudi

接下来使用外部的catalog来处理hudi的数据

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


CREATE EXTERNAL CATALOG hudi_catalog_hms
PROPERTIES
(
    "type" = "hudi",
    "hive.metastore.type" = "hive",
    "hive.metastore.uris" = "thrift://hive-metastore:9083",
    "aws.s3.use_instance_profile" = "false",
    "aws.s3.access_key" = "admin",
    "aws.s3.secret_key" = "password",
    "aws.s3.enable_ssl" = "false",
    "aws.s3.enable_path_style_access" = "true",
    "aws.s3.endpoint" = "http://minio:9000"
);

使用catalog

1

SET CATALOG hudi_catalog_hms;

使用hudi的数据库l

1

show databases;

前言

创建bucket：

创建 表格, 接下来写数据到Hive中：

接下来starRocks中操作了

连接StarRocks

连接starrocks和hudi

接下来查询这个表

创建表格, 接下来写数据到Hive中：