ETL技术在旺店通库存信息与BI数据分析中的应用

  • 轻易云集成顾问-彭亮

案例分享：旺店通旗舰版数据集成到MySQL

在本文中，我们将深入探讨如何通过轻易云数据集成平台，将旺店通·旗舰版的库存信息查询结果高效同步到MySQL数据库，实现实时的数据监控与智能分析。

任务背景

我们面对的主要挑战是将大量来自wms.StockSpec.search2接口的数据快速且无缝地写入到MySQL数据库。为确保数据的完整性和一致性，整个过程需要解决包括接口调用频率限制、分页处理及异常重试等多个技术难题。

方案概述

项目方案命名为“旺店通旗舰版-库存信息查询-->BI泰海-库存信息表(库存查询2)”。该方案旨在利用轻易云的平台特性，实现对旺店通·旗舰版API的高效抓取，并通过自定义转换规则，将这些数据批量导入至MySQL，以供后续BI分析使用。

  1. 高吞吐量的数据写入能力

    • 利用批次操作，将从API获取的大量库存数据一次性提交给MySQL，以提升处理效率。

  2. 集中监控和告警系统

    • 实时跟踪每一条数据集成任务的状态，通过可视化界面展示处理进度和性能指标，确保问题能被及时捕捉并解决。

  3. 分段分页处理与限流机制

    • 由于wms.StockSpec.search2接口存在请求频率限制，我们设计了智能调度策略，在保证不触发限流机制的前提下最大化调用效率。同时，对返回的大体量分页结果，通过流水线式分段读取，有序推进整体工作流程。

  4. 定制化映射与转换逻辑

    • 数据格式差异常导致对接无法顺畅进行。我们针对业务需求，自定义了一套灵活多变的数据映射规则，使得源头结构匹配目标库字段要求，并完美适应两端平台间的信息交互需求。

金蝶与CRM系统接口开发配置

调用旺店通·旗舰版接口wms.StockSpec.search2获取并加工数据

在数据集成过程中，调用源系统的API接口是至关重要的一步。本文将详细探讨如何通过轻易云数据集成平台调用旺店通·旗舰版的wms.StockSpec.search2接口，以获取并加工库存信息数据。

接口调用配置

首先，我们需要配置接口调用的元数据。根据提供的元数据配置，我们可以看到该接口使用POST方法进行请求，主要参数包括分页参数和业务参数。

{
  "api": "wms.StockSpec.search2",
  "method": "POST",
  "number": "spec_no",
  "id": "rec_id",
  "request": [
    {
      "field": "pager",
      "label": "分页参数",
      "type": "object",
      "children": [
        {
          "field": "page_size",
          "label": "分页大小",
          "type": "string",
          "value": "50",
          "parent": "pager"
        },
        {
          "field": "page_no",
          "label": "页号",
          "type": "string",
          "value": "1",
          "parent": "pager"
        }
      ]
    },
    {
      "field": "params",
      "label": "业务参数",
      "type": "object",
      "children": [
        {
          "field": "start_time",
          "label": "开始时间",
          "type": "string",
          "value":"{{LAST_SYNC_TIME|datetime}}"
        },
        {
          {
            field: 'end_time',
            label: '结束时间',
            type: 'string',
            value: '{{CURRENT_TIME|datetime}}'
        }
      ]
    }
  ],
  'autoFillResponse': true,
  'effect': 'QUERY'
}

请求参数解析

  1. 分页参数：

    • page_size：每页返回的数据条数，默认值为50。
    • page_no：当前请求的页码，默认值为1。

  2. 业务参数：

    • start_time：查询的开始时间，使用动态变量{{LAST_SYNC_TIME|datetime}}表示上次同步时间。
    • end_time：查询的结束时间，使用动态变量{{CURRENT_TIME|datetime}}表示当前时间。

这些参数确保了我们能够按需分页获取数据，并且可以灵活地设置查询时间范围，从而实现增量数据同步。

数据请求与清洗

在实际操作中，我们需要通过轻易云平台配置好上述元数据，然后发起API请求。以下是一个示例代码片段，用于展示如何在轻易云平台上进行配置和调用：

import requests
import json

url = 'https://api.wangdiantong.com/wms.StockSpec.search2'
headers = {'Content-Type': 'application/json'}
payload = {
    'pager': {
        'page_size': '50',
        'page_no': '1'
    },
    'params': {
        'start_time': '{{LAST_SYNC_TIME|datetime}}',
        'end_time': '{{CURRENT_TIME|datetime}}'
    }
}

response = requests.post(url, headers=headers, data=json.dumps(payload))
data = response.json()

# 数据清洗逻辑
cleaned_data = []
for item in data['data']:
    cleaned_record = {
        'spec_no': item['spec_no'],
        'rec_id': item['rec_id'],
        # 添加其他需要的字段
    }
    cleaned_data.append(cleaned_record)

数据转换与写入

在完成数据请求和清洗后，我们需要将清洗后的数据转换为目标系统所需的格式，并写入到BI泰海-库存信息表中。以下是一个示例代码片段，用于展示如何进行数据转换和写入：

import pandas as pd

# 将清洗后的数据转换为DataFrame
df = pd.DataFrame(cleaned_data)

# 数据写入逻辑（假设目标系统支持SQLAlchemy）
from sqlalchemy import create_engine

engine = create_engine('mysql+pymysql://user:password@host/dbname')
df.to_sql('bi_taihai_stock_info', con=engine, if_exists='replace', index=False)

通过上述步骤，我们实现了从旺店通·旗舰版获取库存信息，并将其加工后写入到BI泰海-库存信息表中。这一过程不仅保证了数据的一致性和完整性，还提高了业务透明度和效率。 企业微信与ERP系统接口开发配置

数据集成与ETL转换：从旺店通到BI泰海

在数据集成生命周期的第二步中，我们将重点探讨如何将已经集成的源平台数据进行ETL转换，并最终写入目标平台MySQL。本文将详细介绍如何利用轻易云数据集成平台的元数据配置，实现从旺店通旗舰版库存信息查询到BI泰海库存信息表的无缝对接。

ETL转换与写入目标平台

在本案例中，我们需要将旺店通旗舰版的库存信息转换为BI泰海能够接收的格式，并通过MySQL API接口写入目标数据库。以下是关键步骤和技术细节：

1. 数据请求与清洗

首先，通过API接口从旺店通获取库存信息。假设我们已经完成了数据请求和初步清洗，接下来需要进行数据转换。

2. 数据转换配置

根据提供的元数据配置，定义每个字段的映射关系和类型。以下是具体的字段映射：

{
    "api": "batchexecute",
    "effect": "EXECUTE",
    "method": "POST",
    "idCheck": true,
    "request": [
        {"field": "rec_id", "label": "明细唯一键", "type": "string", "value": "{rec_id}"},
        {"field": "defect", "label": "残次品", "type": "string", "value": "{defect}"},
        {"field": "stock_num", "label": "库存量", "type": "string", "value": "{stock_num}"},
        // ... 省略部分字段 ...
        {"field": "status", "label": "状态", "type": "string", "value": "{status}"}
    ],
    // 其他请求参数
    ...
}

3. 构建SQL语句

利用main_sql字段构建插入语句，将清洗后的数据写入目标表wdt_wms_stockspec_search。示例如下：

REPLACE INTO wdt_wms_stockspec_search (
    rec_id, defect, stock_num, wms_sync_stock, wms_stock_diff, spec_no, spec_id,
    goods_no, goods_name, spec_code, brand_name, spec_name, barcode, unpay_num,
    subscribe_num, order_num, sending_num, purchase_num, transfer_num,
    to_purchase_num, purchase_arrive_num, wms_preempty_stock, weight,
    img_url, warehouse_no, warehouse_id, warehouse_name, warehouse_type,
    available_send_stock, created, modified, part_paid_num,
    refund_exch_num, refund_num, refund_onway_num,
    return_exch_num, return_num, return_onway_num,
    to_transfer_num,wms_preempty_diff,wms_sync_time,
    remark ,lock_num ,flag_id ,flag_name ,brand_no ,
    to_other_out_num ,to_process_out_num ,to_process_in_num ,
    last_pd_time ,last_inout_time ,status
) VALUES

4. 数据写入

通过POST请求，将构建好的SQL语句和对应的数据发送至MySQL API接口，实现数据的批量插入或更新操作。

{
    "api":"batchexecute",
    ...
}

在实际操作中，确保每个字段的数据类型和格式都符合目标数据库的要求。例如，日期时间字段需要进行格式化处理：

{"field":"created","label":"创建时间","type":"string","value":"{{created|datetime}}"}

5. 批量处理与性能优化

为了提高效率，可以设置批量处理参数，例如limit字段，控制每次操作的数据量：

{"field":"limit","label":"limit","type":"string","value":"500"}

通过合理设置批处理大小，可以有效减少API调用次数，提高系统性能。

总结

通过上述步骤，我们实现了从旺店通旗舰版到BI泰海库存信息表的数据集成与ETL转换。在这个过程中，充分利用轻易云数据集成平台提供的元数据配置，实现了高效、透明的数据处理流程。关键在于准确定义字段映射关系、构建合适的SQL语句，并通过API接口实现数据的无缝对接。 如何对接金蝶云星空API接口

版权声明

该文章《ETL技术在旺店通库存信息与BI数据分析中的应用》为原创内容，版权归 广东轻亿云软件科技有限公司 所有。 未经本公司书面授权，严禁转载、引用、复制或建立镜像等行为。 若需转载或引用本文内容，请联系本公司获取授权，并在显著位置注明作者姓名、文章出处（包括原文链接）等信息。

数据集成平台架构
更多系统对接方案