数据加工与分析

场景描述

在销售数据分析场景中，需要统计截止某日期前的订单明细，按退货标志和订单状态分组计算汇总指标，包括总数量、基础价格总额、折扣后价格总额、含税价格总额，以及各指标的平均值和订单数。通过Aura DataFrame的链式操作，可以依次完成筛选、派生列计算、分组聚合和排序，实现与SQL等价的数据加工流程，为销售报表和经营决策提供数据支撑。

前提条件

• 已创建表并写入数据。具体操作，请参见创建表并写入数据。
• 运行时环境要求Python版本为3.11。

示例

本示例对sales_line_items表执行筛选、派生列计算、分组聚合和排序等Aura DataFrame操作，展示完整的数据加工与分析流程。

1. 导入依赖并连接AuraJobV2端点。
2. 加载sales_line_items数据集。
3. 筛选ship_date在截止日期之前的记录。
4. 计算折扣价和含税价格等派生列。
5. 按return_flag和line_status分组聚合（sum、avg、count等）。
6. 按return_flag和line_status排序。
7. 执行查询并打印结果。
8. 关闭AuraJobV2端点连接。

import os
from datetime import datetime, timedelta
from aura_frame.multimodal import ai_lake

target_database = "test"  #在LakeFormation中创建数据库，例如：test。具体操作，请参见创建LakeFormation实例并规划元数据
#连接AuraJobV2端点，具体参数说明，请参见建立与AuraJobV2端点的连接。
con = ai_lake.connect(
    aura_endpoint=os.getenv("aura_endpoint"),         # AuraJobV2端点所在的终端节点（Endpoint）
    aura_endpoint_id=os.getenv("aura_endpoint_id"),    # AuraJobV2端点ID
    aura_endpoint_name=os.getenv("aura_endpoint_name"),# AuraJobV2端点名称
    aura_workspace_id=os.getenv("aura_workspace_id"),  # 工作空间ID
    lf_catalog_name=os.getenv("lf_catalog_name"),      # LakeFormation Catalog名称
    access_key=os.getenv("access_key"),                # 访问密钥ID（AK）
    secret_key=os.getenv("secret_key"),                # 访问密钥（SK）
    default_database=target_database,                       # LakeFormation数据库名称
    use_single_cn_mode=True,   # 是否开启单CN模式
    logging_level=logging.INFO # 设置日志级别
)
try:
    # 加载数据集
    ds = con.load_dataset("sales_line_items", database=target_database)
    cutoff = datetime(1998, 12, 1) - timedelta(days=90)
    # 筛选：过滤不满足条件的行
    q = ds.filter(expr=ds.ship_date <= cutoff)
    # 计算列表达式
    discount_price = ds.unit_price * (1 - ds.discount_rate)
    charge = discount_price * (1 + ds.tax_rate)
    # 分组聚合
    q = q.groupby(["return_flag", "line_status"])
    q = q.aggregate(
        sum_qty=ds.quantity.sum(),
        sum_base_price=ds.unit_price.sum(),
        sum_disc_price=discount_price.sum(),
        sum_charge=charge.sum(),
        avg_qty=ds.quantity.mean(),
        avg_price=ds.unit_price.mean(),
        avg_disc=ds.discount_rate.mean(),
        count_order=ds.count(),
    )
    # 排序
    q = q.order_by(["return_flag", "line_status"])
    # 触发执行
    df = q.execute()
    print(df)
finally:
    con.close()

相关文档

• Session APIs
• Dataset APIs
• Table APIs