n8n中的数据分析：从Python到JavaScript的代码节点应用策略

对于数据科学家而言，首次体验n8n工作流时，其便捷性可能令人感到惊喜。该平台无需查阅冗长的API文档，即可轻松连接各类API，还能通过Gmail或Google Sheets触发工作流，数分钟内便可部署实用功能。

然而，一个显著的局限在于，n8n的云实例并未原生优化以支持Python环境，这对于习惯使用Python进行数据分析的用户来说，可能构成挑战。

与许多数据科学家一样，日常数据分析工具箱主要基于NumPy和Pandas构建。为了保持在熟悉的领域内工作，文章常常将计算任务外包给外部API，而非直接使用n8n的JavaScript代码节点。

例如，一个生产计划优化工具的实现就是通过一个工作流来编排的，该工作流包含一个调用FastAPI微服务的Agent节点。这种方法虽然有效，但有客户提出希望在n8n用户界面上完全可见数据分析任务的执行过程。

文章作者意识到有必要学习足够的JavaScript，以便利用n8n的原生代码节点执行数据处理任务。

本文将探讨如何在n8n代码节点内部使用小型JavaScript代码片段，来执行日常数据分析任务。为演示目的，文章将使用一个销售交易数据集，并逐步进行ABC和帕累托分析，这两种分析在供应链管理中应用广泛。

文章将提供Pandas与n8n JavaScript代码节点的对比示例，以便将熟悉的Python数据分析步骤直接转换为自动化的n8n工作流。

核心思想是在云企业级n8n实例的能力范围内（即不使用社区节点），为小型数据集或快速原型设计实现这些解决方案。

本次实验将以快速比较研究作为总结，对比其与FastAPI调用的性能表现。读者可以根据文章中分享的Google Sheet和工作流模板，复现整个工作流。

在n8n中使用JavaScript构建数据分析工作流

在开始构建节点之前，文章将介绍本次分析的背景。

供应链管理中的ABC与帕累托图表

本教程建议读者构建一个简单的工作流，该工作流从Google Sheets获取销售交易数据，并将其转换为全面的ABC和帕累托图表。这将复现作者初创公司LogiGreen开发的LogiGreen Apps中的ABC和帕累托分析模块。

其目标是为超市连锁店的库存团队生成一套可视化图表，帮助他们理解销售在各门店间的分布情况。文章将重点生成两个可视化图表。

第一个图表展示了销售商品的ABC-XYZ分析：

• X轴 (销售额百分比 %)：各商品对总收入的贡献。
• Y轴 (变异系数)：各商品的需求变异性。
• 垂直红线根据销售额份额将商品分为A、B和C类。
• 水平蓝线标记了稳定需求与变动需求(CV=1)。

这些指标共同突出了哪些商品是高价值且稳定 (A类，低CV)，哪些是低价值或高度变动的，从而指导库存管理中的优先级排序。

第二个可视化图表是销售额的帕累托分析：

• X轴：SKU百分比（按销售额排名）。
• Y轴：年度销售额累计百分比。
• 曲线说明了少量商品如何贡献大部分收入。

简而言之，这突出（或未突出）了经典的帕累托法则，即80%的销售额可能来自20%的SKU。

这些图表是如何生成的？简单来说，就是通过Python实现。

在一个YouTube频道上，有提供完整的教程，展示了如何使用Pandas和Matplotlib完成此操作。本教程的目标是利用n8n的原生JavaScript节点，在Google Sheet中准备销售交易数据并生成这些图表。

在n8n中构建数据分析工作流

建议构建一个手动触发的工作流，以方便开发过程中的调试。

读者若要跟随本教程，需要完成以下步骤：

• 复制此链接提供的电子表格：Google Sheet
• 从n8n作者资料下载模板

现在，可以使用第二个节点（将从工作表Input Data中提取数据集）连接已复制的Google Sheet。

该数据集包含每日粒度的零售销售交易：

• ITEM: 可以在多个门店销售的商品
• SKU: 在特定门店销售的SKU
• FAMILY: 商品组
• CATEGORY: 产品类别，可包含多个商品组
• STORE: 代表销售地点的代码
• DAY: 交易日期
• QTY: 销售数量（单位）
• TO: 销售额（欧元）

输出是JSON格式的表格内容，可供其他节点使用。

Python 代码

import pandas as pd
df = pd.read_csv("sales.csv")

接下来，可以开始处理数据集以构建两个可视化图表。

步骤 1: 过滤掉销量为零的交易

首先，从过滤掉销量QTY等于零的交易这一简单操作开始。

无需JavaScript；一个简单的Filter节点即可完成任务。

Python 代码

df = df[df["QTY"] != 0]

步骤 2: 为帕累托分析准备数据

首先需要按ITEM聚合销售数据并按销售额排名产品。

Python 代码

sku_agg = (df.groupby("ITEM", as_index=False)
             .agg(TO=("TO","sum"), QTY=("QTY","sum"))
             .sort_values("TO", ascending=False))

在工作流中，这一步骤将在JavaScript节点TO, QTY GroupBY ITEM中完成：

const agg = {};
for (const {json} of items) {
  const ITEM = json.ITEM;
  const TO = Number(json.TO);
  const QTY = Number(json.QTY);
  if (!agg[ITEM]) agg[ITEM] = { ITEM, TO: 0, QTY: 0 };
  agg[ITEM].TO += TO;
  agg[ITEM].QTY += QTY;
}
const rows = Object.values(agg).sort((a,b)=> b.TO - a.TO);
return rows.map(r => ({ json: r }));

此节点返回按ITEM分类的销售额（TO）和数量（QTY）的排名表：

1. 初始化agg为一个以ITEM为键的字典。
2. 遍历n8n的行数据。

• 将TO和QTY转换为数字。
• 将QTY和TO值添加到每个ITEM的累计总额中。

1. 最终将字典转换为按TO降序排列的数组并返回数据。

现在数据已准备好，可以对销售数量（QTY）或销售额（TO）进行帕累托分析。为此，需要计算累计销售额并对SKU进行排名，从最高贡献者到最低贡献者。

Python 代码

abc = sku_agg.copy()  # from Step 2, already sorted by TO desc
total = abc["TO"].sum() or 1.0
abc["cum_turnover"] = abc["TO"].cumsum()
abc["cum_share"]    = abc["cum_turnover"] / total             
abc["sku_rank"]     = range(1, len(abc) + 1)
abc["cum_skus"]     = abc["sku_rank"] / len(abc)               
abc["cum_skus_pct"] = abc["cum_skus"] * 100

这一步骤将在代码节点Pareto Analysis中完成：

const rows = items
  .map(i => ({
    ...i.json,
    TO: Number(i.json.TO || 0),   
    QTY: Number(i.json.QTY || 0),
  }))
  .sort((a, b) => b.TO - a.TO);

const n = rows.length; // number of ITEM
const totalTO = rows.reduce((s, r) => s + r.TO, 0) || 1;

从上一个节点收集数据集items：

1. 对于每一行，清理字段TO和QTY（以防有缺失值）。
2. 按销售额降序排序所有SKU。
3. 将商品总数和总销售额存储在变量中。

let cumTO = 0;
rows.forEach((r, idx) => {
  cumTO += r.TO;
  r.cum_turnover = cumTO;                     
  r.cum_share = +(cumTO / totalTO).toFixed(6); 
  r.sku_rank = idx + 1;
  r.cum_skus = +((idx + 1) / n).toFixed(6);   
  r.cum_skus_pct = +(r.cum_skus * 100).toFixed(2);
});

return rows.map(r => ({ json: r }));

然后遍历所有已排序的商品：

1. 使用变量cumTO计算累计贡献。
2. 向每一行添加多个帕累托指标：

• cum_turnover: 截至当前商品的累计销售额。
• cum_share: 销售额累计份额。
• sku_rank: 商品排名。
• cum_skus: SKU累计数量占总SKU的比例。
• cum_skus_pct: 与cum_skus相同，但以百分比表示。

至此，帕累托图表的数据准备工作完成。

该数据集将由节点Update Pareto Sheet存储到Pareto工作表中。通过一些巧妙的操作，可以在第一个工作表中生成此图表：

现在可以继续进行ABC XYZ图表的创建。

步骤 3: 计算需求变异性和销售贡献

虽然可以复用帕累托图表的输出来计算销售贡献，但为清晰起见，文章将每个图表视为独立处理。为清晰起见，‘需求变异性’与‘销售额百分比’节点中的代码将被拆分为多个片段。

代码块 1: 定义均值和标准差函数

function mean(a){ return a.reduce((s,x)=>s + x, 0) / (a.length || 1); }
function stdev_samp(a){
  if (a.length <= 1) return 0;
  const m = mean(a);
  const v = a.reduce((s,x)=> s + (x - m) ** 2, 0) / (a.length - 1);
  return Math.sqrt(v);
}

这两个函数将用于计算变异系数（Cov）：

• mean(a): 计算数组的平均值。
• stdev_samp(a): 计算样本标准差。

它们接收作为输入的数据是我们在第二个代码块中构建的每个ITEM的每日销售分布。

代码块 2: 创建每个ITEM的每日销售分布

const series = {};  // ITEM -> { day -> qty_sum }
let totalQty = 0;

for (const { json } of items) {
  const item = String(json.ITEM);
  const day  = String(json.DAY);
  const qty  = Number(json.QTY || 0);

  if (!series[item]) series[item] = {};
  series[item][day] = (series[item][day] || 0) + qty;
  totalQty += qty;
}

Python 代码

import pandas as pd
import numpy as np
df['QTY'] = pd.to_numeric(df['QTY'], errors='coerce').fillna(0)
daily_series = df.groupby(['ITEM', 'DAY'])['QTY'].sum().reset_index()

现在可以计算应用于每日销售分布的指标。

const out = [];
for (const [item, dayMap] of Object.entries(series)) {
  const daily = Object.values(dayMap); // daily sales quantities
  const qty_total = daily.reduce((s,x)=>s+x, 0);
  const m = mean(daily);               // average daily sales
  const sd = stdev_samp(daily);        // variability of sales
  const cv = m ? sd / m : null;        // coefficient of variation
  const share_qty_pct = totalQty ? (qty_total / totalQty) * 100 : 0;

  out.push({
    ITEM: item,
    qty_total,
    share_qty_pct: Number(share_qty_pct.toFixed(2)),
    mean_qty: Number(m.toFixed(3)),
    std_qty: Number(sd.toFixed(3)),
    cv_qty: cv == null ? null : Number(cv.toFixed(3)),
  });
}

对于每个ITEM，计算以下指标：

• qty_total: 总销售额。
• mean_qty: 日均销售额。
• std_qty: 日销售额标准差。
• cv_qty: 变异系数（用于XYZ分类的变异性衡量指标）。
• share_qty_pct: 对总销售额的贡献百分比（用于ABC分类）。

以下是Python版本，以防读者在阅读过程中迷失：

summary = daily_series.groupby('ITEM').agg(
    qty_total=('QTY', 'sum'),
    mean_qty=('QTY', 'mean'),
    std_qty=('QTY', 'std')
).reset_index()

summary['std_qty'] = summary['std_qty'].fillna(0)

total_qty = summary['qty_total'].sum()
summary['cv_qty'] = summary['std_qty'] / summary['mean_qty'].replace(0, np.nan)
summary['share_qty_pct'] = 100 * summary['qty_total'] / total_qty

工作已接近完成。只需按贡献降序排序，为ABC分类映射做准备：

out.sort((a,b) => b.share_qty_pct - a.share_qty_pct);
return out.map(r => ({ json: r }));

现在，对于每个ITEM，都拥有了创建散点图所需的关键指标。

在此步骤中，仅缺少ABC分类。

步骤 4: 添加ABC分类

将上一个节点的输出作为输入。

let rows = items.map(i => i.json);
rows.sort((a, b) => b.share_qty_pct - a.share_qty_pct);

为确保万无一失，按销售额份额（%）降序排列商品 → 最重要的SKU排在前面。（此步骤通常已在上一代码节点结束时完成，可选择省略。）

然后可以根据预设条件进行分类：

• A: 合计占前5%销售额的SKU。
• B: 合计占接下来15%销售额的SKU。
• C: 20%之后的所有SKU。

let cum = 0;
for (let r of rows) {
  cum += r.share_qty_pct;

  // 3) 根据累计百分比分配类别
  if (cum <= 5) {
    r.ABC = 'A';   // 前 5%
  } else if (cum <= 20) {
    r.ABC = 'B';   // 接下来 15%
  } else {
    r.ABC = 'C';   // 剩余部分
  }

  r.cum_share = Number(cum.toFixed(2));
}

return rows.map(r => ({ json: r }));

通过Python代码可以这样实现：

df = df.sort_values('share_qty_pct', ascending=False).reset_index(drop=True)
df['cum_share'] = df['share_qty_pct'].cumsum()
def classify(cum):
    if cum <= 5:
        return 'A'
    elif cum <= 20:
        return 'B'
    else:
        return 'C'
df['ABC'] = df['cum_share'].apply(classify)

现在，结果可用于生成此图表，该图表可在Google Sheet的第一个工作表中找到：

由于对Google Sheets的知识有限，文章作者难以找到一种“手动”解决方案来创建具有正确颜色映射的散点图。因此，文章使用了Google Sheets中提供的Google Apps Script来创建它。

作为附加内容，n8n模板中还添加了更多节点，用于执行相同类型的分组操作，以计算按门店或按ITEM-门店对的销售额。

这些节点可用于创建类似这样的可视化图表：

总结本教程，可以确信任务已圆满完成。若要观看工作流的实时演示，可以查阅这个简短的教程。

在n8n云实例上运行此工作流的客户，现在可以清晰地看到数据处理的每一步。

但这是否意味着性能有所损失？

这正是下一节将要探讨的问题。

性能对比研究：n8n JavaScript节点 vs. FastAPI中的Python

为回答这个问题，文章准备了一个直接的实验。使用两种不同的方法在n8n内部处理相同的数据集和转换：

1. 纯JavaScript节点：所有功能直接在n8n内部实现。
2. 外包给FastAPI微服务：将JavaScript逻辑替换为对Python端点的HTTP请求。

这两个端点连接到直接从托管微服务的VPS实例加载数据的函数。

@router.post("/launch_pareto")
async def launch_speedtest(request: Request):
    try:
        session_id = request.headers.get('session_id', 'session')

        folder_in = f'data/session/speed_test/input'
        if not path.exists(folder_in):
                makedirs(folder_in)

        file_path = folder_in + '/sales.csv'
        logger.info(f"[SpeedTest]: Loading data from session file: {file_path}")
        df = pd.read_csv(file_path, sep=";")
        logger.info(f"[SpeedTest]: Data loaded successfully: {df.head()}")

        speed_tester = SpeedAnalysis(df)
        output = await speed_tester.process_pareto()

        result = output.to_dict(orient="records")
        result = speed_tester.convert_numpy(result)

        logger.info(f"[SpeedTest]: /launch_pareto completed successfully for {session_id}")
        return result
    except Exception as e:
        logger.error(f"[SpeedTest]: Error /launch_pareto: {str(e)}
{traceback.format_exc()}")
        raise HTTPException(status_code=500, detail=f"Failed to process Speed Test Analysis: {str(e)}")

@router.post("/launch_abc_xyz")
async def launch_abc_xyz(request: Request):
    try:
        session_id = request.headers.get('session_id', 'session')

        folder_in = f'data/session/speed_test/input'
        if not path.exists(folder_in):
                makedirs(folder_in)

        file_path = folder_in + '/sales.csv'
        logger.info(f"[SpeedTest]: Loading data from session file: {file_path}")
        df = pd.read_csv(file_path, sep=";")
        logger.info(f"[SpeedTest]: Data loaded successfully: {df.head()}")

        speed_tester = SpeedAnalysis(df)
        output = await speed_tester.process_abcxyz()

        result = output.to_dict(orient="records")
        result = speed_tester.convert_numpy(result)

        logger.info(f"[SpeedTest]: /launch_abc_xyz completed successfully for {session_id}")
        return result
    except Exception as e:
        logger.error(f"[SpeedTest]: Error /launch_abc_xyz: {str(e)}
{traceback.format_exc()}")
        raise HTTPException(status_code=500, detail=f"Failed to process Speed Test Analysis: {str(e)}")

本次测试将仅关注数据处理性能。

SpeedAnalysis类包含了上一节中列出的所有数据处理步骤：

• 按ITEM分组销售额
• 按降序排列ITEM并计算累计销售额
• 计算按ITEM分类的销售分布的标准差和均值

class SpeedAnalysis:
    def __init__(self, df: pd.DataFrame):
        config = load_config()

        self.df = df

    def processing(self):
        try:
            sales = self.df.copy()
            sales = sales[sales['QTY']>0].copy()
            self.sales = sales

        except Exception as e:
            logger.error(f'[SpeedTest] Error for processing : {e}
{traceback.format_exc()}')

    def prepare_pareto(self):
        try:
            sku_agg = self.sales.copy()
            sku_agg = (sku_agg.groupby("ITEM", as_index=False)
             .agg(TO=("TO","sum"), QTY=("QTY","sum"))
             .sort_values("TO", ascending=False))

            pareto = sku_agg.copy()  
            total = pareto["TO"].sum() or 1.0
            pareto["cum_turnover"] = pareto["TO"].cumsum()
            pareto["cum_share"]    = pareto["cum_turnover"] / total              
            pareto["sku_rank"]     = range(1, len(pareto) + 1)
            pareto["cum_skus"]     = pareto["sku_rank"] / len(pareto)               
            pareto["cum_skus_pct"] = pareto["cum_skus"] * 100
            return pareto                    
        except Exception as e:
            logger.error(f'[SpeedTest]Error for prepare_pareto: {e}
{traceback.format_exc()}')

    def abc_xyz(self):
            daily = self.sales.groupby(["ITEM", "DAY"], as_index=False)["QTY"].sum()
            stats = (
                daily.groupby("ITEM")["QTY"]
                .agg(
                    qty_total="sum",
                    mean_qty="mean",
                    std_qty="std"
                )
                .reset_index()
            )
            stats["cv_qty"] = stats["std_qty"] / stats["mean_qty"].replace(0, np.nan)
            total_qty = stats["qty_total"].sum()
            stats["share_qty_pct"] = (stats["qty_total"] / total_qty * 100).round(2)
            stats = stats.sort_values("share_qty_pct", ascending=False).reset_index(drop=True)
            stats["cum_share"] = stats["share_qty_pct"].cumsum().round(2)
            def classify(cum):
                if cum <= 5:
                    return "A"
                elif cum <= 20:
                    return "B"
                else:
                    return "C"
            stats["ABC"] = stats["cum_share"].apply(classify)
            return stats

    def convert_numpy(self, obj):
        if isinstance(obj, dict):
            return {k: self.convert_numpy(v) for k, v in obj.items()}
        elif isinstance(obj, list):
            return [self.convert_numpy(v) for v in obj]
        elif isinstance(obj, (np.integer, int)):
            return int(obj)
        elif isinstance(obj, (np.floating, float)):
            return float(obj)
        else:
            return obj

    async def process_pareto(self):
        """Main processing function that calls all other methods in order."""
        self.processing()
        outputs = self.prepare_pareto()
        return outputs

    async def process_abcxyz(self):
        """Main processing function that calls all other methods in order."""
        self.processing()
        outputs = self.abc_xyz().fillna(0)
        logger.info(f"[SpeedTest]: ABC-XYZ analysis completed {outputs}.")
        return outputs

这些端点准备就绪后，即可开始测试。

结果如上所示：

• 纯JavaScript工作流：整个过程在略多于11.7秒内完成。

大部分时间花费在更新工作表和在n8n节点内执行迭代计算。

• FastAPI支持的工作流：等效的“外包”过程在约11.0秒内完成。

繁重的计算任务被卸载到Python微服务，其处理速度比原生JavaScript节点更快。

换句话说，将复杂计算外包给Python实际上可以提高性能。原因是FastAPI端点直接执行优化后的Python函数，而n8n内部的JavaScript节点必须通过循环进行迭代。对于大型数据集，性能差异可能会更大，不容忽视。

结论

这表明，通过使用小段JavaScript代码，可以在n8n中进行简单的数据处理。然而，供应链分析产品可能需要更高级的处理，涉及优化和高级统计库。

为此，客户可以接受“黑盒”方法，正如这篇Towards Data Science文章中介绍的生产计划工作流所示。

但对于轻量级处理任务，可以将其集成到工作流中，为非代码用户提供可见性。

在另一个项目中，n8n被用于连接供应链IT系统以进行采购订单的电子数据交换（EDI）。

这个为一家小型物流公司部署的工作流，完全使用JavaScript节点解析EDI消息。

如本教程所示，电子数据交换消息的解析完全通过JavaScript节点完成。这有助于提高解决方案的健壮性，并通过将维护工作移交给客户来减轻工作量。

那么，最佳方法是什么？

n8n应作为核心分析产品的编排与集成工具。这些分析产品需要特定的输入格式，可能与客户的数据不完全匹配。因此，建议使用JavaScript代码节点来执行这种预处理。

例如，上述工作流将Google Sheet（包含输入数据）连接到运行配送计划优化算法的FastAPI微服务。其目的是将优化算法集成到配送规划人员用于组织门店配送的Google Sheet中。

JavaScript代码节点用于将从Google Sheet收集的数据转换为算法所需的输入格式。通过在工作流内部完成这些工作，它仍然受控于在自己实例中运行工作流的客户。而优化部分可以保留在托管于服务商实例上的微服务中。

为了更好地理解此设置，可以观看这个简短的演示。希望本教程及上述示例能让读者对n8n在数据分析方面的应用有所了解。欢迎分享对这种方法的评论，以及对如何改进工作流性能的看法。