【系统架构设计百科】数据湖与数据仓库：分析架构的演进路线

-- 星型模型示例：订单事实表
CREATE TABLE fact_orders (
    order_id        BIGINT PRIMARY KEY,
    customer_key    INT REFERENCES dim_customer(customer_key),
    product_key     INT REFERENCES dim_product(product_key),
    date_key        INT REFERENCES dim_date(date_key),
    store_key       INT REFERENCES dim_store(store_key),
    quantity        INT,
    unit_price      DECIMAL(10, 2),
    discount_amount DECIMAL(10, 2),
    total_amount    DECIMAL(12, 2)
);

-- 维度表：客户维度
CREATE TABLE dim_customer (
    customer_key    INT PRIMARY KEY,
    customer_id     VARCHAR(32),
    customer_name   VARCHAR(128),
    segment         VARCHAR(32),
    region          VARCHAR(64),
    registration_date DATE,
    -- 缓慢变化维度（SCD Type 2）
    effective_from  DATE,
    effective_to    DATE,
    is_current      BOOLEAN
);

-- 维度表：时间维度
CREATE TABLE dim_date (
    date_key        INT PRIMARY KEY,
    full_date       DATE,
    year            INT,
    quarter         INT,
    month           INT,
    week_of_year    INT,
    day_of_week     INT,
    is_holiday      BOOLEAN
);

-- Snowflake 中处理半结构化数据
CREATE TABLE raw_events (
    event_id    VARCHAR(64),
    event_time  TIMESTAMP,
    payload     VARIANT  -- 半结构化 JSON 数据
);

-- 查询 VARIANT 字段
SELECT
    event_id,
    payload:user_id::STRING AS user_id,
    payload:action::STRING AS action,
    payload:properties:page_url::STRING AS page_url
FROM raw_events
WHERE event_time >= DATEADD(hour, -1, CURRENT_TIMESTAMP());

# 数据沼泽的典型问题：读取到不一致的数据
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("DataSwampDemo").getOrCreate()

# 写入任务正在往这个路径写 Parquet 文件
# 如果读取任务在写入过程中执行，可能读到部分数据
df = spark.read.parquet("s3://data-lake/raw/events/2026/04/13/")

# 没有 schema 校验，上游改了字段名不会报错
# 只是查询结果变成 NULL
df.select("user_id", "event_type", "timestamp").show()
# 如果上游把 "user_id" 改成了 "userId"，
# 这里 user_id 列全是 NULL，但不会抛异常

# 批处理层：每日订单汇总（PySpark）
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder \
    .appName("BatchLayer-DailyOrderSummary") \
    .getOrCreate()

# 读取全量订单数据
orders = spark.read.parquet("s3://data-lake/raw/orders/")

# 计算每日、每地区的订单汇总
daily_summary = orders \
    .filter(F.col("order_status") == "completed") \
    .groupBy(
        F.to_date("order_time").alias("order_date"),
        F.col("region")
    ) \
    .agg(
        F.count("order_id").alias("order_count"),
        F.sum("total_amount").alias("total_revenue"),
        F.avg("total_amount").alias("avg_order_value"),
        F.countDistinct("customer_id").alias("unique_customers")
    )

# 写入批处理视图
daily_summary.write \
    .mode("overwrite") \
    .partitionBy("order_date") \
    .parquet("s3://data-lake/batch-views/daily-order-summary/")

// 速度层：实时订单计数（Flink Java）
public class RealtimeOrderCounter {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();

        // 从 Kafka 消费订单事件
        KafkaSource<OrderEvent> source = KafkaSource.<OrderEvent>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("order-events")
            .setGroupId("speed-layer")
            .setValueOnlyDeserializer(new OrderEventDeserializer())
            .build();

        DataStream<OrderEvent> orders = env.fromSource(
            source,
            WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(
                Duration.ofSeconds(10)
            ).withTimestampAssigner((event, ts) -> event.getEventTime()),
            "OrderSource"
        );

        // 按地区和 1 分钟窗口聚合
        DataStream<OrderSummary> realtimeSummary = orders
            .filter(e -> "completed".equals(e.getStatus()))
            .keyBy(OrderEvent::getRegion)
            .window(TumblingEventTimeWindows.of(Time.minutes(1)))
            .aggregate(new OrderAggregator());

        // 写入实时视图（HBase 或 Redis）
        realtimeSummary.addSink(new HBaseSink("realtime_order_summary"));

        env.execute("SpeedLayer-RealtimeOrderCounter");
    }
}

# 服务层：合并批处理和实时视图
import datetime
from typing import Dict

class LambdaServingLayer:
    def __init__(self, batch_store, realtime_store):
        self.batch_store = batch_store      # 批处理视图（如 Hive）
        self.realtime_store = realtime_store  # 实时视图（如 Redis）

    def query_order_summary(self, region: str, date: str) -> Dict:
        # 批处理视图的最新数据截止时间
        batch_cutoff = self.batch_store.get_latest_batch_time()

        # 从批处理视图获取基准数据
        batch_result = self.batch_store.query(
            region=region,
            date=date,
            up_to=batch_cutoff
        )

        # 从实时视图获取增量数据（批处理截止时间之后的数据）
        realtime_result = self.realtime_store.query(
            region=region,
            date=date,
            after=batch_cutoff
        )

        # 合并结果
        return {
            "region": region,
            "date": date,
            "order_count": batch_result["order_count"]
                         + realtime_result["order_count"],
            "total_revenue": batch_result["total_revenue"]
                           + realtime_result["total_revenue"],
            "data_source": "merged",
            "batch_cutoff": str(batch_cutoff),
            "freshness": str(datetime.datetime.now())
        }

# Kappa 架构的重放机制（Flink Python API）
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import KafkaSource, KafkaOffsetsInitializer

env = StreamExecutionEnvironment.get_execution_environment()

# 重放模式：从最早的 offset 开始消费
replay_source = KafkaSource.builder() \
    .set_bootstrap_servers("kafka:9092") \
    .set_topics("order-events") \
    .set_group_id("kappa-replay-v2") \
    .set_starting_offsets(KafkaOffsetsInitializer.earliest()) \
    .set_value_only_deserializer(OrderEventDeserializer()) \
    .build()

# 使用与实时处理完全相同的处理逻辑
orders = env.from_source(
    replay_source,
    WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_seconds(10)),
    "ReplaySource"
)

# 相同的处理逻辑，输出到新的目标表
result = process_orders(orders)  # 复用同一个函数
result.add_sink(create_sink("order_summary_v2"))

env.execute("Kappa-Replay-V2")

{
  "add": {
    "path": "part-00000-abc123.parquet",
    "size": 52428800,
    "partitionValues": {"order_date": "2026-04-13"},
    "modificationTime": 1744531200000,
    "dataChange": true,
    "stats": "{\"numRecords\":125000,\"minValues\":{\"order_id\":1},\"maxValues\":{\"order_id\":125000}}"
  }
}
{
  "remove": {
    "path": "part-00000-old456.parquet",
    "deletionTimestamp": 1744531200000,
    "dataChange": true
  }
}

# Delta Lake 的 ACID 事务示例
from delta.tables import DeltaTable
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("DeltaLakeDemo") \
    .config("spark.sql.extensions",
            "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog",
            "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# 原子性写入：要么全部成功，要么全部回滚
new_orders = spark.read.json("s3://staging/orders/2026-04-13/")
new_orders.write \
    .format("delta") \
    .mode("append") \
    .partitionBy("order_date") \
    .save("s3://data-lake/orders/")

# MERGE 操作：upsert（更新+插入）
delta_table = DeltaTable.forPath(spark, "s3://data-lake/orders/")

delta_table.alias("target").merge(
    new_orders.alias("source"),
    "target.order_id = source.order_id"
).whenMatchedUpdate(set={
    "order_status": "source.order_status",
    "updated_at": "source.updated_at"
}).whenNotMatchedInsertAll().execute()

# 时间旅行：查询历史版本的数据
historical_orders = spark.read \
    .format("delta") \
    .option("versionAsOf", 5) \
    .load("s3://data-lake/orders/")

# 或者按时间戳查询
yesterday_orders = spark.read \
    .format("delta") \
    .option("timestampAsOf", "2026-04-12T00:00:00") \
    .load("s3://data-lake/orders/")

# Schema 约束：阻止不匹配的写入
# 假设表的 schema 是 (order_id: long, amount: double, status: string)
bad_data = spark.createDataFrame([
    (1, "not_a_number", "completed")  # amount 应该是 double，这里是 string
], ["order_id", "amount", "status"])

# 这行会抛出 AnalysisException：
# 数据 schema 和表 schema 不匹配
bad_data.write.format("delta").mode("append") \
    .save("s3://data-lake/orders/")

# Schema 演进：显式允许新增列
new_data_with_extra_col = spark.createDataFrame([
    (1001, 99.99, "completed", "CN")
], ["order_id", "amount", "status", "country"])

new_data_with_extra_col.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .save("s3://data-lake/orders/")

-- Iceberg 分区演进示例（Spark SQL）

-- 创建表，初始按月分区
CREATE TABLE catalog.db.orders (
    order_id    BIGINT,
    order_time  TIMESTAMP,
    customer_id BIGINT,
    amount      DOUBLE,
    status      STRING
) USING iceberg
PARTITIONED BY (month(order_time));

-- 数据量增长后，改为按天分区
-- 历史数据不需要重写，新数据按天分区，旧数据仍按月分区
ALTER TABLE catalog.db.orders
    SET PARTITION SPEC (day(order_time));

-- 进一步细化：按小时分区（用于高频写入场景）
ALTER TABLE catalog.db.orders
    SET PARTITION SPEC (hour(order_time));

-- Iceberg 查询引擎会自动处理混合分区：
-- 扫描 2024 年数据时用月分区过滤
-- 扫描 2025 年数据时用天分区过滤
-- 扫描 2026 年数据时用小时分区过滤
SELECT * FROM catalog.db.orders
WHERE order_time BETWEEN '2026-04-01' AND '2026-04-13';

-- Hive 风格：用户必须知道分区结构
SELECT * FROM hive_orders
WHERE year = 2026 AND month = 4 AND day = 13;  -- 必须按分区列查询

-- Iceberg 风格：用户用原始列查询，引擎自动做分区裁剪
SELECT * FROM iceberg_orders
WHERE order_time >= '2026-04-13 00:00:00'
  AND order_time <  '2026-04-14 00:00:00';
-- Iceberg 自动将时间条件映射到分区，跳过无关文件

-- Flink SQL：同一段 SQL 可以跑在流模式和批模式下

-- 定义 Kafka 数据源（流模式）
CREATE TABLE order_events (
    order_id    BIGINT,
    customer_id BIGINT,
    amount      DOUBLE,
    event_time  TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'order-events',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format' = 'json',
    'scan.startup.mode' = 'latest-offset'
);

-- 定义 Iceberg 输出表
CREATE TABLE order_summary (
    window_start TIMESTAMP(3),
    region       STRING,
    order_count  BIGINT,
    total_amount DOUBLE
) WITH (
    'connector' = 'iceberg',
    'catalog-name' = 'lakehouse',
    'catalog-type' = 'hive',
    'warehouse' = 's3://data-lake/warehouse'
);

-- 同一段聚合逻辑：
-- 流模式下，每分钟输出一次增量结果
-- 批模式下，一次性处理全部历史数据
INSERT INTO order_summary
SELECT
    TUMBLE_START(event_time, INTERVAL '1' MINUTE) AS window_start,
    region,
    COUNT(order_id) AS order_count,
    SUM(amount) AS total_amount
FROM order_events
GROUP BY
    TUMBLE(event_time, INTERVAL '1' MINUTE),
    region;

# 流批一体的语义对齐：显式处理迟到数据
from pyflink.table import EnvironmentSettings, TableEnvironment

# 批模式设置
batch_settings = EnvironmentSettings.new_instance() \
    .in_batch_mode() \
    .build()
batch_env = TableEnvironment.create(batch_settings)

# 流模式设置
stream_settings = EnvironmentSettings.new_instance() \
    .in_streaming_mode() \
    .build()
stream_env = TableEnvironment.create(stream_settings)

# 关键：在两种模式下使用完全相同的 SQL
# 但需要额外配置来保证语义一致
UNIFIED_SQL = """
    SELECT
        window_start,
        region,
        COUNT(*) AS order_count,
        SUM(amount) AS total_amount
    FROM TABLE(
        TUMBLE(TABLE order_events, DESCRIPTOR(event_time), INTERVAL '1' HOUR)
    )
    GROUP BY window_start, region
"""

# 批模式执行
batch_env.execute_sql(UNIFIED_SQL)

# 流模式执行（需要额外配置水位线和允许延迟）
stream_env.get_config().set(
    "table.exec.emit.early-fire.enabled", "true"
)
stream_env.get_config().set(
    "table.exec.emit.early-fire.delay", "60s"
)
stream_env.execute_sql(UNIFIED_SQL)

// Flink + Iceberg 的 Exactly-Once 写入配置
StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment();

// 开启 Checkpoint，间隔 60 秒
env.enableCheckpointing(60000);
env.getCheckpointConfig().setCheckpointingMode(
    CheckpointingMode.EXACTLY_ONCE
);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
env.getCheckpointConfig().setCheckpointTimeout(300000);

// Iceberg Sink 配合 Checkpoint 实现 Exactly-Once
// 每次 Checkpoint 完成时，Iceberg Sink 将暂存的数据文件
// 原子性地提交到 Iceberg 表
FlinkSink.forRowData(dataStream)
    .table(icebergTable)
    .tableLoader(tableLoader)
    .writeParallelism(4)
    .upsert(true)  // 开启 upsert 模式
    .equalityFieldColumns(Arrays.asList("order_id"))
    .append();

env.execute("ExactlyOnce-Iceberg-Sink");

-- Delta Lake 的 OPTIMIZE 命令：合并小文件
OPTIMIZE delta.`s3://data-lake/orders/`
WHERE order_date >= '2026-04-01';

-- Z-ORDER 优化：按查询模式重新组织数据布局
OPTIMIZE delta.`s3://data-lake/orders/`
ZORDER BY (customer_id, order_date);

-- Iceberg 的 Compaction（通过 Spark 调用）
-- 重写小文件，合并为更大的文件
CALL catalog.system.rewrite_data_files(
    table => 'db.orders',
    strategy => 'sort',
    sort_order => 'order_date ASC, customer_id ASC',
    options => map(
        'target-file-size-bytes', '134217728',  -- 128MB
        'min-file-size-bytes', '67108864',       -- 64MB
        'max-file-size-bytes', '201326592'       -- 192MB
    )
);

-- 清理过期快照，释放被删除文件的存储空间
CALL catalog.system.expire_snapshots(
    table => 'db.orders',
    older_than => TIMESTAMP '2026-04-06 00:00:00',
    retain_last => 10
);

# 数据契约定义示例（data-contract.yaml）
apiVersion: datacontract/v1.0
kind: DataContract
metadata:
  name: order-events-contract
  owner: order-service-team
  consumers:
    - analytics-team
    - recommendation-team

schema:
  type: object
  properties:
    order_id:
      type: integer
      description: "订单唯一标识"
      constraints:
        - not_null
        - unique
    customer_id:
      type: integer
      description: "客户 ID"
      constraints:
        - not_null
        - "references: customer_master.customer_id"
    amount:
      type: number
      description: "订单金额（单位：元）"
      constraints:
        - not_null
        - "range: [0.01, 10000000]"
    order_time:
      type: timestamp
      description: "下单时间"
      constraints:
        - not_null
        - "freshness: <= 5 minutes"
    status:
      type: string
      description: "订单状态"
      constraints:
        - not_null
        - "enum: [created, paid, shipped, completed, cancelled]"

quality:
  completeness:
    threshold: 99.5%      # 非 NULL 字段的完整率不低于 99.5%
  uniqueness:
    columns: [order_id]
    threshold: 100%        # 主键必须完全唯一
  freshness:
    max_delay: 5 minutes   # 数据延迟不超过 5 分钟
  volume:
    min_records_per_hour: 1000   # 每小时至少 1000 条记录
    max_records_per_hour: 500000 # 每小时不超过 50 万条

sla:
  availability: 99.9%
  notification:
    channels: [slack, pagerduty]
    on_violation: block_pipeline  # 违反契约时阻断管道

# 基于 Great Expectations 的数据质量校验
import great_expectations as gx

# 创建数据上下文
context = gx.get_context()

# 定义数据源
datasource = context.data_sources.add_spark("lakehouse_source")
data_asset = datasource.add_dataframe_asset("order_events")

# 定义期望套件（Expectation Suite）
suite = context.suites.add(
    gx.ExpectationSuite(name="order_events_quality")
)

# 添加数据质量规则
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="amount",
        min_value=0.01,
        max_value=10000000
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeInSet(
        column="status",
        value_set=["created", "paid", "shipped", "completed", "cancelled"]
    )
)
suite.add_expectation(
    gx.expectations.ExpectTableRowCountToBeBetween(
        min_value=1000,
        max_value=500000
    )
)

# 运行校验
batch = data_asset.add_batch_definition_whole_dataframe(
    "daily_batch"
).get_batch(batch_parameters={"dataframe": order_df})

validation_result = batch.validate(suite)

# 根据校验结果决定是否继续管道
if not validation_result.success:
    failed_expectations = [
        r for r in validation_result.results if not r.success
    ]
    raise DataQualityException(
        f"数据质量校验失败：{len(failed_expectations)} 项不合格",
        details=failed_expectations
    )

# 基于 OpenLineage 的血缘追踪集成
# OpenLineage 是一个开放的血缘追踪标准
# 支持 Spark、Flink、Airflow 等引擎

# Spark 配置 OpenLineage
spark = SparkSession.builder \
    .appName("OrderETL") \
    .config("spark.extraListeners",
            "io.openlineage.spark.agent.OpenLineageSparkListener") \
    .config("spark.openlineage.transport.type", "http") \
    .config("spark.openlineage.transport.url",
            "http://lineage-server:5000") \
    .config("spark.openlineage.namespace", "production") \
    .getOrCreate()

# 正常编写 ETL 逻辑
# OpenLineage 自动捕获输入/输出/转换信息
raw_orders = spark.read.format("delta") \
    .load("s3://data-lake/raw/orders/")

cleaned_orders = raw_orders \
    .filter(F.col("amount") > 0) \
    .filter(F.col("status").isNotNull()) \
    .withColumn("amount_usd",
                F.col("amount") * F.col("exchange_rate"))

cleaned_orders.write.format("delta") \
    .mode("overwrite") \
    .save("s3://data-lake/curated/orders/")

# OpenLineage 自动记录：
# 输入：s3://data-lake/raw/orders/
# 输出：s3://data-lake/curated/orders/
# 转换：filter(amount > 0), filter(status IS NOT NULL),
#        withColumn(amount_usd = amount * exchange_rate)

# 阶段一：Hive 表迁移到 Iceberg
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("MigrateToIceberg") \
    .config("spark.sql.catalog.lakehouse", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.lakehouse.type", "hive") \
    .config("spark.sql.catalog.lakehouse.uri", "thrift://metastore:9083") \
    .getOrCreate()

# 从 Hive 表读取数据
hive_orders = spark.sql("SELECT * FROM hive_db.orders")

# 创建 Iceberg 表
spark.sql("""
    CREATE TABLE lakehouse.analytics.orders (
        order_id        BIGINT,
        customer_id     BIGINT,
        order_time      TIMESTAMP,
        amount          DOUBLE,
        refund_amount   DOUBLE,
        status          STRING,
        region          STRING
    ) USING iceberg
    PARTITIONED BY (days(order_time))
    TBLPROPERTIES (
        'write.format.default' = 'parquet',
        'write.parquet.compression-codec' = 'zstd',
        'write.target-file-size-bytes' = '134217728'
    )
""")

# 数据迁移
hive_orders.writeTo("lakehouse.analytics.orders").append()

# 数据一致性校验
iceberg_count = spark.sql(
    "SELECT COUNT(*) FROM lakehouse.analytics.orders"
).collect()[0][0]
hive_count = spark.sql(
    "SELECT COUNT(*) FROM hive_db.orders"
).collect()[0][0]

assert iceberg_count == hive_count, \
    f"数据量不一致：Iceberg={iceberg_count}, Hive={hive_count}"

-- 阶段二：统一的 Flink SQL（流批共用）
-- 这段 SQL 既可以在流模式下运行（实时消费 Kafka）
-- 也可以在批模式下运行（全量扫描 Iceberg 表）

-- 流模式数据源
CREATE TABLE kafka_orders (
    order_id        BIGINT,
    customer_id     BIGINT,
    order_time      TIMESTAMP(3),
    amount          DOUBLE,
    refund_amount   DOUBLE,
    status          STRING,
    region          STRING,
    WATERMARK FOR order_time AS order_time - INTERVAL '30' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'order-events',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format' = 'json'
);

-- 批模式数据源（同一张 Iceberg 表）
CREATE TABLE iceberg_orders (
    order_id        BIGINT,
    customer_id     BIGINT,
    order_time      TIMESTAMP(3),
    amount          DOUBLE,
    refund_amount   DOUBLE,
    status          STRING,
    region          STRING
) WITH (
    'connector' = 'iceberg',
    'catalog-name' = 'lakehouse',
    'catalog-type' = 'hive',
    'warehouse' = 's3://data-lake/warehouse'
);

-- 统一的计算逻辑（核心：用同一段 SQL）
-- 关键修改：refund_amount 使用 COALESCE 处理异步到达
INSERT INTO iceberg_order_summary
SELECT
    TUMBLE_START(order_time, INTERVAL '1' HOUR) AS window_start,
    region,
    COUNT(*) AS order_count,
    SUM(CASE
        WHEN status = 'completed'
             AND COALESCE(refund_amount, 0) = 0
        THEN amount
        ELSE 0
    END) AS effective_revenue,
    SUM(amount) AS total_revenue
FROM kafka_orders  -- 流模式用 kafka_orders，批模式换成 iceberg_orders
GROUP BY
    TUMBLE(order_time, INTERVAL '1' HOUR),
    region;

-- 最简方案示例：PostgreSQL + dbt
-- 对于 GB 级数据，这可能就是全部所需

-- dbt 模型文件：models/order_summary.sql
{{ config(materialized='table', schema='analytics') }}

WITH completed_orders AS (
    SELECT
        DATE_TRUNC('day', order_time) AS order_date,
        region,
        COUNT(*) AS order_count,
        SUM(amount) AS total_revenue,
        SUM(CASE WHEN refund_amount = 0 THEN amount ELSE 0 END)
            AS effective_revenue,
        COUNT(DISTINCT customer_id) AS unique_customers
    FROM {{ source('raw', 'orders') }}
    WHERE status = 'completed'
    GROUP BY 1, 2
)

SELECT
    order_date,
    region,
    order_count,
    total_revenue,
    effective_revenue,
    unique_customers,
    effective_revenue / NULLIF(order_count, 0) AS avg_effective_value
FROM completed_orders