Palantir Foundry Function - 累计收入分析

Question

我想写一个函数如下：

原始数据：

customer	product	revenue
Customer A	Product 1	EUR 10
Customer A	Product 2	EUR 10
Customer B	Product 1	EUR 5
Customer B	Product 2	EUR 2
Customer C	Product 1	EUR 5

目标数据：

customer	revenue	cumulative revenue
Customer A	EUR 20	EUR 20
Customer B	EUR 7	EUR 27
Customer C	EUR 5	EUR 32

我确切地知道如何在 PySpark 中做到这一点，但不熟悉带有功能的打字稿，因为我想在前端触发“即时”计算。

这是 PySpark 代码：

from pyspark.sql import functions as F, window as W
window = W.Window.partitionBy(F.col("helper")).orderBy(F.col("net_revenue").desc())
df = ( df .groupby("customer") .agg( F.sum("net_revenue").alias("net_revenue") ) .withColumn('helper', F.lit(1)) .withColumn( "cumulative_revenue", F.sum("net_revenue").over(window) ) )

您能告诉我如何编写那段功能代码吗？

Answer 1

Functions 目前不支持此工作流的快速简洁方式。 .groupBy()、.sum() 和 .segmentBy() 函数无法帮助您完成您正在尝试做的事情。

此外，我们有意将字符串（在您的情况下为 customer）上的分组结果数限制为 1000 个。

解决此问题的最佳方法是遍历原始数据的每一行，创建一个地图来跟踪自定义的收入和累积收入。

之后（如果我理解您想要正确执行的操作），您可以编辑每个计算收入所引用的对象，例如 data.revenue = newRevenue。

获取代码示例，让您了解如何执行此操作：

interface CustomerRevenueData {
    revenue: Integer;
    cumulativeRevenue: Integer;
}

@Function()
public cumulativeRevenueAnalysis(): void {
    const data = Objects().search().initialDataObject().all();

    // The string key represents the customer name
    const customRevenueData = new FunctionsMap<string, CustomerRevenueData>();

   data.forEach(dataPoint => {
       // your logic here, updating the revenue/cumulative revenue for each customer you find
   })
}