developing-datacloud-code-extension

developing-datacloud-code-extension 技能

概述

此技能提供完整的工作流，用于开发、测试和部署自定义 Python 代码扩展到 Salesforce Data Cloud。代码扩展允许你编写 Python 转换逻辑，以从 Data Lake 对象（DLO）和数据模型对象（DMO）读取数据并写入数据。

适用场景

• 用户想要创建新的代码扩展项目
• 用户需要在本地测试代码扩展
• 用户想要扫描代码以查找所需权限
• 用户需要将代码扩展部署到 Data Cloud
• 用户正在使用 Data Cloud 转换
• 用户想要以编程方式读取/写入 DLO 或 DMO 数据

前置条件检查

执行任何代码扩展命令之前，请验证前置条件：

1. 已安装插件的 SF CLI

   sf plugins --core | grep data-code-extension
   

   如果尚未安装：

   sf plugins install @salesforce/plugin-data-codeextension
   

2. Python 3.11

   python --version  # Should show 3.11.x
   

3. Data Cloud Custom Code SDK

   pip list | grep salesforce-data-customcode
   

   如果尚未安装：

   pip install salesforce-data-customcode
   

4. Docker 正在运行（仅部署时需要）

   docker ps
   

5. 已认证的 org

   sf org display --target-org <org_alias> --json
   

技能工作流

阶段 1：初始化项目

使用脚手架创建新的代码扩展项目。

命令：

对于基于脚本的代码扩展（批量转换）：

sf data-code-extension script init --package-dir <directory>

对于基于函数的代码扩展（实时转换）：

sf data-code-extension function init --package-dir <directory>

必需选项：

• --package-dir, -p - 将创建软件包的目录路径

创建的内容：

my-transform/              # Project root
├── payload/               # CRITICAL: This is what --package-dir must point to for deploy
│   ├── entrypoint.py      # Main transformation code
│   └── config.json        # Code extension configuration
├── requirements.txt       # Python dependencies
└── README.md

工作流中的目录上下文

**重要：**了解目录结构对于成功部署至关重要。

命令及其目录要求：

命令	运行目录	路径/文件参数
init	父目录	<project-name> 或 .
scan	项目根目录	./payload/entrypoint.py
run	项目根目录	./payload/entrypoint.py
deploy	项目根目录	--package-dir ./payload（必需）

关键：deploy 命令中的 --package-dir 参数必须指向 payload 目录，而不是项目根目录。

阶段 2：开发转换逻辑

编辑 payload/entrypoint.py，编写转换逻辑。

脚本示例（批量）：

from datacustomcode import Client

client = Client()

# Read from DLO
df = client.read_dlo('Employee__dll')

# Transform data (uppercase position field)
df['position_upper'] = df['position'].str.upper()

# Write to output DLO
client.write_to_dlo('Employee_Upper__dll', df, 'overwrite')

函数示例（实时）：

from datacustomcode import FunctionClient

def transform(event, context):
    client = FunctionClient(context)
    input_data = event['data']
    output = {
        'name': input_data['name'].upper(),
        'status': 'processed'
    }
    return output

常见操作：

• client.read_dlo('DLO_Name__dll') - 从 DLO 读取
• client.read_dmo('DMO_Name') - 从 DMO 读取
• client.write_to_dlo('DLO_Name__dll', df, 'overwrite') - 写入 DLO
• client.write_to_dmo('DMO_Name', df, 'upsert') - 写入 DMO

阶段 3：扫描权限

扫描入口文件以检测所需权限并生成 config.json。

命令：

sf data-code-extension script scan --entrypoint ./payload/entrypoint.py

检测内容：

• DLO/DMO 的读取权限
• DLO/DMO 的写入权限
• Python 软件包依赖项
• 更新 config.json 和 requirements.txt

阶段 4：验证 DLO 架构（测试前检查）

关键：在本地运行测试之前，请验证代码中使用的所有 DLO 均已存在，并且包含预期字段。

步骤 4a：从 config.json 中提取 DLO

扫描完成后，检查生成的 config.json，确定所有 DLO：

cat payload/config.json

步骤 4b：验证每个 DLO 的架构

使用 getting-datacloud-schema 技能，确认 DLO 存在并检查字段名称。

对于代码中引用的每个 DLO：

1. 确认 DLO 存在：

   python3 scripts/get_dlo_schema.py <org_alias> <dlo_name>
   

2. 确认字段名称匹配——将 entrypoint.py 中使用的字段与 DLO 架构进行比较。

3. 检查所有 DLO：

   • 验证 read 权限中的所有 DLO
   • 验证 write 权限中的所有 DLO
   • 检查字段名称是否完全匹配（区分大小写）
   • 确认数据类型与操作兼容

步骤 4c：验证清单

继续运行之前，请确保：

• config.json 中的所有 DLO 在目标 org 中均存在
• 代码中使用的所有字段名称均存在于 DLO 架构中
• 字段数据类型与转换逻辑匹配
• 已正确识别主键字段
• 写入目标 DLO 已创建且可访问

阶段 5：本地测试

验证 DLO 架构后，在本地针对你的 Data Cloud 组织运行代码扩展。

命令：

sf data-code-extension script run --entrypoint <entrypoint_file> --target-org <org_alias> [options]

选项：

• --target-org, -o - SF CLI 组织别名（必填）
• --config-file, -c - 自定义配置文件路径

如果遇到错误：

• 重新验证 DLO 架构
• 检查字段名称是否完全匹配
• 验证数据类型是否兼容
• 查看错误消息，排查字段/DLO 问题

阶段 6：部署到 Data Cloud

将代码扩展部署到 Data Cloud，以便按计划或按需执行。

重要：必须指定 --package-dir ./payload，以指向通过 init 创建的 payload 目录。

命令：

sf data-code-extension script deploy --target-org <org_alias> --name <name> --package-dir ./payload --package-version <version> --description <description> [options]

必需选项：

• --target-org, -o - SF CLI 组织别名
• --name, -n - 代码扩展部署名称
• --package-dir - payload 目录的路径（必填 - 从项目根目录运行时必须为 ./payload）
• --package-version - 版本字符串（默认值：0.0.1）
• --description - 代码扩展描述

可选选项：

• --cpu-size - CPU 大小：CPU_L、CPU_XL、CPU_2XL（默认值）、CPU_4XL
• --function-invoke-opt - 函数调用选项（适用于函数类型）
• --network - Docker 网络（默认值：default）

部署后：

• 在 Salesforce UI 中进入 Data Cloud
• 前往 Data Transforms 部分
• 按名称找到你的部署
• 点击“立即运行”以执行
• 设置定期执行计划

错误处理

常见问题及解决方案

错误	解决方案
command data-code-extension not found	sf plugins install @salesforce/plugin-data-codeextension
datacustomcode CLI not found	pip install salesforce-data-customcode
Python version mismatch	使用 pyenv：pyenv install 3.11.0 && pyenv local 3.11.0
Cannot connect to Docker daemon	启动 Docker Desktop
No org found for alias	sf org login web --alias <org_alias>
config.json not found	sf data-code-extension script scan --entrypoint ./payload/entrypoint.py
DLO not found	验证 DLO 是否存在（使用 getting-datacloud-schema skill），检查拼写和 __dll 后缀
Permission denied writing	重新运行 scan，验证目标 DLO 是否存在且可写
Deploy fails - wrong directory	确保 --package-dir 指向 payload/ 目录，而非项目根目录

最佳实践

开发

1. 测试前务必先扫描——代码更改后运行 scan
2. 先在本地测试——部署前使用 run 命令
3. 使用版本控制——每次测试成功后执行 git commit
4. 为部署指定版本——使用语义化版本（1.0.0、1.1.0 等）
5. 从项目根目录使用 --package-dir ./payload 进行部署

性能

• CPU_L：小型数据集（< 1M 条记录）
• CPU_2XL：中型数据集（1M-10M 条记录）
• CPU_4XL：大型数据集（> 10M 条记录）

安全

1. 不要硬编码凭据——仅使用 SF CLI 身份验证
2. 验证输入数据——检查空值和数据类型
3. 限制写入权限——仅授予必要的 DLO/DMO 访问权限

与其他技能集成

与 getting-datacloud-schema skill 搭配使用（验证时至关重要）：

在测试代码扩展前，必须使用 getting-datacloud-schema skill 验证 DLO。

与 Datakit Workflow 搭配使用：

1. 通过代码扩展创建 DLO
2. 使用 datakit workflow 将 DLO 映射到 DMO
3. 在细分和激活中使用 DMO

命令参考

命令	用途	必需参数
script init	创建新的脚本项目	--package-dir
function init	创建新的函数项目	--package-dir
script scan	生成配置	入口点文件
script run	本地测试	入口点文件、--target-org
script deploy	部署到 Data Cloud	--target-org、--name、--package-dir、--package-version、--description

资源

• SF CLI 插件：https://github.com/salesforcecli/plugin-data-code-extension
• Python SDK：https://github.com/forcedotcom/datacloud-customcode-python-sdk
• Data Cloud 文档：https://help.salesforce.com/s/articleView?id=sf.c360_a_intro.htm
• Python SDK PyPI：https://pypi.org/project/salesforce-data-customcode/

备注

• 代码扩展在隔离的 Python 3.11 环境中运行
• Docker 仅在部署时需要，本地测试不需要
• 仅使用 SF CLI 身份验证（无需单独的凭据文件）
• Scan 命令会从代码中自动检测权限
• 本地运行使用实际的 Data Cloud 数据（而非模拟数据）
• 部署会进行版本管理，并且可以在 UI 中回滚