连接器套件

使用这个包来建立一个翼型生产者连接器。

安装

pnpm add @useairfoil/connector-kit effect@^4

使用

• Connector， Resource， Fetch, Cursor 定义数据流。
• Manifest 定义用户配置。
• ConnectorApp 检查配置并运行HTTP服务器。
• Webhook 定义webhook路由。
• Ingestor 将批处理写入控制台或冰山。
• StateStore 保持摄入进度。

这里有一个小资源：

import { Connector, Cursor, Fetch, Iceberg, Resource } from "@useairfoil/connector-kit";
import { Effect, Schema } from "effect";

const PostSchema = Schema.Struct({
  id: Schema.String.pipe(Iceberg.field(1)),
  updatedAt: Schema.String.pipe(Iceberg.field(2)),
}).pipe(Iceberg.identifier("id"));

const Posts = Resource.entity({
  name: "posts",
  rowSchema: PostSchema,
  version: "updatedAt",

  check: Effect.void,
  backfill: Fetch.page({
    pageCursor: Cursor.number(),
    cutoff: Cursor.isoDateTime(),
    fetch: () =>
      Effect.succeed({
        rows: [],
        nextPageCursor: 1,
        hasMore: false,
      }),
  }),
});

const connector = Connector.define({
  name: "producer-example",
  resources: [Posts],
});

当一个提供程序日志覆盖多个资源（如事件API）时，给出 连接器是单一的。 changes 饲料代替 changes 在每个资源上。这个 Feed每运行一次读取日志，并根据资源名称返回行：

const connector = Connector.define({
  name: "producer-example",
  resources: [Posts, Comments],
  changes: Fetch.feed({
    resources: [Posts, Comments],
    cursor: Cursor.string(),
    interval: "5 minutes",
    fetch: ({ cursor }) =>
      readEvents(cursor).pipe(
        Effect.map((page) => ({
          rows: { posts: page.posts, comments: page.comments },
          cursor: page.next,
          // Runs again without waiting for the interval.
          hasMore: page.hasMore,
        })),
      ),
  }),
});

资源本身 changes 可以返回 hasMore 同样的方式，要劈一个长 跨运行读取。

提要中的每个资源都存储提要光标，因此状态显示它 资源。提要中的资源也不能定义自己的资源 changes.

机翼型存储引擎支持对冰山行向上操作。使用 Iceberg.identifier 注释标识符行（主键）的帮助器。这 注释必须总是在最后使用。

反转操作替换完整的行内容，因此更改提要 必须包括所有字段。如果webhook事件不包括所有字段，则 实现必须在将资源推送到引擎之前获取资源。

翼型存储引擎实现软删除。如果需要删除一行，只需推送即可 键，版本列（如果有的话）， _deleted = true 以及任何其他必需的列。 如果目标服务支持恢复记录，则连接器必须发送 _deleted = false 每一行。

运行

在本地测试时，您可以在内存中运行连接器：

• StateStore.layerMemory：在内存中存储连接器的状态。
• Ingestor.layerConsole：向控制台输出新批。

为了持久的跑步：

• RuntimeConfig.layerHosted()：从文件中加载配置。
• StateStore.layerSql()：在PostgreSQL中存储连接器的状态。
• Ingestor.layerArrowFlightConfig(connector)：向Apache Iceberg发布数据 使用翼型存储引擎和箭飞行。

内置HTTP服务器具有 /health， /status, /metrics 端点。

配置

这个 AIRFOIL_TABLE_BINDINGS 环境变量控制使用哪些资源 与目标表同步。必须至少定义一个资源。

AIRFOIL_TABLE_BINDINGS={"posts":{"namespace":["default"],"name":"posts"}}

在表名中只使用小写字母、数字和下划线。S3表 目录拒绝其他字符，例如 -.

演示数据

带有演示的连接器具有 demo 命令。它将虚构的行写入 绑定表，然后退出。它不需要提供者凭证，也不需要 PostgreSQL的状态。

producer-stripe demo --size 128MiB --seed 42

• --size 要写入的未压缩箭头大小，如 512KiB， 128MiB， 或者 1GiB。运行可以通过一个组块进行，大约为1 MiB 当前的连接器。
• --seed 选择数据。日期取决于跑步开始的时间，所以是两次 只有当它们同时启动并绑定时，才与相同的种子匹配 相同的表。
• 编写一次的设置表（如产品目录）不会增长。如果只有 这些是绑定的，运行写入它们一次，并在小于大小时停止。演示 对于有限的历史记录，例如每日报告，当它们用完时也要停止。
• 它需要 AIRFOIL_CONFIG_PATH， ENGINE_URI， AIRFOIL_CATALOG, AIRFOIL_TABLE_BINDINGS。一个空 {} 配置文件就足够了。

一个演示列出了它的 setup 和 group 表，并使用 Demo.define。从组索引构建键，因此它们是唯一的和其他的 行可以指向它们。一行省略的必填字段获取占位符值 从行模式中。每一行都要根据它的行模式进行检查 写的。