作者 | 郭炜

编辑 | Debra Chen

在当今的商业环境中，大数据的管理和应用已经成为企业决策和运营的核心组成部分。然而，随着数据量的爆炸性增长，如何有效利用这些数据成为了一个普遍的挑战。

本文将探讨大数据架构、大模型的集成，以及如何将大模型集成到公司大数据架构中，并使用Apache SeaTunnel和WhaleStudio将公司内部数据进行“百科全书化”，利用大数据和大模型来提升企业运营效率。

大模型在整体公司大数据架构中的位置

当今，无论大企业还是小公司，其实都会遇到同样的问题：公司里沉淀的数据量巨大，但到底该怎么使用？

大模型的横空出世让数据利用有了全新的使用途径，问题是如何大量获得公司的数据，变成“你”的大模型？

以及如何将大模型灌入公司内部数据，并“百科全书”化？

大数据与大模型架构概览

为了更好地回答这些问题，我们首先需要弄清楚大模型在企业复杂的数据结构中处于什么位置。目前，全球流行的大数据结构图如下所示：

企业在处理大数据时，通常会将数据分为实时数据和批量数据两大类。实时数据可以来自车联网、数据库日志、点击流等多种来源，而批量数据则可能包括文件、报表、CSV文件等。这些数据可以通过各种工具和技术，如Apache Kafka、Amazon Kinesis等进行处理，最终被整合到企业的大数据分析系统中。

大模型在大数据架构中扮演着至关重要的角色。它们能够处理和分析大量数据，为企业提供深入的洞察和预测。大模型可以通过两种主要方法进行集成：

1. 基于开源模型的优化：企业可以使用开源的大模型，并根据自己的数据进行优化，以提高模型的性能。这种方法虽然复杂，对于普通用户来做操作比较困难，但可以训练出高度定制化的模型，具体训练方法可以参考《用一杯星巴克的钱，训练自己私有化的ChatGPT》
2. 数据向量化：另一种方法是将数据向量化，即将数据转换为大模型易于处理和查询的格式，然后快速地将其放入企业自己的向量数据库中。

这就是大模型在大数据架构中所处的位置和作用，大模型作为大数据架构的核心技术组件，在数据转换、预测分析和智能应用等方面发挥着不可替代的作用，是实现大数据价值的关键所在。

数据高速公路：Apache SeaTunnel& WhaleStudio

数据同步是大数据架构中的另一个关键环节。使用如Apache NiFi、Apache Spark、Sqoop等工具，可以实现数据在不同系统和数据库之间的实时和批量同步。这些工具支持跨云和混合云环境，能够处理来自各种数据源的数据，并将其同步到目标数据库或数据仓库中。但是因为依赖开源，它们的数据源支持力度非常有限。

Apache SeaTunnel：新一代实时多源数据同步工具，大数据的高速公路

有一个非常形象的比喻可以简单明了地概括Apache SeaTunnel的作用——大数据的高速公路。它可以把各种各样的数据源，如MySQL、RedShift、Kafka等数据，实时和批量数据同步至目标数据库。区别于Apache NiFi、Apache Spark，新一代实时多源数据同步工具Apache SeaTunnel目前已经可以支持上百种源数据库/目的地的数据同步与集成，并支持以跨云和混合云的方式同步数据，便于不同的用户进一步进行大数据和大模型训练。

Apache SeaTunnel 典型案例

目前，Apache SeaTunnel在全球已经有大量用户，其中一个典型用户是JP Morgan（摩根大通银行）。

摩根大通银行是一家全球知名的拥有超过 200,000 名员工的金融巨头，其中包括 30,000 多名数据专业人员（工程师、分析师、科学家和顾问），正在与复杂的遗留系统和新兴的数据环境作斗争。该机构在 10 多个不同的数据平台组成的迷宫中运营，需要一种强大、安全且高效的数据集成方法。

对摩根大通银行来说，最重要的挑战是通过复杂的隐私和访问控制对数据进行摄取和处理，这虽然对于数据保护至关重要，但通常会延迟数据集成过程。再加上该公司向AWS的过渡阶段（两年后仍在进行中），以及对Snowflake等现代数据库解决方案的实验，对灵活的数据集成解决方案的需求很迫切。

在追求敏捷性的过程中，摩根大通银行对比了若干流行的数据同步产品，比如Fivetran、Airbyte，但最终选择了支持Spark集群来实现最佳性能的替代方案——Apache SeaTunnel。

原因就在于SeaTunnel与其现有的Spark基础设施兼容，一个关键优势是Apache SeaTunnel与Java代码库的无缝集成，允许从摩根大通银行的主要编码环境直接触发数据迁移作业。摩根大通银行利用SeaTunnel从 Oracle、DB2、PostgreSQL、DynamoDB和 SFTP文件等源获取数据，在Spark集群上处理数据，并最终将其加载到S3（摩根大通银行的集中式数据存储库）中，随后集成到Snowflake和Amazon Athena进行高级分析。

Apache SeaTunnel的一个突出功能是能够显式地处理数据类型转换，确保不同系统之间的数据完整性，这是摩根大通银行银行多元化数据生态系统的重要组成部分。

为什么我们需要Apache SeaTunnel？

既然已经有Flink、Spark等各种流行的数据处理工具存在，为什么我们需要Apache SeaTunnel呢？和摩根大通银行一样，深入了解这个工具，你会发现这个问题并不难回答。

• Apache SeaTunnel支持开发版，目前支持130+ Connectors，商业版产品更是（WhaleTunnel）支持150+种数据库，这是其他产品所无法比拟的；

• SeaTunnel性能优势：比Airbyte快30倍，比DataX快30%；（性能报告可参考《最新性能对比报告:SeaTunnel 是 Airbyte 30 倍！》

• 易于部署：可以在3分钟内部署Apache SeaTunnel，支持在Spark/Flink/Zeta上运行。

使用方式简单

在使用方式上，Apache SeaTunnel也秉持着为广泛大数据从业者服务为宗旨，使用方式以简单易用为主要设计目标。

• 可以使用SQL-like代码创建同步作业。
• 支持Source Connector、Sink Connector和Transform操作。

想要更简单的方式？AWS Market Place上的WhaleStudio

如果创建代码来进行数据集成对我们来说有挑战，还有更加简单易用的方式可以选择。白鲸开源基于Apache DolphinScheduler和Apache SeaTunnel打造的商业产品WhaleStudio ，是分布式、云原生并带有强大可视化界面的DataOps系统，增加了商业客户所需的企业级特性，零基础用户也可以简单上手：

• 所见即所得的数据Mapping和处理
• 全可视化操作的调度和数据处理，无需代码处理
• 全面兼容AWS及多云、混合云架构
• 多团队协作和开发
• 性能卓越、超过150种数据源的连接，包括
  • AWS S3, Aurora, Redshift
  • SAP
  • Oracle, MySQL
  • Hudi, Iceberg

简单来说，WhaleStudio的使用流程和大模型集成可以简单概括为以下几点：

1. 数据源连接：首先，需要在WhaleStudio中配置数据源。这包括CSV文件、数据库、云存储服务等。用户可以通过拖放的方式将数据源组件添加到工作流中，并设置连接参数。
2. 数据转换：数据在传输过程中可能需要进行清洗和转换以适应目标系统。WhaleStudio提供了多种数据转换工具，包括数据过滤、字段映射、数据合并等。
3. 数据加载：转换后的数据需要加载到目标数据库或数据仓库中。WhaleStudio支持多种目标系统，包括关系型数据库、NoSQL数据库和云数据服务。
4. API集成：为了使数据能够被大模型理解，需要通过API将数据转换为特定的格式。WhaleStudio可以调用外部API，并将转换后的数据输出到大模型中。
5. 流程监控：用户可以实时监控数据流的状态，查看数据同步的进度和任何可能出现的错误。
6. 数据同步与更新
   1. 定时任务：WhaleStudio支持定时任务，允许用户设置在特定时间自动运行数据流，以确保数据的实时更新。
   2. 数据版本控制：通过版本控制，用户可以追踪数据流的变更历史，并在必要时回滚到之前的版本。

如何将大模型灌入公司内部数据，并“百科全书”化

如上文所述数据的“高速公路”有了，那么如何通过“高速公路”将数据放到大模型中并利用呢？

上图以一个示例展示了大模型如何将公司内部数据“百科全书”化的概略图，将MySQL数据库中的所有关于图书的文章，通过图形化的方式输入大模型中，即以向量的方式让大模型理解，并最终将输入的数据以语言的方式进行问答。下面以实战案例详细解说此流程。

实战案例：在AWS上利用WhaleStudio+大模型将图书馆检索从书名检索到语义检索

现有的图书搜索解决方案（例如公共图书馆使用的解决方案）十分依赖于关键词匹配，而不是对书名实际内容的语义理解。因此会导致搜索结果并不能很好地满足我们的需求，甚至与我们期待的结果大相径庭。这是因为仅仅依靠关键词匹配是不够的，因为它无法实现语义理解，也就无法理解搜索者真正的意图。

有更好的方法可以让我们更加准确、高效地进行图书搜索。通过使用特定的API，可以将图书数据转换为大模型能够理解的格式，从而实现语义级别的搜索和问答功能。这种方法不仅提高了搜索的准确性，还为企业提供了一种新的数据利用方式。

WhaleStudio是一个强大的数据集成和处理平台，它允许用户通过图形化界面来设计和实施数据流。WhaleStudio被用于将图书馆的图书数据集成到大模型中，以便进行更深层次的语义搜索和问答。

下面我们来演示一下如何使用WhaleStudio、Milvus和OpenAI进行相似度搜索，实现对整个书名的语义理解，从而让搜索结果更加精准。

准备工作

1. 在实验之前，我们需要去官网获取一个OpenAI的token，

2. 在AWS MarketPlace部署 WhaleStudio

3. 然后部署一个Milvus的实验环境(https://milvus.io/docs/install_standalone-docker.md）。

4. 我们还需要准备好将用于这个例子的数据，可以从这里下载，把它放到/tmp/milvus_test/book下（https://www.kaggle.com/datasets/jealousleopard/goodreadsbooks）

5. 配置WhaleStudio任务

建立项目→新建工作流定义→建立SeaTunel任务→copy脚本到任务里

2. 脚本代码

env {
  # You can set engine configuration here
  execution.parallelism = 1
  job.mode = "BATCH"
  checkpoint.interval = 5000
  #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
}

source {
  # This is a example source plugin **only for test and demonstrate the feature source plugin**
  LocalFile {
    schema {
      fields {
        bookID = string
        title_1 = string
        title_2 = string
      }
    }
    path = "/tmp/milvus_test/book"
    file_format_type = "csv"
  }
}

transform {
}

sink {
  Milvus {
    milvus_host = localhost
    milvus_port = 19530
    username = root
    password = Milvus
    collection_name = title_db
    openai_engine = text-embedding-ada-002
    openai_api_key = sk-xxxx
    embeddings_fields = title_2
  }
}

3. 点击运行

4. 简单数据预处理也可以利用可视化界面

5. 查询数据库，确认已经有数据

6. 使用如下代码通过语义搜索书名

import json
import random
import openai
import time
from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility

COLLECTION_NAME = 'title_db'  # Collection name
DIMENSION = 1536  # Embeddings size
COUNT = 100  # How many titles to embed and insert.
MILVUS_HOST = 'localhost'  # Milvus server URI
MILVUS_PORT = '19530'
OPENAI_ENGINE = 'text-embedding-ada-002'  # Which engine to use
openai.api_key = 'sk-******'  # Use your own Open AI API Key here

connections.connect(host=MILVUS_HOST, port=MILVUS_PORT)

collection = Collection(name=COLLECTION_NAME)

collection.load()


def embed(text):
    return openai.Embedding.create(
        input=text, 
        engine=OPENAI_ENGINE)["data"][0]["embedding"]

def search(text):
    # Search parameters for the index
    search_params={
        "metric_type": "L2"
    }

    results=collection.search(
        data=[embed(text)],  # Embeded search value
        anns_field="title_2",  # Search across embeddings
        param=search_params,
        limit=5,  # Limit to five results per search
        output_fields=['title_1']  # Include title field in result
    )

    ret=[]
    for hit in results[0]:
        row=[]
        row.extend([hit.id, hit.score, hit.entity.get('title_1')])  # Get the id, distance, and title for the results
        ret.append(row)
    return ret

search_terms=['self-improvement', 'landscape']

for x in search_terms:
    print('Search term:', x)
    for result in search(x):
        print(result)
    print()

7. 运行结果

结果： 如果我们按照之前的老方法关键词搜索，书名中必须包含自我提升、提升等关键词；但是提供大模型进行语义级别的理解，则可以检索到更加符合我们需求的书名。比如在上面的例子中，我们搜索的关键词为self-improvement（自我提升），展示的书名《关系之舞：既亲密又独立的相处艺术》、《尼各马可伦理学》等虽然不包含相关关键词，却很明显更加符合我们的要求。

结语

大数据和大模型为企业提供了前所未有的数据处理能力和洞察力。通过有效的数据架构设计、大模型集成、实时与批量数据处理以及数据同步，企业可以更好地利用其数据资源，提升运营效率，并在竞争激烈的市场中保持领先。

Apache SeaTunnel和WhaleStudio作为企业数据高速公路，帮助快速对接企业内部数据，实现数据的向量化和“百科全书化”。其中，WhaleStudio作为一个数据集成工具，为企业提供了一个简单、高效且功能强大的解决方案，让企业可以轻松地将数据同步到大模型中，实现更深层次的数据分析和应用，从而提升企业的数据处理能力和业务洞察力。

本文由 白鲸开源科技 提供发布支持！