使用轻易云平台实现ETL转换并写入管易云系统

  • 轻易云集成顾问-彭萍

kw_仓库同步：金蝶云星空与管易云的数据集成解决方案

在复杂的企业资源管理系统中，数据的高效流动和无缝对接是业务成功的关键。本文将分享一个基于轻易云数据集成平台，实现金蝶云星空与管易云之间库存数据同步的案例——kw_仓库同步，通过具体技术要点剖析集成过程中的各类挑战及解决方案。

1. 数据源获取：调用金蝶云星空API

首先，我们需要从金蝶云星空中获取最新库存数据。为此，采用了executeBillQuery接口，该接口支持批量查询并能灵活设置过滤条件以满足不同业务场景需求。然而，由于API本身存在分页和限流问题，需要仔细处理以确保每次请求都能可靠返回完整的数据。此外，为了避免遗漏单据，还必须设立重试机制，当发生意外错误时进行有效补偿。

步骤：
1. 设置API请求参数，包括分页信息、时间范围等。
2. 调用executeBillQuery接口获取初始响应。
3. 检查是否有下一页，如果有则继续调用直到所有页面读取完毕。
4. 对可能出现的异常情况进行监控和处理，以确保不会漏单。

2. 数据转换及质量保证

从金蝶云星空提取的数据格式通常与管易云要求的不一致。因此，在传输前需要使用自定义数据转换逻辑，将其适配到目标系统所需结构。同时，启用了轻易云的数据质量监控功能，对每一条记录进行校验，及时发现并纠正潜在的问题。这不仅提高了整体准确性，也规避了因不合规数据导致的业务风险。

转化逻辑示例：
- 将日期格式YYYY-MM-DD HH:MM:SS转为标准UTC时间戳。
- 校验SKU编码是否符合统一规则，不合规记录标记为异常待处理。

3. 大量数据快速写入至管易云

为了实现容量庞大的实时库存更新，此次方案采用了gy.erp.warehouse.add接口，并结合定时任务机制，实现批量异步写入。针对大量并发请求进行了优化配置，以维持高吞吐量性能表现。在实际应用中，这意味着数百万级别的数据可以迅速且稳定地导入到管易系统，大大提升操作效率。

执行流程：
1. 将转化后的批量数据按指定大小分组打包发送至目标API端点。
2. 实现失败重试策略，对于未成功写入部分重新提交直至完成确认（最多三次）。
3. 全程配有详细日志记录便
![金蝶与SCM系统接口开发配置](https://pic.qeasy.cloud/D27.png~tplv-syqr462i7n-qeasy.image)
### 调用金蝶云星空接口executeBillQuery获取并加工数据

在数据集成生命周期的第一步，我们需要从源系统金蝶云星空获取数据，并进行初步加工。本文将详细探讨如何通过调用金蝶云星空的`executeBillQuery`接口来实现这一目标。

#### API接口配置

首先，了解我们将使用的API接口配置。根据提供的元数据配置，我们需要使用`POST`方法调用`executeBillQuery`接口，以下是具体的请求参数和配置细节：

- **API**: `executeBillQuery`
- **Method**: `POST`
- **分页参数**:
  - `pageSize`: 100
  - `StartRow`: 根据当前页数计算
- **请求字段**:
  - `FStockId`: 仓库ID
  - `FNumber`: 编码
  - `FName`: 名称
  - `FGroup`: 分组

#### 请求参数详解

1. **基本请求字段**：
   ```json
   [
     {"field":"FStockId","label":"id","type":"string","value":"FStockId"},
     {"field":"FNumber","label":"编码","type":"string","value":"FNumber"},
     {"field":"FName","label":"名称","type":"string","value":"FName"},
     {"field":"FGroup","label":"分组","type":"string","value":"FGroup"}
   ]

这些字段是我们从金蝶云星空获取仓库信息所需的基本字段。

  1. 其他请求参数： 
    [
 {"field":"Limit","label":"最大行数","type":"string","describe":"金蝶的查询分页参数","value":"{PAGINATION_PAGE_SIZE}"},
 {"field":"StartRow","label":"开始行索引","type":"string","describe":"金蝶的查询分页参数","value":"{PAGINATION_START_ROW}"},
 {"field":"TopRowCount","label":"返回总行数","type":"int"},
 {"field":"FilterString","label":"过滤条件","type":"string","describe":"","value":""},
 {"field":"FieldKeys","label":"需查询的字段key集合","type": "array", "parser":{"name": "ArrayToString", "params": ","}},
 {"field": "FormId", "label": "业务对象表单Id", "type": "string", "describe": "", "value": "BD_STOCK"}
]
    • Limit 和 StartRow 用于分页控制。
    • FilterString 可以用于设置过滤条件，例如：FAUDITDATE>='{{LAST_SYNC_TIME|dateTime}}'。
    • FieldKeys 用于指定需要查询的字段集合。
    • FormId 固定为仓库表单ID：BD_STOCK。

调用示例

以下是一个完整的调用示例：

{
  "api": "executeBillQuery",
  "method": "POST",
  "number": "FNumber",
  "id": "FStockId",
  "pagination": {
    "pageSize": 100,
    "startRow": "{PAGINATION_START_ROW}"
  },
  "request": [
    {"field": "FStockId", "label": "id", "type": "string", "value": ""},
    {"field": "FNumber", "label": "编码", "type": "", value: ""},
    {"field": "FName", label: "", type: "", value: ""},
    {"field": "", label: "", type: "", value: ""}
  ],
  otherRequest: [
    { field: 'Limit', label: '最大行数', type: 'string', describe: '', value: '{PAGINATION_PAGE_SIZE}' },
    { field: 'StartRow', label: '开始行索引', type: 'string', describe: '', value: '{PAGINATION_START_ROW}' },
    { field: 'TopRowCount', label: '返回总行数', type: 'int' },
    { field: 'FilterString', label: '', type:'', describe:'', value:''},
    { field:'FieldKeys', label:'需查询的字段key集合', type:'array'},
    { field:'FormId', label:'业务对象表单Id', type:'string'}
]
}

数据处理与清洗

在获取到原始数据后，需要对数据进行初步清洗和处理。例如，可以根据业务需求对某些字段进行格式化或转换。假设我们需要将日期格式化为标准格式，可以使用如下代码：

import datetime

def format_date(date_str):
    return datetime.datetime.strptime(date_str, '%Y-%m-%d').strftime('%Y-%m-%d')

# 示例：清洗数据中的日期字段
for record in data:
    record['FAUDITDATE'] = format_date(record['FAUDITDATE'])

通过上述步骤，我们可以确保从金蝶云星空获取的数据符合后续处理和分析的要求。

总结

通过调用金蝶云星空接口executeBillQuery，我们能够高效地获取并初步加工所需的数据。这一步骤是整个数据集成生命周期中的关键环节，为后续的数据转换与写入奠定了坚实基础。 打通用友BIP数据接口

使用轻易云数据集成平台进行ETL转换并写入管易云API接口的技术案例

在数据集成生命周期的第二步，我们将重点讨论如何将已经集成的源平台数据进行ETL（Extract, Transform, Load）转换，并最终写入目标平台——管易云API接口。本文将详细介绍相关技术细节和操作步骤。

数据请求与清洗

在ETL过程的第一步，数据请求与清洗阶段，我们需要从源平台获取原始数据，并对其进行初步清洗和标准化处理。这一步主要是为了确保数据的准确性和一致性，为后续的数据转换打下基础。

数据转换与写入

接下来，我们进入数据转换与写入阶段，这是整个ETL过程的核心部分。在这一阶段，我们需要将清洗后的数据转换为目标平台所能接收的格式，并通过API接口将其写入到管易云系统中。

元数据配置解析

根据提供的元数据配置，我们需要调用管易云的gy.erp.warehouse.add API接口，该接口用于添加仓库信息。以下是元数据配置的详细解析：

  • API接口: gy.erp.warehouse.add
  • 请求方法: POST
  • ID检查: true

请求字段包括：

  • warehouse_code（仓库代码）：对应源数据中的FNumber
  • warehouse_name（仓库名称）：对应源数据中的FName
  • contact_name（联系人）
  • contact_phone（联系电话）
  • contact_mobile（联系手机）
  • province（省信息）
  • city（市信息）
  • district（区信息）
  • address（详细地址）
  • note（备注）
数据映射与转换

在进行数据转换时，我们需要将源平台的数据字段映射到目标平台所需的数据字段。例如，源平台中的仓库编号字段FNumber需要映射到管易云API接口中的warehouse_code字段。类似地，仓库名称字段FName需要映射到warehouse_name字段。

以下是一个具体的数据映射示例：

{
    "warehouse_code": "{FNumber}",
    "warehouse_name": "{FName}",
    "contact_name": "{ContactName}",
    "contact_phone": "{ContactPhone}",
    "contact_mobile": "{ContactMobile}",
    "province": "{Province}",
    "city": "{City}",
    "district": "{District}",
    "address": "{Address}",
    "note": "{Note}"
}
API请求构建

完成数据映射后，我们需要构建API请求，将转换后的数据发送到管易云系统。以下是一个使用Python语言构建API请求的示例代码：

import requests

# 定义API URL和头部信息
api_url = "https://api.guanyiyun.com/gy.erp.warehouse.add"
headers = {
    "Content-Type": "application/json",
    "Authorization": "Bearer YOUR_ACCESS_TOKEN"
}

# 构建请求体
payload = {
    "warehouse_code": source_data["FNumber"],
    "warehouse_name": source_data["FName"],
    "contact_name": source_data.get("ContactName", ""),
    "contact_phone": source_data.get("ContactPhone", ""),
    "contact_mobile": source_data.get("ContactMobile", ""),
    "province": source_data.get("Province", ""),
    "city": source_data.get("City", ""),
    "district": source_data.get("District", ""),
    "address": source_data.get("Address", ""),
    "note": source_data.get("Note", "")
}

# 发送POST请求
response = requests.post(api_url, headers=headers, json=payload)

# 检查响应状态
if response.status_code == 200:
    print("Data successfully written to Guanyi Cloud.")
else:
    print(f"Failed to write data. Status code: {response.status_code}, Response: {response.text}")

在上述代码中，我们首先定义了API URL和头部信息，然后构建了请求体，将映射后的数据填充进去。最后，通过发送POST请求，将数据写入到管易云系统中。

异常处理与日志记录

为了确保整个ETL过程的稳定性和可靠性，我们还需要对异常情况进行处理，并记录日志。这样可以帮助我们及时发现并解决问题，提高系统的健壮性。

以下是一个简单的异常处理和日志记录示例：

import logging

# 配置日志记录
logging.basicConfig(filename='etl_process.log', level=logging.INFO)

try:
    # 发送POST请求
    response = requests.post(api_url, headers=headers, json=payload)

    # 检查响应状态
    if response.status_code == 200:
        logging.info("Data successfully written to Guanyi Cloud.")
        print("Data successfully written to Guanyi Cloud.")
    else:
        logging.error(f"Failed to write data. Status code: {response.status_code}, Response: {response.text}")
        print(f"Failed to write data. Status code: {response.status_code}, Response: {response.text}")
except Exception as e:
    logging.exception("An error occurred during the ETL process.")

通过上述代码，我们可以将异常情况记录到日志文件中，方便后续排查和解决问题。

综上所述，通过轻易云数据集成平台，我们可以高效地实现从源平台到目标平台的数据ETL转换，并通过管易云API接口将转换后的数据写入目标系统。这一过程不仅提高了业务流程的自动化程度，也增强了系统间的数据一致性和可靠性。 打通钉钉数据接口

版权声明

该文章《使用轻易云平台实现ETL转换并写入管易云系统》为原创内容，版权归 广东轻亿云软件科技有限公司 所有。 未经本公司书面授权，严禁转载、引用、复制或建立镜像等行为。 若需转载或引用本文内容，请联系本公司获取授权，并在显著位置注明作者姓名、文章出处（包括原文链接）等信息。

数据集成平台集成生命周期
更多系统对接方案