中文介绍

2025年9月：LakeSoul发布了最新版本3.0.0，请查看我们的发布说明

LakeSoul是一个云原生的湖仓框架，支持可扩展的元数据管理、ACID事务、高效灵活的upsert操作、模式演化以及流批一体化处理。

LakeSoul支持多种计算引擎读写湖仓表数据，包括Spark、Flink、Presto和PyTorch，并且支持批处理、流处理、MPP和AI等多种计算模式。LakeSoul还兼容HDFS和S3等存储系统。

LakeSoul最初由DMetaSoul公司创建，自2023年5月起作为沙盒项目捐赠给Linux基金会AI与数据组织。

LakeSoul实现了行级和列级的增量upsert操作，并支持并发更新。

LakeSoul采用类似LSM树的结构来支持带有主键的哈希分区表上的更新，在提供优化的读时合并性能的同时，实现了极高的写入吞吐量（详情请参阅性能基准测试）。LakeSoul通过使用PostgreSQL实现元数据管理的扩展性和ACID控制。

LakeSoul使用Rust语言实现了原生的元数据层和IO层，并提供了C/Java/Python接口，以支持大数据和AI等多种计算框架的连接。

LakeSoul支持并发的批处理或流式读写。无论是读取还是写入都支持CDC语义，结合自动模式演化和精确一次保证，可以轻松构建实时数据仓库。

LakeSoul支持多工作空间和RBAC权限管理。它利用PostgreSQL的RBAC和行级安全策略来实现元数据的权限隔离。结合Hadoop的用户和组机制，可以实现物理数据的隔离。LakeSoul的权限隔离对SQL、Java和Python作业均有效。

LakeSoul支持自动的分层多级压缩、自动的表生命周期维护、自动的数据资产统计以及自动的冗余数据清理功能，从而降低运维成本并提升易用性。

更多详细功能请参考我们的文档页面：文档

快速入门

按照快速入门指南，您可以快速搭建一个测试环境。

教程

请在文档网站中查找相关教程：

• 查看Python数据处理与AI模型训练在LakeSoul上的示例，了解如何将AI与湖仓结合，构建统一的现代化数据基础设施。
• 查看LakeSoul Flink CDC整库同步教程，学习如何将整个MySQL数据库实时同步到LakeSoul，实现自动建表、自动DDL同步以及精确一次保证。
• 查看Flink SQL使用指南，了解如何在批处理和流处理模式下使用Flink SQL读写LakeSoul，同时支持Flink Changelog Stream语义以及行级的upsert和删除操作。
• 查看多流合并与宽表构建教程，学习如何在不进行Join的情况下，并发地合并具有相同主键（但其他列不同的）的多个流。
• 查看Upsert数据与Merge UDF教程，了解如何使用upsert数据和Merge UDF来自定义合并逻辑。
• 查看快照API使用指南，学习如何进行快照读取（时间旅行）、快照回滚和清理。
• 查看增量查询教程，了解如何在Spark中以批处理或流式方式执行增量查询。

使用文档

请在文档网站中查找使用文档： 使用文档

快速开始

教程

使用文档

功能路线图

• 数据科学与人工智能
  • 原生 Python 读取器（无需 PySpark）
  • PyTorch 数据集及分布式训练
• 元数据管理 (#23)
  • 多级分区：多个范围分区及最多一个哈希分区
  • 并发写入与自动冲突解决
  • MVCC 及读隔离
  • 通过 Postgres 事务实现的写事务（两阶段提交）
  • Schema Evolution：支持列的添加与删除
• 表操作
  • 哈希分区表的 LSM-Tree 风格 upsert
  • 带 upsert 增量文件的哈希分区 Merge on read
  • 非哈希分区表的 Copy on write 更新
  • 自动去重合并服务
• 数据仓库
  • CDC 流式摄取与自动 DDL 同步
  • 增量查询与快照查询
    • 快照查询 (#103)
    • 增量查询 (#103)
    • 增量流式源 (#130)
    • Flink 流/批源
  • 多工作空间与 RBAC
• Spark 集成
  • Table/Dataframe API
  • 支持 SQL，但不包括 upsert 操作
  • 查询优化
    • 对主键操作的 Shuffle/Join 消除
  • Merge UDF（合并算子）
  • Merge Into SQL 支持
    • 基于主键匹配的 Merge Into SQL（Merge on read）
• Flink 集成与 CDC 摄取 (#57)
  • Table API
    • 批/流 Sink
    • 批/流 Source
    • ChangeLog 流语义的流 Source/Sink
    • 精确一次 Source 和 Sink
  • Flink CDC
    • 自动模式变更（DDL）同步
    • 自动建表（依赖于 #78）
    • 支持将不同 schema 的多张源表写入同一目标表 (#84)
• Hive 集成
  • 合并后导出至 Hive 分区
  • Apache Kyuubi（Hive JDBC）集成
• 实时数据仓库
  • CDC 摄取
  • 时间旅行（快照读取）
  • 快照回滚
  • 自动全局合并服务
  • MPP 引擎集成（依赖于 #66）
    • Presto
    • 与 Presto 原生执行（使用 Velox）兼容
    • Apache Doris
• 云原生 IO (#66)
  • 对象存储 IO 优化
  • 原生向量化 Merge on read
  • 支持多层存储类别，并配备本地磁盘数据缓存

社区准则

社区准则

反馈与贡献

如果您有任何问题，请随时提出 issue 或发起讨论。

欢迎加入我们的 Discord 服务器进行交流。

联系我们

请发送邮件至 lakesoul-technical-discuss@lists.lfaidata.foundation。

开源许可证

LakeSoul 采用 Apache License v2.0 开源。

LakeSoul 快速上手指南

LakeSoul 是一个云原生湖仓一体框架，支持可扩展的元数据管理、ACID 事务、高效 Upsert 操作、Schema 演进以及统一的流批处理。它兼容 Spark、Flink、Presto 等多种计算引擎，并支持 HDFS 和 S3 等存储系统。

1. 环境准备

在开始之前，请确保您的开发环境满足以下要求：

• 操作系统: Linux 或 macOS (Windows 需通过 WSL2 运行)
• Java: JDK 8 或 JDK 11 (推荐 JDK 11)
• Scala: 2.12.x (通常随 Spark/Flink 环境自带)
• 数据库: PostgreSQL 9.4+ (用于存储元数据，必须预先安装并启动)
• 计算引擎 (任选其一或全部):
  • Apache Spark 3.x
  • Apache Flink 1.14+
• 构建工具: Maven 3.6+ (如需从源码编译)

注意: LakeSoul 依赖 PostgreSQL 进行元数据管理和 ACID 控制，请务必提前配置好 PostgreSQL 服务并创建对应的数据库和用户。

2. 安装步骤

方式一：使用 Maven 依赖（推荐）

如果您已有 Spark 或 Flink 项目，只需在 pom.xml 中添加 LakeSoul 连接器依赖。

Spark 连接器:

<dependency>
    <groupId>io.lakesoul</groupId>
    <artifactId>lakesoul-spark</artifactId>
    <version>3.0.0</version>
</dependency>

Flink 连接器:

<dependency>
    <groupId>io.lakesoul</groupId>
    <artifactId>lakesoul-flink</artifactId>
    <version>3.0.0</version>
</dependency>

国内加速: 如果下载依赖较慢，建议在 Maven settings.xml 中配置阿里云镜像：

<mirror>
  <id>aliyunmaven</id>
  <mirrorOf>*</mirrorOf>
  <name>Aliyun Public</name>
  <url>https://maven.aliyun.com/repository/public</url>
</mirror>

方式二：本地测试环境搭建

若要快速搭建本地测试环境，可参考官方脚本初始化元数据（需确保 PostgreSQL 已运行）：

1. 克隆项目：

   git clone https://github.com/lakesoul-io/LakeSoul.git
   cd LakeSoul
   

2. 初始化元数据表结构（假设 PG 地址为 localhost:5432，用户/库名为 lakesoul）：

   # 执行 SQL 初始化脚本 (路径可能在 tools 或 docs 中，具体参考官方 Setup 文档)
   psql -h localhost -p 5432 -U lakesoul -d lakesoul -f scripts/init_meta.sql
   

3. 设置环境变量（在提交任务时传递）：

   export LAKESOUL_META_DB_URL="jdbc:postgresql://localhost:5432/lakesoul"
   export LAKESOUL_META_DB_USER="lakesoul"
   export LAKESOUL_META_DB_PASSWORD="your_password"
   

3. 基本使用

以下示例展示如何使用 Spark 和 Flink 进行最简单的数据写入与读取。

场景 A：使用 Spark 进行 Upsert 写入与查询

LakeSoul 在 Spark 中支持主键表的 Upsert 操作。

import org.apache.spark.sql.SparkSession
import io.lakesoul.LakeSoulTable

val spark = SparkSession.builder()
  .appName("LakeSoulQuickStart")
  .master("local[*]")
  .config("spark.sql.extensions", "io.lakesoul.sql.LakeSoulSparkSessionExtension")
  .config("spark.sql.catalog.spark_catalog", "io.lakesoul.catalog.LakeSoulCatalog")
  // 配置元数据连接信息
  .config("lakesoul.meta.db.url", "jdbc:postgresql://localhost:5432/lakesoul")
  .config("lakesoul.meta.db.user", "lakesoul")
  .config("lakesoul.meta.db.password", "your_password")
  .getOrCreate()

import spark.implicits._

// 1. 准备测试数据
val data = Seq(
  (1, "Alice", 25),
  (2, "Bob", 30)
).toDF("id", "name", "age")

// 2. 写入数据 (自动建表，指定主键为 id)
data.write
  .format("lakesoul")
  .option("rangePartitions", "id") // 可选：范围分区
  .option("hashPartitions", "id")  // 可选：哈希分区 (用于高效 Upsert)
  .option("primaryKeys", "id")     // 指定主键
  .mode("overwrite")
  .save("/tmp/lakesoul_table")

// 3. 执行 Upsert 更新 (插入新数据或更新已存在的主键)
val updateData = Seq(
  (1, "Alice Updated", 26), // 更新 id=1
  (3, "Charlie", 35)        // 新增 id=3
).toDF("id", "name", "age")

updateData.write
  .format("lakesoul")
  .option("primaryKeys", "id")
  .mode("append") // LakeSoul 会根据主键自动合并
  .save("/tmp/lakesoul_table")

// 4. 读取数据
val df = spark.read.format("lakesoul").load("/tmp/lakesoul_table")
df.show()

场景 B：使用 Flink CDC 实时同步 MySQL 数据

LakeSoul 结合 Flink CDC 可实现整库实时同步，支持自动建表和 Schema 演进。

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkCDCSync {
    public static void main(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        TableEnvironment tEnv = TableEnvironment.create(settings);

        // 加载 LakeSoul 连接器
        tEnv.executeSql("CREATE CATALOG lakesoul WITH ('type'='lakesoul')");
        tEnv.useCatalog("lakesoul");

        // 配置元数据 (也可以通过 JVM 参数传递)
        tEnv.executeSql("SET 'lakesoul.meta.db.url' = 'jdbc:postgresql://localhost:5432/lakesoul'");
        tEnv.executeSql("SET 'lakesoul.meta.db.user' = 'lakesoul'");
        tEnv.executeSql("SET 'lakesoul.meta.db.password' = 'your_password'");

        // 定义 MySQL CDC Source (示例)
        String sourceDDL = 
            "CREATE TABLE mysql_source (" +
            "  id INT," +
            "  name STRING," +
            "  age INT," +
            "  PRIMARY KEY (id) NOT ENFORCED" +
            ") WITH (" +
            "  'connector' = 'mysql-cdc'," +
            "  'hostname' = 'localhost'," +
            "  'port' = '3306'," +
            "  'username' = 'root'," +
            "  'password' = 'root'," +
            "  'database-name' = 'test_db'," +
            "  'table-name' = 'users'" +
            ")";
        tEnv.executeSql(sourceDDL);

        // 写入 LakeSoul (自动创建目标表，支持 Schema 同步)
        String sinkDDL = 
            "CREATE TABLE lakesoul_sink (" +
            "  id INT," +
            "  name STRING," +
            "  age INT," +
            "  PRIMARY KEY (id) NOT ENFORCED" +
            ") WITH (" +
            "  'connector' = 'lakesoul'," +
            "  'table.path' = '/tmp/lakesoul_flink_table'" +
            ")";
        tEnv.executeSql(sinkDDL);

        // 执行同步
        tEnv.executeSql("INSERT INTO lakesoul_sink SELECT * FROM mysql_source");
    }
}

关键特性提示

• 时间旅行 (Time Travel): 读取历史版本数据，例如 spark.read.option("time_travel_version", "v1").format("lakesoul").load(...)。
• 并发写入: LakeSoul 原生支持多任务并发写入同一张表，无需外部锁。
• 自动 Compaction: 系统会自动触发后台 Compaction 服务优化小文件，无需手动干预。

更多详细教程（如 Python AI 训练集成、多流合并等）请访问 LakeSoul 官方文档。