从金蝶物料到轻易云：ETL转换和写入的技术实现

  • 轻易云集成顾问-杨嫦

金蝶云星空与轻易云集成平台的数据对接案例分享：查询金蝶物料（新环境）pro

在系统集成的实践中，如何高效、稳定地将不同平台和系统的数据进行无缝对接，是一个极具挑战性的问题。本文将介绍一种实际运行的解决方案——"查询金蝶物料（新环境）pro"，即通过轻易云数据集成平台实现金蝶云星空数据的高效整合。

首先，我们利用金蝶云星空提供的executeBillQuery API接口抓取物料数据。这一过程中，为确保数据不漏单，我们采用了定时作业机制，通过周期性任务调度来定期可靠地调用该接口。为了应对分页和限流问题，我们设计了合理的请求策略，以保证在获取大量数据时依旧能够保持稳定连接和响应速度。

然后，针对从金蝶云星空获取到的数据需要写入到轻易云集成平台，这一步尤为关键。我们使用轻易云集成平台提供的API功能，将这些原始数据以扁平化形式快速、高效地批量写入目标库。在此过程中，自定义的数据转换逻辑成为必不可少的一环，通过映射配置可以灵活适配各种业务需求和特定的数据结构。同时，可视化的数据流设计工具使得整个过程透明清晰，更加便于管理与监控。

实时监控及告警系统是本解决方案的重要保障之一，它确保了每个任务状态都能被及时获知。一旦出现异常情况，如网络波动或接口响应超时，会立即触发错误重试机制，最大程度上减小因突发问题导致的数据丢失风险。此外，在整个任务执行期间，详细日志记录全面捕捉操作细节，有助于后续问题排查与优化改进。

这样的一套技术架构，不仅增强了企业处理大规模、复杂性业务流程数据的能力，同时也提高了整体运作效率和可靠性。后续章节将具体展开如何实现这一对接过程中的各项技术细节，以及注意事项和最佳实践方法。 如何开发金蝶云星空API接口

调用金蝶云星空接口executeBillQuery获取并加工数据

在轻易云数据集成平台中，调用源系统接口是数据集成生命周期的第一步。本文将详细介绍如何通过调用金蝶云星空接口executeBillQuery来获取物料信息，并对其进行初步加工处理。

接口配置与请求参数

首先，我们需要配置接口的基本信息和请求参数。根据提供的元数据配置，接口为executeBillQuery，请求方法为POST。以下是具体的请求参数配置：

{
  "api": "executeBillQuery",
  "method": "POST",
  "number": "FNumber",
  "id": "FMATERIALID",
  "pagination": {
    "pageSize": 100
  },
  "idCheck": true,
  "request": [
    {"field":"FMATERIALID","label":"FMATERIALID","type":"string","value":"FMATERIALID"},
    {"field":"FNumber","label":"编码","type":"string","value":"FNumber"},
    {"field":"FName","label":"名称","type":"string","value":"FName"},
    {"field":"FSpecification","label":"规格型号","type":"string","value":"FSpecification"},
    {"field":"FMnemonicCode","label":"助记码","type":"string","value":"FMnemonicCode"},
    {"field":"FOldNumber","label":"旧物料编码","type":"string","value":"FOldNumber"},
    {"field":"FBARCODE","label":"条码","type":"string","value":"FBARCODE"},
    {"field":"FDescription","label":"描述","type":"string","value":"FDescription"},
    {"field":"FMaterialGroup_FNumber","label":"物料分组","type":"string","value":"FMaterialGroup.FNumber"},
    {"field":"FErpClsID","label":"物料属性","type":"string","value":"FErpClsID"},
    {"field":"FDocumentStatus","label":"数据状态","type":"string","value":"FDocumentStatus"},
    {"field":"FForbidStatus","label":"禁用状态","type":"string","value":"FForbidStatus"},
    {"field": "FilterString", "label": "过滤条件", "type": "string", "describe": "示例写法 FSupplierId.FNumber = 'VEN00010' and FApproveDate>=", "value": "FApproveDate>='{{LAST_SYNC_TIME|dateTime}}' and FUseOrgId.fnumber = '100'"},
    {"field": "FieldKeys", "label": "需查询的字段key集合", "type": "array", "describe": "金蝶分录主键ID格式：FPOOrderEntry_FEntryId,其它格式 FPurchaseOrgId.FNumber", "parser":{"name": "ArrayToString", "params": ","}},
    {"field": "FormId", "label": "业务对象表单Id", "type": "string", "describe": "必须填写金蝶的表单ID如:PUR_PurchaseOrder",    "value":    "BD_MATERIAL"}
  ]
}

请求示例

在实际操作中，我们需要构建一个HTTP POST请求来调用该接口。以下是一个示例请求体：

{
  "_api_": "/k3cloud/Kingdee.BOS.WebApi.ServicesStub.DynamicFormService.ExecuteBillQuery.common.kdsvc",
  "_method_":   "POST",
  "_args_":{
      "_FormId_":"",
      "_FieldKeys_":"",
      "_FilterString_":"",
      "_TopRowCount_":"",
      "_StartRow_":"",
      "_Limit_":"",
      "_OrderString_":"",
      "_SubSystemType_":"",
      "_IsDeleteEntry_":"",
      "_IsAutoSubmitAndAudit_":"",
      "_IsAutoAudit_":"",
      "_IsAutoSubmitAndAuditWhenSaveFaild_":"",
      "_NeedReturnData_":"",
      "_NeedUpLoadAttachmentData_":"",
      "_NeedReturnDeletedDataIds_":"",
      "_NeedReturnDeletedDataIdsWhenSaveFaild_":"",
      "_NeedReturnDeletedDataIdsWhenSaveSuccessed_":"",
      "_NeedReturnDeletedDataIdsWhenDeleteFaild_":"",
      "_NeedReturnDeletedDataIdsWhenDeleteSuccessed_":"",
      "_NeedReturnDeletedDataIdsWhenUnAuditFaild_":"",
      "_NeedReturnDeletedDataIdsWhenUnAuditSuccessed_" :""
   }
}

数据清洗与转换

在获取到原始数据后，需要对数据进行清洗和转换，以确保其符合目标系统的要求。常见的数据清洗操作包括：

  1. 字段映射：将源系统中的字段映射到目标系统中的对应字段。例如，将FMATERIALID映射到目标系统中的material_id。
  2. 数据格式转换：将日期、数字等字段转换为目标系统所需的格式。
  3. 数据过滤：根据业务需求过滤掉不需要的数据。例如，只保留状态为“有效”的物料信息。

以下是一个简单的数据清洗示例：

def clean_data(raw_data):
    cleaned_data = []
    for item in raw_data:
        cleaned_item = {
            'material_id': item['FMATERIALID'],
            'code': item['FNumber'],
            'name': item['FName'],
            'specification': item['FSpecification'],
            'mnemonic_code': item['FMnemonicCode'],
            'old_code': item['FOldNumber'],
            'barcode': item['FBARCODE'],
            'description': item['FDescription'],
            'group_code': item['FMaterialGroup_FNumber'],
            'status': item['FDocumentStatus']
        }
        if cleaned_item['status'] == '有效':
            cleaned_data.append(cleaned_item)
    return cleaned_data

数据写入

完成数据清洗和转换后，下一步就是将处理后的数据写入目标系统。这一步通常涉及调用目标系统的API接口，将清洗后的数据逐条或批量写入。

def write_to_target_system(cleaned_data):
    for data in cleaned_data:
        response = requests.post('https://target-system-api.com/data', json=data)
        if response.status_code != 200:
            print(f"Failed to write data: {data}")

通过以上步骤，我们实现了从金蝶云星空获取物料信息并进行初步加工处理的全过程。这只是轻易云数据集成平台生命周期中的第一步，后续还包括更多的数据处理和监控环节，以确保整个数据流动过程高效、透明。 如何对接钉钉API接口

使用轻易云数据集成平台进行ETL转换和数据写入的技术案例

在数据集成生命周期的第二步，我们需要将已经从源平台（如金蝶物料系统）获取的数据进行ETL（Extract, Transform, Load）转换，并最终写入目标平台（轻易云集成平台）。本文将详细探讨如何通过API接口完成这一过程。

数据提取与清洗

首先，从金蝶物料系统中提取原始数据。假设我们已经通过轻易云数据集成平台的全透明可视化操作界面完成了数据请求与清洗阶段，获取了所需的原始数据。

数据转换

接下来，我们需要对提取的数据进行转换，以符合目标平台API接口所能接收的格式。在这个过程中，元数据配置是关键。以下是一个示例元数据配置：

{
  "api": "空操作",
  "method": "POST",
  "idCheck": true
}

此配置表明我们将使用POST方法调用目标API接口，并且需要进行ID检查。

  1. 字段映射：确保源数据字段与目标API接口字段一一对应。例如，如果源数据包含material_id, material_name, quantity等字段，我们需要将其映射到目标API接口相应的字段。

  2. 数据类型转换：有时源数据和目标API接口所要求的数据类型不一致，需要进行类型转换。例如，将字符串类型的日期格式转换为目标API所需的日期对象。

  3. 数据验证：根据元数据配置中的idCheck属性，确保每条记录在写入之前都经过ID检查，以避免重复或无效的数据。这一步通常包括查重、格式验证等操作。

数据写入

完成数据转换后，下一步是将处理好的数据写入目标平台。根据元数据配置，我们使用POST方法调用目标API接口。以下是一个示例代码片段，用于实现这一过程：

import requests
import json

# 假设已经完成了ETL转换后的数据
transformed_data = [
    {
        "material_id": "12345",
        "material_name": "Example Material",
        "quantity": 100
    },
    # 更多记录...
]

# API URL和头信息
api_url = "https://api.example.com/endpoint"
headers = {
    "Content-Type": "application/json",
    "Authorization": "Bearer your_token_here"
}

# 遍历每条记录并发送POST请求
for record in transformed_data:
    response = requests.post(api_url, headers=headers, data=json.dumps(record))

    # 检查响应状态码
    if response.status_code == 200:
        print(f"Record {record['material_id']} successfully written.")
    else:
        print(f"Failed to write record {record['material_id']}: {response.text}")

在上述代码中，我们遍历每条经过ETL处理后的记录，并使用POST方法将其发送到目标API接口。如果响应状态码为200，则表示写入成功；否则，输出错误信息以便排查问题。

实时监控与日志记录

为了确保整个ETL过程顺利进行，实时监控和日志记录是必不可少的。轻易云数据集成平台提供了实时监控功能，可以帮助我们跟踪每个环节的数据流动和处理状态。此外，通过日志记录可以捕捉到任何异常情况，方便后续分析和问题解决。

总结来说，通过合理配置元数据并严格按照ETL流程操作，可以高效地实现从金蝶物料系统到轻易云集成平台的数据转换与写入。这不仅提高了业务透明度和效率，也确保了数据的一致性和可靠性。 如何开发钉钉API接口

版权声明

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

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