数据接入

数据接入把客户系统的数据送进 Semantic。源系统只读，平台不写回源系统。源系统里的一张表只同步一次，存成数据集；之后的计算、对象和查询都读数据集，不再回到源系统。这一页讲先找已有的数据，两条接入路线怎么选，每条路线怎么做，以及数据怎么接到 Object Type 上。

先找已有的数据

这一节让你在接新数据之前，先找到已经同步的数据。组织接进来的表，多半已经同步成了数据集。

$ aidc semantic datasets --search 物料
cell-demo · 2 个数据集（搜「物料」）
materials                      302,125 行 · 6.0 MB · 2026-10-07 21:47Z · 加工 物料 ← raw-erp-mara、raw-erp-makt
  列 materialId, materialNumber, materialType, materialGroup, baseUnit, chineseName
  ri.foundry.main.dataset.…
raw-erp-mara                   302,125 行 · 1.6 MB · 2026-10-07 19:05Z · 表导入 物料一般主数据 ← erp-mirror：MARA
  ri.foundry.main.dataset.…
$ 

• --search 按名字、来源表、中文名、列名和对象类型搜。每个词都要对上。
• 看前几行：aidc semantic connectivity dataset <RID> --rows 20。
• 要业务数字（联表、聚合、口径），用 加工 从数据集算。要逐条看、改、链接的业务实体，把数据集 接到 Object Type。
• 找不到，再按下一节接入。不要直接查源系统，也不要把同一张表再推一份进数据流。

两条路线怎么选

这一节让你选出一条接入路线。源系统里的表，用 Data Connection。只有要秒级推送的小信号，才用数据流。

两条路线的数据最后都要绑定到 Object Type 上，见 接到 Object Type。

数据流

这一节让你把源系统的变化推进 Semantic。发布端只发变化。平台按序记账。订阅端几秒内收到更新。

声明数据流

先声明数据流：名字、主键和字段白名单。发布的数据只能包含白名单里的字段。

$ aidc semantic streams create orders --title "在手订单" --key order_no \
>   --field order_no:string:合同号 --field qty:number:数量 --field due:datetime:交期
新建数据流 cell-demo/orders（3 个字段，主键 order_no）
下一步：aidc semantic streams key cell-demo/orders --label "装在哪"
$ 

• 字段类型只有 string、number、boolean、datetime 和 json。
• 同名再声明一次是更新。字段白名单随之改变。
• 证件号、银行卡号、密码这类字段，平台拒收。
• --retention 设保留的事件条数。缺省保留最近 1000 条。

签发布 Key

发布 Key 只能往这一条流发布。Key 只显示一次。--label 会进审计记录，写清楚装在哪。

$ aidc semantic streams key orders --label "工厂服务器 · 订单发布端"
发布 Key（只显示这一次，只能发布 cell-demo/orders）：
  aidc-pk-…
装到发布端：AIDC_PUBLISH_KEY=<上面这把> aidc semantic streams pipe cell-demo/orders -- <适配器命令>
$ 

把 Key 放进发布端的环境变量 AIDC_PUBLISH_KEY。不要写进命令行参数。

常驻运行发布端

这一节讲怎么让发布端在源系统旁边常驻运行。用 systemd 等服务管理器启动它，不要用定时任务。

$ AIDC_PUBLISH_KEY=aidc-pk-… aidc semantic streams pipe cell-demo/orders -- python3 orders_adapter.py
发布端启动：cell-demo/orders（主键 order_no）← python3 orders_adapter.py
2026-10-08T14:03:12 发布 → seq 128（2 条，3 行，182ms）
$ 

命令输出里的时间是 UTC（国际标准时间）。北京时间要加 8 小时。

写适配器

适配器是一个常驻程序，任何语言都可以写。它用数据源自己的方式发现变化。它每行向标准输出打印一个 JSON 对象。

适配器每一轮至少打一行 status。

# orders_adapter.py：常驻进程。先用便宜的探针看有没有变化，变了才取全表。
import json, time

last = None
while True:
    try:
        version = probe_version()            # 换成探针，如 Change Tracking 的版本号
        if version != last:
            rows = read_open_orders()        # 换成读数据源的代码，只读
            print(json.dumps({"rows": rows}, default=str), flush=True)
            last = version
        print(json.dumps({"status": "ok"}), flush=True)
    except Exception as err:
        print(json.dumps({"status": "source_unreachable", "detail": str(err)[:200]}), flush=True)
    time.sleep(5)

发布端负责下面这些事：

• 第一次发布全量快照。之后按主键比较，只发变化。
• 发布失败时，按 1、2、4 秒的间隔重试，最长间隔 30 秒。同一批事件带同一组 ID，服务端不会收两次。
• 每 30 秒发一次心跳。订阅端凭心跳显示「在线」。
• 距上次全量快照满 6 小时后，发布端在下一次收到 rows 时发布全量快照。需要定期校准时，适配器也要定期输出全表。
• 适配器超过 15 分钟没有输出，状态报「部分异常」。
• 适配器退出后，按退避时间重启，最长间隔 60 秒。同时报「连不上数据源」，附上标准错误输出的最后三行。
• 服务端拒收一批（字段不在白名单里）时，丢掉这一批，状态报「部分异常」。
• 发布 Key 被吊销，或流被删除（返回 401、403 或 404）时，发布端停止，退出码是 1。
• 按 Ctrl-C 或 SIGTERM 停止时，发布端报「已停止」。它最多等 10 秒，把积压的事件发完，退出码是 0。

平台这一侧也去掉没变的数据。快照和当前状态完全相同，或 patch 一行都没改时，平台丢掉这个事件。它不占序号，不推给订阅端。patch 只记下真正改了的行。

怎么发现变化

这一节让你为每种源系统选一种发现变化的办法。能推送的源用推送。不能推送的源，在源系统旁边做一个增量探针。一个探针服务所有订阅者。不要为每个看板建一个定时任务。

在应用里订阅

这一节让应用实时收到数据流的变化。应用订阅前，要在清单的 streams 里声明这条流。

"streams": ["orders"]

import { semantic } from "/developer/sdk/v1/aidc.js";

const orders = semantic.stream("orders");
const sub = orders.subscribe({
  onUpdate({ rows, upserted }) {
    render(rows);                  // 合并后的当前全部行，按主键排序
    flash(upserted);               // 这次新增或变化的行
  },
  onNotice({ data }) {
    toast(data.message);           // 发布端发来的通知
  },
  onStatus({ connection, publisher }) {
    badge(connection, publisher);
  },
});

const { rows, seq } = await orders.get();   // 一次性读当前状态
// 不再需要时：sub.close()

• 第一次连上，先收到全量。之后每次变化收到增量。断线后按序号续传。落后超过 500 条、事件断档或补齐范围内含快照时，订阅端直接读取最新状态。
• connection 的取值是 connecting、live、reconnecting、paused 或 closed。页面隐藏超过 1 分钟，连接断开，状态是 paused。切回页面后自动续上。
• publisher.live 为 false 时，界面提示「数据可能不是最新」。publisher.status 为 source_unreachable 时，界面提示「连不上数据源」。
• 连接的时长计入应用的计算分钟。计算分钟用完后，订阅不再重连。

接到 Object Type

这一节让你把数据集或数据流绑定到 Object Type。绑定写在 Object Type 定义的 datasources 里。改绑定，就是改本体。

对象只建业务粒度：一个真实的实体一个 Object Type。原始表不要一对一建成对象；先用加工算出业务粒度的数据集，再接上它。

"datasources": [
  { "type": "dataset", "dataset": "customer-sales-monthly", "propertyMapping": { "rowId": "row_id", "customerId": "customer_id", "salesAmount": "sales_amount" } }
]

$ aidc semantic datasource set sales.CustomerSalesMonthly --dataset customer-sales-monthly --map rowId=row_id --map customerId=customer_id --map salesAmount=sales_amount
接上数据源 sales.CustomerSalesMonthly ← 数据集 customer-sales-monthly（写进了类型定义）；首次构建 ri.foundry.main.build.…
$ 

• 对象跟着数据集的事务走。SNAPSHOT 里没有的行，对象标「源头已消失」。APPEND 只增改。
• Object Type 原来接的是数据流，改成数据集时加 --replace。原来的行数据源从定义里去掉，已有的对象保留。
• 不加 --replace 时，数据集不能映射别的数据源已经映射的属性（主键除外），返回 409 datasource_conflict。

数据流的绑定写法：

"datasources": [
  { "type": "stream", "stream": "mes-reports", "mode": "upsert", "propertyMapping": { "workOrderNo": "wo_no", "qtyDone": "qty_done" } }
]

$ aidc semantic datasource set production.work_order --stream mes-reports --map workOrderNo=wo_no --map qtyDone=qty_done --mode upsert
接上数据源 production.work_order ← mes-reports（写进了类型定义），首次同步：变化 12 个
$ aidc semantic datasource list
● production.work_order        ← 数据流 mes-reports  upsert  同步到 seq 418
$ 

• propertyMapping 左边是属性的 API name，用 camelCase。右边是流里的字段名，照源系统写。
• 一个 Object Type 恰好有一个主键属性。主键必须映射。
• 只在语义层维护的属性（writeback）和派生属性不映射。
• 同一条流在一个 Object Type 里只出现一次。
• 一个 Object Type 可以有多个绑定。每个绑定都要映射主键。两个绑定不要映射同一个属性。有多个绑定的 Object Type 不能配对象或属性的安全策略。
• mode 可以是 upsert（缺省）或 mirror。upsert 按主键更新对象，这一批里没出现的对象不消失。mirror 把流的全量当成全部对象。两者对对象的影响见 数据管道。
• 重新定义时没写 datasources，平台沿用现有的绑定，并返回一条警告。要去掉全部绑定，写 "datasources": []。
• 保存定义后，平台先全量同步一次。之后只同步变化。

aidc semantic datasource sync 同步平台数据源。立即重同步数据流绑定，用 aidc semantic datasource resync <类型> --stream <流名>。remove、resync 的用法见 参考 · 数据接入。Object Type 本身怎么定义，见 本体。

Data Connection

这一节让你用只读账号，由平台直接读取数据库。平台每次执行一条只读的 SELECT。数据库在客户内网时，先装一个 agent proxy。

装 agent proxy

agent proxy 装在客户网络里的一台机器上。它只向外连接平台，不开端口，只转发数据。查询仍在平台上执行。数据库能被平台直连时，不需要它。

1. 登记 agent，拿只显示一次的凭证。

$ aidc semantic connectivity agent register --ontology cell-demo --api-name site-box --display-name "工厂机房"
已登记：工厂机房（agentProxy）
  ri.…
凭证（只显示这一次）：aidc-ag-…
在客户网络里的机器上：
  curl -fsSLO https://www.ai-dc.ai/developer/agent-proxy/install-agent-proxy.sh
  sudo sh install-agent-proxy.sh --api https://www.ai-dc.ai --agent <agent RID>   # 粘贴上面的凭证（不回显）
  systemctl status aidc-agent-proxy
$ 

2. 在那台机器上，用 root 运行安装脚本。<agent RID> 是上一步输出的 ri.…。

curl -fsSLO https://www.ai-dc.ai/developer/agent-proxy/install-agent-proxy.sh
sudo sh install-agent-proxy.sh --api https://www.ai-dc.ai --agent <agent RID>

• 脚本在终端里询问凭证，输入时不回显。凭证不进命令行，也不进 shell 历史。
• 脚本把 agent proxy 装成 systemd 服务 aidc-agent-proxy。服务开机自启，挂了自动拉起。
• 缺省运行用户是新建的系统用户 aidc-agent。用 --user 指定别的用户。
• 安装需要 root 权限，和 Python 3.8 以上。

3. 查看 agent 的状态。状态是 online，表示控制通道已连上平台。

aidc semantic connectivity agents --ontology cell-demo

建出口策略

出口策略规定平台能连接的目的地：主机、端口，以及经哪个 agent。平台只连策略允许的地址。策略建好后不能修改，只能吊销。

aidc semantic connectivity egress create --ontology cell-demo --file erp-via-agent.json

• 策略文件的字段照 API 请求体写。字段见 参考 · 数据接入。
• 同名、同目的地的策略再建一次，返回已有的那条。同名、目的地不同，返回 409。
• 直连的目的地不能是内网地址、本机地址或云元数据地址。

建 Connection

Connection 记录一个数据库的地址、只读账号和出口策略。口令写一次，平台加密保存，之后不再回显。

aidc semantic connectivity connection --ontology cell-demo --file erp-connection.json
aidc semantic connectivity config <connectionRid>      # 看配置。口令只显示名字
printf '%s' "$PASSWORD" | aidc semantic connectivity secret <connectionRid> <秘密名>   # 换口令，从标准输入读

• 平台提供的 JDBC 驱动是 PostgreSQL 和 SQL Server。自带的驱动不收。
• 账号只给只读权限。平台的查询只读。

将下面的 JSON 保存为 erp-connection.json。替换主机、端口、数据库名、只读账号和占位值。

{
  "apiName": "orders-db",
  "displayName": "订单数据库",
  "configuration": {
    "type": "jdbc",
    "url": "jdbc:postgresql://db.example.com:5432/orders",
    "driverClass": "org.postgresql.Driver",
    "credentials": {
      "username": "reader",
      "password": {
        "type": "asPlaintextValue",
        "value": "<只读账号口令>"
      }
    }
  },
  "worker": {
    "type": "foundryWorker",
    "networkEgressPolicyRids": ["<出口策略 RID>"]
  }
}

先探索源

这一节让你在建导入之前，看清源里有哪些表。探索只读，不改源。

$ aidc semantic connectivity explore <connectionRid> --search order
dbo.orders                                       TABLE
dbo.order_lines                                  TABLE
$ aidc semantic connectivity explore <connectionRid> --table dbo.orders
dbo.orders（TABLE）
  order_no                         nvarchar(32)             非空 · PK
  customer_id                      int                      非空
  status                           nvarchar(16)             可空
  created_at                       datetime2                非空
  外键 customer_id → dbo.customers(id)
$ 

• 列出的表最多 2,000 张。用 --search 缩小范围。
• 样本行只在直连的源上可用。加 --table dbo.orders --preview --rows 5 取样本。最多取 50 行。平台不保存样本。经 agent proxy 的源只能看表结构。
• 探索只给本组织的 developer。

建 TableImport 并执行

一个 TableImport 是一条只读的 SELECT，结果写进一个数据集。每张表建一个 TableImport。导入方式有两种。SNAPSHOT 每次读到的是全部结果，可以带时间窗口。APPEND 用增量列，只读上次之后改过的行。

aidc semantic connectivity import <connectionRid> --file <TableImport.json>
aidc semantic connectivity imports <connectionRid>

$ aidc semantic connectivity execute <connectionRid> <tableImportRid> --wait
succeeded  读 1200 · 变 35 · 消失 0 · 对象变化 35 · 2310 ms
  ri.…
$ 

• --wait 最多等待 60 秒。返回时仍可能是 queued 或 running。不加 --wait，命令返回当前构建状态、统计信息和 RID。用 aidc semantic connectivity build <buildRid> 继续查状态。
• --full 强制全量重建。
• 同一个 TableImport 同时只有一个在途的执行。已经在跑时，命令返回那一次。
• 执行失败时，退出码是 8。
• 数据集怎么记账（事务、两种导入方式的差别、每周对账），见 数据管道。

将下面的 APPEND 示例保存为 TableImport.json。按源表替换查询、增量列和初值。

{
  "apiName": "orders",
  "displayName": "订单",
  "importMode": "APPEND",
  "config": {
    "type": "postgreSqlImportConfig",
    "query": "SELECT order_no, qty, updated_at FROM public.orders WHERE updated_at > ?::timestamp ORDER BY updated_at",
    "initialIncrementalState": {
      "type": "timestampColumnInitialIncrementalState",
      "columnName": "updated_at",
      "currentValue": "2026-01-01T00:00:00Z"
    }
  }
}

全量导入用下面的 SNAPSHOT 示例。按源表替换查询。

{
  "apiName": "orders",
  "displayName": "订单",
  "importMode": "SNAPSHOT",
  "config": {
    "type": "postgreSqlImportConfig",
    "query": "SELECT order_no, qty, updated_at FROM public.orders ORDER BY updated_at"
  }
}

挂 webhook（REST 数据源）

REST 数据源的 Connection 上可以挂 webhook。webhook 是一组对外部系统的 HTTP 请求。webhook 不存凭证，凭证在 Connection 上。Action 用 webhook 的方式有两种：

• writeback：校验通过后、规则执行前调用。外部系统拒绝或超时，整个 Action 不生效。
• side effect：提交之后调用。失败不回滚，也不报给执行人。

aidc semantic connectivity webhooks --ontology cell-demo
aidc semantic connectivity webhook test <webhookRid> --inputs '{"content":"你好"}' --dry-run   # 只渲染请求，秘密打码

Action 怎么配置 webhook，见 本体。

命令行

本页讲的主要命令如下。

声明数据流、订阅、登记 agent、建出口策略、建导入和执行的全部参数，见 参考 · 数据接入。

API

数据流的端点。{命名空间} 是组织的命名空间，如 cell-demo。

限制

常见错误

下一步

• 数据管道：数据集、加工、新鲜度与计算账。
• 文件：Media sets 与 Space：图片、文档、站点与分享。
• 对象：从表到对象，把数据接成对象。
• 本体：定义 Object Type、链接和 Action。
• 参考 · 数据接入：全部参数、输出和退出码。