ETL在数据集成中的应用：从金蝶云到轻易云

  • 轻易云集成顾问-胡秀丛

案例分享：金蝶云星空数据集成到轻易云集成平台

在现代企业的运营过程中，数据的高度一致性和实时可用性至关重要。本案例将重点探讨如何通过使用executeBillQuery API接口，将金蝶云星空系统中的客户信息无缝集成到轻易云集成平台，实现高效的数据流动管理。

集成背景与目标

本次系统对接任务旨在实现对金蝶云星空中客户信息的批量查询，并将其可靠地写入到轻易云集成平台。为了达到这一目标，关键步骤包括定期抓取数据、处理分页和限流问题、自定义数据转换逻辑以及确保异常处理机制的有效运行。

实现方案概述

  1. API调用与初始化：首先，利用金蝶云星空提供的executeBillQuery API接口进行多次调用，并处理好分页参数，以获取完整的数据集合。在此过程中，需要注意API速率限制以避免请求被拒绝。
  2. 数据转换与映射：由于两大系统之间存在一定的数据格式差异，因此需要设计自定义的数据映射逻辑，对原始数据进行预处理，确保它们符合轻易云接受格式。
  3. 批量写入操作：使用轻易云的平台API——写入操作功能，将预处理后的客户信息快速且准确地导入到统一管理数据库中。该过程需支持高吞吐量的数据读写能力，以保障及时性。
  4. 监控与告警设置：通过集中式监控系统，对整个数据集成任务实时跟踪，包括状态检测、性能指标以及异常告警等，为后续维护提供支持。

上述环节相辅相成为此次“查询-金蝶客户信息”项目奠定了坚实基础。下面我们详述每个步骤中的技术细节及实际应用场景，力求为同行业从业者提供可参考借鉴之道。 数据集成平台API接口配置

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

在数据集成的生命周期中，调用源系统接口获取数据是至关重要的一步。本文将深入探讨如何通过轻易云数据集成平台调用金蝶云星空的executeBillQuery接口，获取客户信息并进行初步加工。

接口配置与请求参数

首先，我们需要了解executeBillQuery接口的基本配置和请求参数。根据元数据配置，以下是该接口的详细信息：

  • API名称: executeBillQuery
  • 操作类型: QUERY
  • 请求方法: POST
  • 业务对象表单Id: BD_Customer

请求参数主要包括客户信息字段和分页参数。具体字段如下：

{
  "FCUSTID": "FCUSTID",
  "FNumber": "编码",
  "FName": "名称",
  "FCreateOrgId_FNumber": "创建组织",
  "FUseOrgId_FNumber": "使用组织",
  "FDescription": "描述",
  "FCustTypeId_FNumber": "客户类别",
  "FGroup_FNumber": "客户分组",
  "FSALDEPTID_FNumber": "销售部门",
  "FSELLER_FNumber": "销售员",
  "FSETTLETYPEID_FNumber": "结算方式",
  "FRECCONDITIONID_FNumber": "收款条件",
  "FShortName": "简称",
  "FADDRESS": "地址",
  "FTEL": "电话",
  "FFAX": "传真",
  "FCompanyClassify_FNumber": "公司类别",
  "FINVOICETITLE": "发票抬头",
  "FINVOICEBANKACCOUNT": "银行账号",
  ...
}

分页参数包括：

  • Limit: 最大行数
  • StartRow: 开始行索引
  • TopRowCount: 返回总行数
  • FilterString: 过滤条件
  • FieldKeys: 查询字段集合

请求示例

为了更好地理解如何调用该接口，我们来看一个具体的请求示例：

{
    "FormId":"BD_Customer",
    "FieldKeys":"FCUSTID,FNumber,FName,FCreateOrgId.FNumber,FUseOrgId.FNumber,FDescription,FCustTypeId.FNumber,FGroup.FNumber,FSALDEPTID.FNumber,FSELLER.FNumber,FSETTLETYPEID.FNumber,FRECCONDITIONID.FNumber,FShortName,FADDRESS,FTEL,FFAX,FCompanyClassify.FNumber,FINVOICETITLE,FINVOICEBANKACCOUNT,FCURRENCYID.FNumber,FTRADINGCURRID,F_POKM_saleorgId.FNumber,F_POKM_StockOrgId.FNumber,F_POKM_SettleOrgId.FNumber",
    ...
    ...
    ...
}

在这个请求中，我们指定了需要查询的字段集合，并设置了分页参数和过滤条件。

数据清洗与转换

获取到原始数据后，需要对其进行清洗和转换，以便后续处理和存储。以下是一些常见的数据清洗与转换操作：

  1. 字段重命名：将原始字段名转换为目标系统所需的字段名。
  2. 数据格式转换：例如，将日期字符串转换为标准日期格式。
  3. 缺失值处理：填补或删除缺失值。
  4. 数据过滤：根据业务需求筛选出符合条件的数据。

实践案例

假设我们从金蝶云星空获取到以下客户信息：

[
    {
        "FCUSTID":"CUST0001",
        ...
        ...
        ...
    },
    {
        ...
        ...
        ...
    }
]

我们需要将这些数据清洗并转换为目标系统所需的格式。例如，将FCUSTID重命名为CustomerID，并将日期字符串"2023-10-01"转换为标准日期格式。

[
    {
        ...
        ...
        ...
    }
]

总结

通过轻易云数据集成平台调用金蝶云星空接口executeBillQuery，我们可以高效地获取客户信息，并对其进行必要的数据清洗与转换。这一步骤不仅确保了数据的一致性和准确性，还为后续的数据处理和分析奠定了坚实基础。在实际应用中，根据具体业务需求灵活调整请求参数和清洗规则，是实现高效数据集成的关键。 钉钉与ERP系统接口开发配置

数据转换与写入：轻易云数据集成平台API接口应用案例

在数据集成生命周期的第二步，ETL（Extract, Transform, Load）转换是至关重要的一环。本文将深入探讨如何将已经集成的源平台数据进行ETL转换，转为目标平台轻易云集成平台API接口所能够接收的格式，并最终写入目标平台。

数据请求与清洗

在数据转换之前，首先需要从源系统中获取原始数据并进行清洗。假设我们从金蝶系统中查询客户信息，获取的数据可能包含多种格式和字段。为了确保数据质量，我们需要对这些数据进行初步清洗，包括去除冗余字段、标准化日期格式、校验数据完整性等操作。

import pandas as pd

# 从金蝶系统获取原始客户数据
raw_data = get_data_from_kingdee()

# 数据清洗
cleaned_data = raw_data.drop_duplicates().dropna()
cleaned_data['date'] = pd.to_datetime(cleaned_data['date'], format='%Y-%m-%d')

数据转换

清洗后的数据需要按照目标平台的要求进行转换。在这个过程中，我们需要根据轻易云集成平台API接口的元数据配置，对数据进行适当的调整。例如，将字段名称映射为目标平台所需的格式，或者将某些字段合并或拆分。

# 字段映射
mapped_data = cleaned_data.rename(columns={
    'customer_name': 'name',
    'customer_id': 'id',
    'contact_number': 'phone'
})

# 其他必要的转换操作
mapped_data['full_address'] = mapped_data['address'] + ', ' + mapped_data['city']

数据写入

完成数据转换后，我们使用轻易云集成平台提供的API接口将数据写入目标平台。根据提供的元数据配置，我们需要使用POST方法，并且在执行前进行ID检查。

import requests

api_url = "https://api.qingyiyun.com/execute"
headers = {
    "Content-Type": "application/json"
}

for index, row in mapped_data.iterrows():
    payload = {
        "name": row['name'],
        "id": row['id'],
        "phone": row['phone'],
        "full_address": row['full_address']
    }

    response = requests.post(api_url, json=payload, headers=headers)

    if response.status_code == 200:
        print(f"Data for {row['name']} successfully written.")
    else:
        print(f"Failed to write data for {row['name']}. Status code: {response.status_code}")

API接口特性与注意事项

  1. 异步处理：轻易云集成平台支持全异步处理，这意味着我们可以同时发送多个请求而不必等待每个请求完成。这极大地提高了效率，但也需要注意并发控制和错误处理。

  2. ID检查：根据元数据配置中的idCheck属性，在写入前需要检查ID是否存在，以避免重复写入或覆盖已有记录。这可以通过预先查询目标系统中的现有记录来实现。

  3. 错误处理：在实际应用中，可能会遇到各种错误，如网络问题、API限流等。因此，需要设计健壮的错误处理机制，例如重试策略、日志记录等。

# 示例：增加重试机制和错误日志记录
import time

def write_data_with_retry(payload, retries=3):
    for attempt in range(retries):
        try:
            response = requests.post(api_url, json=payload, headers=headers)
            if response.status_code == 200:
                return True
            else:
                print(f"Attempt {attempt + 1} failed with status code: {response.status_code}")
                time.sleep(2)  # 等待一段时间后重试
        except Exception as e:
            print(f"Attempt {attempt + 1} encountered an error: {e}")
            time.sleep(2)

    return False

for index, row in mapped_data.iterrows():
    payload = {
        "name": row['name'],
        "id": row['id'],
        "phone": row['phone'],
        "full_address": row['full_address']
    }

    success = write_data_with_retry(payload)

    if success:
        print(f"Data for {row['name']} successfully written.")
    else:
        print(f"Failed to write data for {row['name']} after multiple attempts.")

通过以上步骤，我们实现了从源系统到目标平台的数据ETL转换和写入。在实际项目中，根据具体需求，还可以进一步优化和定制化这些步骤，以更好地满足业务需求。 轻易云数据集成平台金蝶集成接口配置

版权声明

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

轻易云企业AIGC应用
更多系统对接方案