Pulumi DeltaStream 供应商

管理DeltaStream资源的普鲁米供应商.

概览

Pulumi 的 DeltaStream 供应商允许您使用基础设施作为代码来管理 DeltaStream 资源 。 DeltaStream是一流数据平台,可以进行实时分析与处理.

此提供者支持 :

• 数据库 - 用于计划和关系的逻辑容器
• 命名空间 - 数据库内的Schema命名空间
• 存储 - 外部数据储存(卡夫卡、克尼斯等)
• 对象 - 关系(结构/变化/表)
• 查询 - 连续输入查询(单一汇)
• 应用 - 有虚拟关系的多汇流应用程序

安装

类型脚本/ Java 脚本( 节点. js)

npm install @deltastream/pulumi-deltastream

走开

go get github.com/deltastreaminc/pulumi-deltastream/sdk/go/deltastream

配置

提供方配置镜像为基础的DeltaStream SQL驱动程序所使用的环境变量.

示例(环境变量):

export DELTASTREAM_SERVER="https://api.deltastream.io/v2"
export DELTASTREAM_API_KEY="<your key>"

或者Pulumi config(建议保密):

pulumi config set deltastream:server https://api.deltastream.io/v2
pulumi config set --secret deltastream:apiKey <your key>

示例使用

类型脚本

import * as pulumi from "@pulumi/pulumi";
import * as deltastream from "@deltastream/pulumi-deltastream";

const provider = new deltastream.Provider("deltastream", {
    server: process.env.DELTASTREAM_SERVER!,
    apiKey: process.env.DELTASTREAM_API_KEY!,
});

// Create a database and a namespace
const db = new deltastream.Database("example_db", { name: "example_db" }, { provider });
const ns = new deltastream.Namespace("example_ns", { database: db.name, name: "example_ns" }, { provider });

// Invoke lookups
const dbInfo = db.name.apply(n => deltastream.getDatabase({ name: n }, { provider }));
const namespaces = db.name.apply(d => deltastream.getNamespaces({ database: d }, { provider }));

export const dbCreatedAt = db.createdAt;
export const nsCreatedAt = ns.createdAt;
export const namespaceCount = namespaces.apply(r => r.namespaces.length);

走开

package main

import (
    ds "github.com/deltastreaminc/pulumi-deltastream/sdk/go/pulumi-deltastream"
    "github.com/pulumi/pulumi/sdk/v3/go/pulumi"
    "os"
)

func main() {
    pulumi.Run(func(ctx *pulumi.Context) error {
        prov, err := ds.NewProvider(ctx, "deltastream", &ds.ProviderArgs{
            Server: pulumi.String(os.Getenv("DELTASTREAM_SERVER")),
            ApiKey: pulumi.String(os.Getenv("DELTASTREAM_API_KEY")),
        })
        if err != nil { return err }

        db, err := ds.NewDatabase(ctx, "db", &ds.DatabaseArgs{ Name: pulumi.String("example_db") }, pulumi.Provider(prov))
        if err != nil { return err }

        ns, err := ds.NewNamespace(ctx, "ns", &ds.NamespaceArgs{ Database: db.Name, Name: pulumi.String("example_ns") }, pulumi.Provider(prov))
        if err != nil { return err }

        ctx.Export("dbCreatedAt", db.CreatedAt)
        ctx.Export("nsCreatedAt", ns.CreatedAt)
        return nil
    })
}

流出查询

DeltaStream 支持两种流式查询资源:

1. 查询资源

这个 Query 资源是用于简单的 INSERT INT 查询, 并使用单个汇 :

类型脚本 :

const query = new deltastream.Query("myQuery", {
    sourceRelationFqns: [source.fqn],
    sinkRelationFqn: sink.fqn,  // Single sink
    sql: pulumi.interpolate`INSERT INTO ${sink.fqn} SELECT * FROM ${source.fqn};`,
}, { provider });

开始:

q, err := ds.NewQuery(ctx, "myQuery", &ds.QueryArgs{
    SourceRelationFqns: pulumi.StringArray{ source.Fqn },
    SinkRelationFqn:    sink.Fqn,  // Single sink
    Sql: pulumi.Sprintf("INSERT INTO %s SELECT * FROM %s;", sink.Fqn, source.Fqn),
}, pulumi.Provider(prov))

2. 应用程序资源(多链接)

这个 Application 资源是用于复杂的流化应用,包括:

• 多语种服务组织 目标不同的汇
• 虚拟中间关系 (CREATE VIRTALSTREAM/变相)
• 复杂的处理逻辑 包含连接、窗口和集合

关键特性 :

• 虚拟关系是应用的内在关系,不要创造出卡夫卡主题
• 只需宣布物理源和汇为附属物
• 虚拟关系自动被排除在依赖跟踪之外

类型脚本 :

// 1. Create physical source and sink relations OUTSIDE the application
const pageviews = new deltastream.DeltaStreamObject("pageviews", {
    database: db.name,
    namespace: "public",
    store: kafkaStore.name,
    sql: pulumi.interpolate`CREATE STREAM pageviews (...) WITH ('topic'='pageviews');`,
}, { provider });

const visitFreq = new deltastream.DeltaStreamObject("visitFreq", {
    database: db.name,
    namespace: "public",
    store: kafkaStore.name,
    sql: pulumi.interpolate`CREATE CHANGELOG visit_freq (...) WITH ('topic'='visit_freq');`,
}, { provider });

// 2. Create APPLICATION with virtual relations and INSERT INTO
const app = new deltastream.Application("myApp", {
    sourceRelationFqns: [pageviews.fqn],      // Physical sources only
    sinkRelationFqns: [visitFreq.fqn],        // Physical sinks only
    sql: pulumi.interpolate`
        BEGIN APPLICATION my_app
            -- Virtual relation (no Kafka topic, internal only)
            CREATE VIRTUAL STREAM virtual.public.filtered AS
                SELECT * FROM ${pageviews.fqn}
                WHERE userid IS NOT NULL;
            
            -- Insert into physical sink
            INSERT INTO ${visitFreq.fqn}
                SELECT window_start, window_end, userid, count(*) as cnt
                FROM TUMBLE(virtual.public.filtered, SIZE 30 SECONDS)
                GROUP BY window_start, window_end, userid;
        END APPLICATION;
    `,
}, { provider, dependsOn: [pageviews, visitFreq] });

开始:

// 1. Create physical source and sink relations OUTSIDE the application
pageviews, err := ds.NewDeltaStreamObject(ctx, "pageviews", &ds.DeltaStreamObjectArgs{
    Database:  db.Name,
    Namespace: pulumi.String("public"),
    Store:     kafkaStore.Name,
    Sql: pulumi.Sprintf("CREATE STREAM pageviews (...) WITH ('topic'='pageviews');"),
}, pulumi.Provider(prov))

visitFreq, err := ds.NewDeltaStreamObject(ctx, "visitFreq", &ds.DeltaStreamObjectArgs{
    Database:  db.Name,
    Namespace: pulumi.String("public"),
    Store:     kafkaStore.Name,
    Sql: pulumi.Sprintf("CREATE CHANGELOG visit_freq (...) WITH ('topic'='visit_freq');"),
}, pulumi.Provider(prov))

// 2. Create APPLICATION with virtual relations and INSERT INTO
app, err := ds.NewApplication(ctx, "myApp", &ds.ApplicationArgs{
    SourceRelationFqns: pulumi.StringArray{ pageviews.Fqn },  // Physical sources only
    SinkRelationFqns:   pulumi.StringArray{ visitFreq.Fqn },  // Physical sinks only
    Sql: pulumi.Sprintf(`
        BEGIN APPLICATION my_app
            -- Virtual relation (no Kafka topic, internal only)
            CREATE VIRTUAL STREAM virtual.public.filtered AS
                SELECT * FROM %s
                WHERE userid IS NOT NULL;
            
            -- Insert into physical sink
            INSERT INTO %s
                SELECT window_start, window_end, userid, count(*) as cnt
                FROM TUMBLE(virtual.public.filtered, SIZE 30 SECONDS)
                GROUP BY window_start, window_end, userid;
        END APPLICATION;
    `, pageviews.Fqn, visitFreq.Fqn),
}, pulumi.Provider(prov))

重要注释 :

• 虚拟关系不得包括在 sourceRelationFqns 或者说 sinkRelationFqns
• 仅应宣布实际关系(与实际的Kafka专题)为依存关系
• 提供者验证虚拟关系没有被错误地声明为依赖性

完整的示例见:

• 实例/应用程序 - 满了 以多个汇为例
• 实例/应用程序 - 有多个汇的完整类型脚本示例

发展

资源

提供者包括下列资源:

查询 vs 应用程序: 每个应用程序何时使用

使用 Query 当:

• 你有一个简单的Insert进入... SELECT查询
• 单源和单汇
• 不需要虚拟中间关系
• 直向转变

使用 Application 当:

• 以不同汇为目标的多种 " INSERT O " 声明
• 需要虚拟中间关系(CREATE VIRTUAL STREAM/CHANGELOG)
• 组合、窗口和聚合的复杂处理
• 想要将相关查询组织成一个单一的逻辑单位

查询资源字段注释

查询资源支持后向兼容的后向字段名称:

• sinkRelationFqn (弦声, 已贬值:用于与单沉INSERT INTO查询的后向兼容性
• sinkRelationFqns (弦声), 已贬值:用于遗留应用查询

建议: 对于新代码 :

• 使用 Query 用于简单的 INSERT 查询的资源
• 使用 Application 多汇流应用程序的资源

为什么改变? 新的 Application 资源提供:

• 类型安全应用特定验证
• 查询和应用程序之间的明确区分
• 开发者在专门领域的更好经验
• 虚拟关系依赖关系的自动验证

移徙: 现有的查询资源仍在继续工作。 见 变动情况 移民指南.

先决条件

• 前进1.24+
• Pulumi CLI( 单独安装; CI 通过 pulumi/actions(中文(简体) )
• 节点.js(仅在工作/验证节点SDK时)
• 可选: Yarn 为更快的节点构建

Pulumi CLI不再由Makefile自动安装(用于供应链安全). 手动安装或通过您的软件包管理器安装 :

curl -fsSL https://get.pulumi.com | sh   # (optional quick start; prefer package managers or pinned action in CI)
# or on macOS
brew install pulumi

该寄存器使用环境变量在 CI 中标出一个 CLI 版本 PULUMI_CLI_VERSION (见 .github/workflows/ci.yml) (中文(简体) ). 如果遇到计划生成差异,请检查本地 pulumi version 并符合这一价值.

建设提供者

make build

运行测试

make test

生成 SDK

make build_sdks

可用 Make 目标

• build - 构建提供者二进制
• schema - 生成提供者计划(需要Pulumi CLI在场)
• generate - 生成所有语言 SDKs(nodejs, go, python)
• build_sdks - 构建全部 SDK( 系统编译)
• install_sdks - 在当地安装/链接SDK以促进发展
• test 运行实例集成测试( 需要已建的提供者 & Pulumi CLI)
• clean - 删除已建的已生成的 SDK( E)
• help - 显示帮助消息

目标故意避免隐含地下载工具;确保Pulumi CLI在运行计划或生成目标之前在您的 PATH 上.

项目结构

.
├── cmd/
│   └── pulumi-resource-deltastream/    # Provider binary entry point
├── provider/                           # Provider implementation
├── sdk/                               # Generated SDKs
│   ├── go/
│   └── nodejs/
├── examples/                          # Example programs
│   ├── application-go/               # Go APPLICATION example (multi-sink)
│   ├── application-ts/               # TypeScript APPLICATION example (multi-sink)
│   ├── query-go/                     # Go Query example (single sink)
│   └── query-ts/                     # TypeScript Query example (single sink)
├── schema.json                        # Provider schema
├── Makefile                          # Build automation
└── README.md                         # This file

许可证

这个项目根据Apache许可证2.0获得许可——见 自由主义 详细文件.

支助

支助和问题:

• GitHub 问题
• DeltaStream 文档
• Pulumi 文档