# 数据加工与分析
演示筛选、计算列、分组聚合、排序的完整Aura DataFrame处理流程。
#### 场景描述
在销售数据分析场景中，需要统计截止某日期前的订单明细，按退货标志和订单状态分组计算汇总指标，包括总数量、基础价格总额、折扣后价格总额、含税价格总额，以及各指标的平均值和订单数。通过Aura DataFrame的链式操作，可以依次完成筛选、派生列计算、分组聚合和排序，实现与SQL等价的数据加工流程，为销售报表和经营决策提供数据支撑。
#### 前提条件
- 已创建表并写入数据。具体操作，请参见[创建表并写入数据](https://support.huaweicloud.com/sdkreference-aidatalake/aidatalake_09_0030.html)。
- 运行时环境要求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](https://support.huaweicloud.com/sdkreference-aidatalake/aidatalake_061_0135.html)
- [Dataset APIs](https://support.huaweicloud.com/sdkreference-aidatalake/aidatalake_061_0132.html)
- [Table APIs](https://support.huaweicloud.com/sdkreference-aidatalake/aidatalake_061_0133.html)