-- 設(shè)置Flink執(zhí)行參數(shù)
SET execution.checkpointing.interval=30000;-- 30秒一次檢查點(diǎn)
SET execution.checkpointing.timeout =60000;-- 檢查點(diǎn)超時(shí)時(shí)間
SET execution.checkpointing.mode= EXACTLY_ONCE;-- 精確一次語(yǔ)義
SET state.backend ='rocksdb';-- 使用RocksDB作為狀態(tài)后端
SET state.ttl.ttl =86400000;-- 狀態(tài)保留1天
SET parallelism.default=12;-- 默認(rèn)并行度12

-- 定義訂單狀態(tài)枚舉類型
CREATETYPE OrderStatus ASENUM(
'PENDING_PAYMENT',-- 待付款
'PAID',-- 已付款
'PROCESSING',-- 處理中
'SHIPPED',-- 已發(fā)貨
'DELIVERED',-- 已送達(dá)
'CANCELLED',-- 已取消
'REFUNDED',-- 已退款
'EXPIRED'-- 已過(guò)期
);

-- 定義庫(kù)存操作類型枚舉
CREATETYPE InventoryOpType ASENUM(
'INCREASE',-- 增加庫(kù)存
'DECREASE',-- 減少庫(kù)存
'FREEZE',-- 凍結(jié)庫(kù)存
'UNFREEZE',-- 解凍庫(kù)存
'ADJUST'-- 調(diào)整庫(kù)存
);

-- 定義物流調(diào)度狀態(tài)枚舉
CREATETYPE LogisticsStatus ASENUM(
'PENDING_DISPATCH',-- 待調(diào)度
'DISPATCHED',-- 已調(diào)度
'IN_TRANSIT',-- 運(yùn)輸中
'OUT_FOR_DELIVERY',-- 配送中
'DELIVERED',-- 已送達(dá)
'FAILED'-- 配送失敗
);

-- 訂單狀態(tài)變更事件流
CREATETABLE order_status_events (
  order_id STRING,-- 訂單ID
  user_id STRING,-- 用戶ID
status OrderStatus,-- 訂單狀態(tài)
  prev_status OrderStatus,-- 上一狀態(tài)
  status_time TIMESTAMP(3),-- 狀態(tài)變更時(shí)間
  payment_time TIMESTAMP(3),-- 支付時(shí)間(如有)
  cancel_reason STRING,-- 取消原因(如有)
  operation_user STRING,-- 操作人
  ext_info MAP<STRING, STRING>,-- 擴(kuò)展信息
  event_time AS PROCTIME(),-- 處理時(shí)間
  WATERMARK FOR status_time AS status_time -INTERVAL'5'SECOND-- 水印定義，允許5秒延遲
)WITH(
'connector'='kafka',
'topic'='order_status_events',
'properties.bootstrap.servers'='kafka-broker:9092',
'properties.group.id'='flink_order_tracking_group',
'format'='json',
'json.fail-on-missing-field'='false',
'json.ignore-parse-errors'='true',
'scan.startup.mode'='earliest-offset',
'properties.auto.offset.reset'='earliest'
);

-- 訂單商品明細(xì)流
CREATETABLE order_item_events (
  order_id STRING,-- 訂單ID
  item_id STRING,-- 商品ID
  sku_id STRING,-- SKU ID
  product_name STRING,-- 商品名稱
  quantity INT,-- 數(shù)量
  price DECIMAL(10,2),-- 單價(jià)
  discount DECIMAL(10,2),-- 折扣金額
  create_time TIMESTAMP(3),-- 創(chuàng)建時(shí)間
  WATERMARK FOR create_time AS create_time -INTERVAL'5'SECOND
)WITH(
'connector'='kafka',
'topic'='order_item_events',
'properties.bootstrap.servers'='kafka-broker:9092',
'properties.group.id'='flink_order_item_group',
'format'='json'
);

-- 商品庫(kù)存基礎(chǔ)表 (CDC方式接入)
CREATETABLE product_inventory (
  product_id STRING,-- 商品ID
  sku_id STRING,-- SKU ID
  total_stock INT,-- 總庫(kù)存
  available_stock INT,-- 可用庫(kù)存
  frozen_stock INT,-- 凍結(jié)庫(kù)存
  locked_stock INT,-- 鎖定庫(kù)存
  update_time TIMESTAMP(3),-- 更新時(shí)間
PRIMARYKEY(product_id, sku_id)NOT ENFORCED
)WITH(
'connector'='mysql-cdc',
'hostname'='mysql-inventory',
'port'='3306',
'username'='flink_user',
'password'='flink_password',
'database-name'='inventory_db',
'table-name'='product_inventory',
'server-time-zone'='Asia/Shanghai'
);

-- 物流單事件流
CREATETABLE logistics_events (
  logistics_id STRING,-- 物流單ID
  order_id STRING,-- 訂單ID
status LogisticsStatus,-- 物流狀態(tài)
  status_time TIMESTAMP(3),-- 狀態(tài)時(shí)間
  location STRING,-- 當(dāng)前位置
  courier_id STRING,-- 快遞員ID
  courier_name STRING,-- 快遞員姓名
  WATERMARK FOR status_time AS status_time -INTERVAL'10'SECOND
)WITH(
'connector'='kafka',
'topic'='logistics_events',
'properties.bootstrap.servers'='kafka-broker:9092',
'properties.group.id'='flink_logistics_group',
'format'='json'
);

-- 物流區(qū)域信息表 (HBase)
CREATETABLE logistics_area_info (
  area_id STRING,-- 區(qū)域ID
  province STRING,-- 省份
  city STRING,-- 城市
  district STRING,-- 區(qū)/縣
  warehouse_id STRING,-- 對(duì)應(yīng)倉(cāng)庫(kù)ID
PRIMARYKEY(area_id)NOT ENFORCED
)WITH(
'connector'='hbase-2.2',
'table-name'='logistics:area_info',
'zookeeper.quorum'='zk-node1,zk-node2,zk-node3',
'zookeeper.znode.parent'='/hbase'
);

-- 創(chuàng)建訂單狀態(tài)事件清洗視圖
CREATEVIEW cleaned_order_status_events AS
SELECT
  order_id,
  user_id,
status,
  prev_status,
  status_time,
  payment_time,
-- 標(biāo)準(zhǔn)化取消原因
CASE
WHENstatus='CANCELLED'THEN
CASE
WHEN cancel_reason ISNULLOR cancel_reason =''THEN'UNKNOWN'
WHEN cancel_reason IN('user_cancel','用戶取消')THEN'USER_CANCEL'
WHEN cancel_reason IN('stock_out','庫(kù)存不足')THEN'STOCK_OUT'
WHEN cancel_reason IN('payment_timeout','支付超時(shí)')THEN'PAYMENT_TIMEOUT'
ELSE'OTHER'
END
ELSENULL
ENDAS cancel_reason_standardized,
  operation_user,
  ext_info,
-- 添加訂單創(chuàng)建時(shí)間(首次狀態(tài)變更)
  FIRST_VALUE(status_time)OVER(PARTITIONBY order_id ORDERBY status_time)AS create_time,
-- 計(jì)算狀態(tài)持續(xù)時(shí)間(與上一狀態(tài)比較)
  status_time - LAG(status_time)OVER(PARTITIONBY order_id ORDERBY status_time)AS status_duration,
  event_time
FROM order_status_events;

-- 訂單狀態(tài)生命周期表 (使用狀態(tài)函數(shù)追蹤完整生命周期)
CREATETABLE order_lifecycle (
  order_id STRING,
  user_id STRING,
  create_time TIMESTAMP(3),
  pending_payment_time TIMESTAMP(3),
  paid_time TIMESTAMP(3),
  processing_time TIMESTAMP(3),
  shipped_time TIMESTAMP(3),
  delivered_time TIMESTAMP(3),
  cancelled_time TIMESTAMP(3),
  refunded_time TIMESTAMP(3),
  expired_time TIMESTAMP(3),
  cancel_reason STRING,
  current_status OrderStatus,
  status_updated_time TIMESTAMP(3),
-- 各狀態(tài)持續(xù)時(shí)間
  pending_payment_duration BIGINT,
  processing_duration BIGINT,
  shipping_duration BIGINT,
  delivery_duration BIGINT,
  overall_duration BIGINT,
  last_updated AS PROCTIME()
)WITH(
'connector'='upsert-kafka',
'topic'='order_lifecycle',
'properties.bootstrap.servers'='kafka-broker:9092',
'key.format'='json',
'key.json.ignore-parse-errors'='true',
'value.format'='json',
'value.json.fail-on-missing-field'='false'
);

-- 寫(xiě)入訂單生命周期表
INSERTINTO order_lifecycle
SELECT
  order_id,
  user_id,
MAX(create_time)AS create_time,
MAX(CASEWHENstatus='PENDING_PAYMENT'THEN status_time END)AS pending_payment_time,
MAX(CASEWHENstatus='PAID'THEN status_time END)AS paid_time,
MAX(CASEWHENstatus='PROCESSING'THEN status_time END)AS processing_time,
MAX(CASEWHENstatus='SHIPPED'THEN status_time END)AS shipped_time,
MAX(CASEWHENstatus='DELIVERED'THEN status_time END)AS delivered_time,
MAX(CASEWHENstatus='CANCELLED'THEN status_time END)AS cancelled_time,
MAX(CASEWHENstatus='REFUNDED'THEN status_time END)AS refunded_time,
MAX(CASEWHENstatus='EXPIRED'THEN status_time END)AS expired_time,
MAX(CASEWHENstatus='CANCELLED'THEN cancel_reason_standardized END)AS cancel_reason,
  LAST_VALUE(status)OVER(PARTITIONBY order_id ORDERBY status_time ROWSBETWEENUNBOUNDEDPRECEDINGANDUNBOUNDEDFOLLOWING)AS current_status,
MAX(status_time)AS status_updated_time,
-- 計(jì)算各狀態(tài)持續(xù)時(shí)間(秒)
  TIMESTAMPDIFF(SECOND,MAX(CASEWHENstatus='PENDING_PAYMENT'THEN status_time END),
MAX(CASEWHENstatus='PAID'THEN status_time END))AS pending_payment_duration,
  TIMESTAMPDIFF(SECOND,MAX(CASEWHENstatus='PAID'THEN status_time END),
MAX(CASEWHENstatus='SHIPPED'THEN status_time END))AS processing_duration,
  TIMESTAMPDIFF(SECOND,MAX(CASEWHENstatus='SHIPPED'THEN status_time END),
MAX(CASEWHENstatus='OUT_FOR_DELIVERY'THEN status_time END))AS shipping_duration,
  TIMESTAMPDIFF(SECOND,MAX(CASEWHENstatus='OUT_FOR_DELIVERY'THEN status_time END),
MAX(CASEWHENstatus='DELIVERED'THEN status_time END))AS delivery_duration,
  TIMESTAMPDIFF(SECOND,MAX(create_time),MAX(status_time))AS overall_duration
FROM cleaned_order_status_events
GROUPBY order_id, user_id;

-- 創(chuàng)建庫(kù)存操作事件視圖
CREATEVIEW inventory_operation_events AS
WITH order_item_agg AS(
-- 聚合訂單商品信息
SELECT
    order_id,
    COLLECT_LIST(ROW(sku_id, quantity))AS items,
MAX(create_time)AS create_time
FROM order_item_events
GROUPBY order_id
)
-- 生成庫(kù)存操作事件
SELECT
  UUID()AS op_id,-- 操作ID
  o.order_id,
  UNNEST(items).sku_id AS sku_id,
  UNNEST(items).quantity AS quantity,
-- 根據(jù)訂單狀態(tài)確定庫(kù)存操作類型
CASE
WHEN o.status='PAID'THEN'FREEZE'-- 支付成功，凍結(jié)庫(kù)存
WHEN o.status='SHIPPED'THEN'DECREASE'-- 已發(fā)貨，減少庫(kù)存
WHEN o.status='CANCELLED'AND o.prev_status IN('PAID','PROCESSING')THEN'UNFREEZE'-- 已取消，解凍庫(kù)存
WHEN o.status='REFUNDED'THEN'INCREASE'-- 已退款，增加庫(kù)存
ELSENULL
ENDAS op_type,
  o.status_time AS op_time,
'ORDER_SYSTEM'AS source_system,
  o.event_time
FROM cleaned_order_status_events o
JOIN order_item_agg oi ON o.order_id = oi.order_id
-- 過(guò)濾出需要庫(kù)存操作的狀態(tài)變更
WHERE
(o.status='PAID'AND o.prev_status ='PENDING_PAYMENT')OR
(o.status='SHIPPED'AND o.prev_status ='PROCESSING')OR
(o.status='CANCELLED'AND o.prev_status IN('PAID','PROCESSING'))OR
(o.status='REFUNDED');

-- 創(chuàng)建庫(kù)存更新結(jié)果表
CREATETABLE inventory_update_results (
  op_id STRING,
  order_id STRING,
  sku_id STRING,
  op_type InventoryOpType,
  quantity INT,
  prev_available_stock INT,
  new_available_stock INT,
  prev_frozen_stock INT,
  new_frozen_stock INT,
  op_time TIMESTAMP(3),
  process_time TIMESTAMP(3),
status STRING,-- SUCCESS, FAILED, RETRY
  message STRING,
PRIMARYKEY(op_id)NOT ENFORCED
)WITH(
'connector'='jdbc',
'url'='jdbc:mysql://mysql-inventory:3306/inventory_db',
'table-name'='inventory_update_results',
'username'='flink_user',
'password'='flink_password',
'sink.buffer-flush.max-rows'='100',
'sink.buffer-flush.interval'='5s',
'sink.max-retries'='3'
);

-- 庫(kù)存更新主邏輯
INSERTINTO inventory_update_results
SELECT
  op_id,
  order_id,
  sku_id,
  op_type,
  quantity,
  prev_available,
  new_available,
  prev_frozen,
  new_frozen,
  op_time,
CURRENT_TIMESTAMPAS process_time,
CASE
WHEN(op_type ='DECREASE'AND prev_available < quantity)THEN'FAILED'
WHEN(op_type ='FREEZE'AND prev_available < quantity)THEN'FAILED'
ELSE'SUCCESS'
ENDASstatus,
CASE
WHEN(op_type ='DECREASE'AND prev_available < quantity)THEN'Insufficient stock'
WHEN(op_type ='FREEZE'AND prev_available < quantity)THEN'Insufficient stock to freeze'
ELSE'Operation successful'
ENDAS message
FROM(
-- 使用Flink的狀態(tài)函數(shù)進(jìn)行庫(kù)存原子更新
SELECT
    op_id,
    order_id,
    sku_id,
    op_type,
    quantity,
    op_time,
-- 根據(jù)操作類型計(jì)算新庫(kù)存值
CASE op_type
WHEN'INCREASE'THEN available_stock + quantity
WHEN'DECREASE'THEN available_stock - quantity
WHEN'FREEZE'THEN available_stock - quantity
WHEN'UNFREEZE'THEN available_stock + quantity
ELSE available_stock
ENDAS new_available,
    available_stock AS prev_available,
-- 處理凍結(jié)庫(kù)存
CASE op_type
WHEN'FREEZE'THEN frozen_stock + quantity
WHEN'UNFREEZE'THEN frozen_stock - quantity
ELSE frozen_stock
ENDAS new_frozen,
    frozen_stock AS prev_frozen
FROM inventory_operation_events
-- 關(guān)聯(lián)當(dāng)前庫(kù)存信息
JOIN product_inventory FOR SYSTEM_TIME ASOF event_time
ON inventory_operation_events.sku_id = product_inventory.sku_id
) t
-- 過(guò)濾掉無(wú)效操作類型
WHERE op_type ISNOTNULL;

-- 創(chuàng)建物流調(diào)度指令表
CREATETABLE logistics_dispatch_commands (
  command_id STRING,
  order_id STRING,
  user_id STRING,
  sku_id STRING,
  quantity INT,
  warehouse_id STRING,
  target_province STRING,
  target_city STRING,
  target_district STRING,
  target_address STRING,
  required_delivery_time TIMESTAMP(3),
  priority STRING,-- HIGH, MEDIUM, LOW
  create_time TIMESTAMP(3),
status STRING,-- PENDING, DISPATCHED, FAILED
PRIMARYKEY(command_id)NOT ENFORCED
)WITH(
'connector'='kafka',
'topic'='logistics_dispatch_commands',
'properties.bootstrap.servers'='kafka-broker:9092',
'key.format'='json',
'value.format'='json',
'sink.partitioner'='round-robin'
);

-- 物流調(diào)度觸發(fā)邏輯
INSERTINTO logistics_dispatch_commands
SELECT
  UUID()AS command_id,
  o.order_id,
  o.user_id,
  oi.sku_id,
  oi.quantity,
  la.warehouse_id,
  u.province,
  u.city,
  u.district,
  u.address,
-- 計(jì)算期望送達(dá)時(shí)間(根據(jù)商品類型)
CASE
WHEN p.category ='fresh'THEN o.status_time +INTERVAL'24'HOUR
WHEN p.category ='digital'THEN o.status_time +INTERVAL'48'HOUR
ELSE o.status_time +INTERVAL'72'HOUR
ENDAS required_delivery_time,
-- 根據(jù)訂單金額確定優(yōu)先級(jí)
CASE
WHENSUM(oi.quantity * oi.price)>1000THEN'HIGH'
WHENSUM(oi.quantity * oi.price)>500THEN'MEDIUM'
ELSE'LOW'
ENDOVER(PARTITIONBY o.order_id)AS priority,
  o.status_time AS create_time,
'PENDING'ASstatus
FROM cleaned_order_status_events o
-- 關(guān)聯(lián)訂單商品信息
JOIN order_item_events oi ON o.order_id = oi.order_id
-- 關(guān)聯(lián)用戶收貨地址
JOIN user_address FOR SYSTEM_TIME ASOF o.event_time
ON o.user_id = user_address.user_id 
AND user_address.is_default =TRUE
-- 關(guān)聯(lián)商品信息獲取分類
JOIN product_info FOR SYSTEM_TIME ASOF o.event_time
ON oi.sku_id = product_info.sku_id
-- 關(guān)聯(lián)物流區(qū)域信息獲取最優(yōu)倉(cāng)庫(kù)
JOIN logistics_area_info la 
ON user_address.district = la.district
-- 僅處理已支付待發(fā)貨的訂單
WHERE o.status='PAID'
AND o.prev_status ='PENDING_PAYMENT'
-- 添加冪等性控制，防止重復(fù)調(diào)度
ANDNOTEXISTS(
SELECT1FROM logistics_dispatch_commands 
WHERE order_id = o.order_id ANDstatus!='FAILED'
);

-- 創(chuàng)建物流時(shí)效監(jiān)控視圖
CREATEVIEW logistics_efficiency_metrics AS
WITH order_logistics AS(
-- 關(guān)聯(lián)訂單與物流信息
SELECT
    o.order_id,
    o.status_time AS paid_time,
    l.logistics_id,
MIN(CASEWHEN l.status='DISPATCHED'THEN l.status_time END)AS dispatched_time,
MIN(CASEWHEN l.status='IN_TRANSIT'THEN l.status_time END)AS transit_time,
MIN(CASEWHEN l.status='OUT_FOR_DELIVERY'THEN l.status_time END)AS delivery_time,
MIN(CASEWHEN l.status='DELIVERED'THEN l.status_time END)AS received_time,
    la.warehouse_id,
    la.city AS warehouse_city,
    u.city AS target_city
FROM cleaned_order_status_events o
LEFTJOIN logistics_events l ON o.order_id = l.order_id
LEFTJOIN logistics_area_info la ON l.location = la.area_id
LEFTJOIN user_address u ON o.user_id = u.user_id AND u.is_default =TRUE
WHERE o.status='PAID'
GROUPBY o.order_id, o.status_time, l.logistics_id, la.warehouse_id, la.city, u.city
)
-- 計(jì)算各環(huán)節(jié)時(shí)效指標(biāo)
SELECT
  order_id,
  logistics_id,
  warehouse_id,
  warehouse_city,
  target_city,
  paid_time,
  dispatched_time,
  transit_time,
  delivery_time,
  received_time,
-- 計(jì)算各階段耗時(shí)(分鐘)
  TIMESTAMPDIFF(MINUTE, paid_time, dispatched_time)AS warehouse_processing_minutes,
  TIMESTAMPDIFF(MINUTE, dispatched_time, transit_time)AS first_mile_minutes,
  TIMESTAMPDIFF(MINUTE, transit_time, delivery_time)AS line_haul_minutes,
  TIMESTAMPDIFF(MINUTE, delivery_time, received_time)AS last_mile_minutes,
  TIMESTAMPDIFF(MINUTE, paid_time, received_time)AS total_delivery_minutes,
-- 判斷是否超時(shí)
CASE
WHEN TIMESTAMPDIFF(MINUTE, paid_time, received_time)>1440THEN'OVERDUE'-- >24小時(shí)
WHEN TIMESTAMPDIFF(MINUTE, paid_time, received_time)>720THEN'AT_RISK'-- >12小時(shí)
ELSE'ON_TIME'
ENDAS delivery_status
FROM order_logistics;

-- 創(chuàng)建物流效率監(jiān)控結(jié)果表
CREATETABLE logistics_efficiency_monitor (
  order_id STRING,
  warehouse_id STRING,
  warehouse_city STRING,
  target_city STRING,
  total_delivery_minutes INT,
  delivery_status STRING,
  warehouse_processing_minutes INT,
  first_mile_minutes INT,
  line_haul_minutes INT,
  last_mile_minutes INT,
  monitoring_time TIMESTAMP(3),
PRIMARYKEY(order_id)NOT ENFORCED
)WITH(
'connector'='elasticsearch-7',
'hosts'='http://es-node1:9200,http://es-node2:9200',
'index'='logistics_efficiency_{yyyyMMdd}',
'document-id.key-delimiter'='$',
'sink.bulk-flush.max-actions'='1000',
'sink.bulk-flush.max-size'='2mb',
'sink.bulk-flush.interval'='10s',
'format'='json'
);

-- 寫(xiě)入物流效率監(jiān)控?cái)?shù)據(jù)
INSERTINTO logistics_efficiency_monitor
SELECT
  order_id,
  warehouse_id,
  warehouse_city,
  target_city,
  total_delivery_minutes,
  delivery_status,
  warehouse_processing_minutes,
  first_mile_minutes,
  line_haul_minutes,
  last_mile_minutes,
CURRENT_TIMESTAMPAS monitoring_time
FROM logistics_efficiency_metrics
WHERE received_time ISNOTNULL;-- 僅處理已收貨訂單

-- 創(chuàng)建訂單異常檢測(cè)視圖
CREATEVIEW abnormal_order_detection AS
SELECT
  order_id,
  user_id,
  current_status,
  status_time,
  create_time,
-- 計(jì)算訂單各階段超時(shí)情況
CASE
WHEN current_status ='PENDING_PAYMENT'
AND TIMESTAMPDIFF(MINUTE, create_time,CURRENT_TIMESTAMP)>30THEN'PAYMENT_TIMEOUT'
WHEN current_status ='PROCESSING'
AND TIMESTAMPDIFF(HOUR, paid_time,CURRENT_TIMESTAMP)>24THEN'PROCESSING_TIMEOUT'
WHEN current_status ='SHIPPED'
AND TIMESTAMPDIFF(HOUR, shipped_time,CURRENT_TIMESTAMP)>72THEN'DELIVERY_TIMEOUT'
ELSENULL
ENDAS abnormal_type,
-- 計(jì)算超時(shí)時(shí)間(分鐘)
CASE
WHEN current_status ='PENDING_PAYMENT'
THEN TIMESTAMPDIFF(MINUTE, create_time,CURRENT_TIMESTAMP)-30
WHEN current_status ='PROCESSING'
THEN TIMESTAMPDIFF(MINUTE, paid_time,CURRENT_TIMESTAMP)-1440
WHEN current_status ='SHIPPED'
THEN TIMESTAMPDIFF(MINUTE, shipped_time,CURRENT_TIMESTAMP)-4320
ELSE0
ENDAS overtime_minutes,
-- 獲取用戶歷史異常訂單數(shù)
(SELECTCOUNT(*)FROM order_lifecycle 
WHERE user_id = o.user_id AND abnormal_type ISNOTNULL)AS user_abnormal_count,
CURRENT_TIMESTAMPAS detection_time
FROM order_lifecycle o
-- 檢測(cè)異常條件
WHERE(
(current_status ='PENDING_PAYMENT'
AND TIMESTAMPDIFF(MINUTE, create_time,CURRENT_TIMESTAMP)>30)
OR
(current_status ='PROCESSING'
AND TIMESTAMPDIFF(HOUR, paid_time,CURRENT_TIMESTAMP)>24)
OR
(current_status ='SHIPPED'
AND TIMESTAMPDIFF(HOUR, shipped_time,CURRENT_TIMESTAMP)>72)
)
-- 排除已處理的異常訂單
AND order_id NOTIN(SELECT order_id FROM abnormal_order_handling);

-- 創(chuàng)建異常訂單處理指令表
CREATETABLE abnormal_order_handling (
  order_id STRING,
  abnormal_type STRING,
  overtime_minutes INT,
  user_abnormal_count INT,
  detection_time TIMESTAMP(3),
handler STRING,
  handling_action STRING,
  handling_time TIMESTAMP(3),
status STRING,-- PENDING, PROCESSED, RESOLVED
  notes STRING,
PRIMARYKEY(order_id)NOT ENFORCED
)WITH(
'connector'='jdbc',
'url'='jdbc:mysql://mysql-order:3306/order_db',
'table-name'='abnormal_order_handling',
'username'='flink_user',
'password'='flink_password'
);

-- 自動(dòng)生成異常訂單處理指令
INSERTINTO abnormal_order_handling
SELECT
  order_id,
  abnormal_type,
  overtime_minutes,
  user_abnormal_count,
  detection_time,
-- 根據(jù)異常類型和用戶歷史異常數(shù)分配處理人員
CASE
WHEN user_abnormal_count >5THEN'vip_customer_service'
WHEN abnormal_type ='DELIVERY_TIMEOUT'THEN'logistics_support'
ELSE'order_support'
ENDAShandler,
-- 自動(dòng)建議處理動(dòng)作
CASE
WHEN abnormal_type ='PAYMENT_TIMEOUT'THEN'CANCEL_ORDER'
WHEN abnormal_type ='PROCESSING_TIMEOUT'THEN'ESCALATE_PROCESSING'
WHEN abnormal_type ='DELIVERY_TIMEOUT'THEN'CHECK_LOGISTICS'
ENDAS handling_action,
CURRENT_TIMESTAMPAS handling_time,
'PENDING'ASstatus,
CASE
WHEN user_abnormal_count >5THEN'High-risk customer, manual review required'
ELSE'Auto-generated handling instruction'
ENDAS notes
FROM abnormal_order_detection;

-- 創(chuàng)建訂單處理實(shí)時(shí)指標(biāo)表
CREATETABLE order_processing_metrics (
  metric_time TIMESTAMP(3),
  order_count BIGINT,
  paid_count BIGINT,
  shipped_count BIGINT,
  delivered_count BIGINT,
  cancelled_count BIGINT,
  avg_payment_time DOUBLE,
  avg_processing_time DOUBLE,
  avg_delivery_time DOUBLE,
  abnormal_order_rate DOUBLE,
PRIMARYKEY(metric_time)NOT ENFORCED
)WITH(
'connector'='prometheus',
'url'='http://prometheus-server:9090/api/v1/write',
'namespace'='ecommerce',
'metric.name'='order_processing_metrics'
);

-- 計(jì)算并輸出訂單處理指標(biāo)
INSERTINTO order_processing_metrics
SELECT
  TUMBLE_START(status_time,INTERVAL'5'MINUTE)AS metric_time,
COUNT(DISTINCT order_id)AS order_count,
COUNT(DISTINCTCASEWHENstatus='PAID'THEN order_id END)AS paid_count,
COUNT(DISTINCTCASEWHENstatus='SHIPPED'THEN order_id END)AS shipped_count,
COUNT(DISTINCTCASEWHENstatus='DELIVERED'THEN order_id END)AS delivered_count,
COUNT(DISTINCTCASEWHENstatus='CANCELLED'THEN order_id END)AS cancelled_count,
-- 計(jì)算平均支付時(shí)間(秒)
AVG(TIMESTAMPDIFF(SECOND, create_time, paid_time))AS avg_payment_time,
-- 計(jì)算平均處理時(shí)間(秒)
AVG(TIMESTAMPDIFF(SECOND, paid_time, shipped_time))AS avg_processing_time,
-- 計(jì)算平均配送時(shí)間(秒)
AVG(TIMESTAMPDIFF(SECOND, shipped_time, delivered_time))AS avg_delivery_time,
-- 異常訂單率
CASEWHENCOUNT(DISTINCT order_id)=0THEN0
ELSECOUNT(DISTINCTCASEWHEN abnormal_type ISNOTNULLTHEN order_id END)*1.0/COUNT(DISTINCT order_id)
ENDAS abnormal_order_rate
FROM order_lifecycle
LEFTJOIN abnormal_order_detection a ON order_lifecycle.order_id = a.order_id
-- 使用5分鐘滾動(dòng)窗口聚合
GROUPBY TUMBLE(status_time,INTERVAL'5'MINUTE);

-- 創(chuàng)建訂單狀態(tài)變更通知表(Kafka)
CREATETABLE order_status_notifications (
  notification_id STRING,
  order_id STRING,
  user_id STRING,
status OrderStatus,
  prev_status OrderStatus,
  status_time TIMESTAMP(3),
  notification_type STRING,-- APP_PUSH, SMS, EMAIL
  message_content STRING,
  priority INT,
  create_time TIMESTAMP(3),
statusAS'PENDING',
PRIMARYKEY(notification_id)NOT ENFORCED
)WITH(
'connector'='kafka',
'topic'='order_status_notifications',
'properties.bootstrap.servers'='kafka-broker:9092',
'key.format'='json',
'value.format'='json'
);

-- 生成訂單狀態(tài)變更通知
INSERTINTO order_status_notifications
SELECT
  UUID()AS notification_id,
  order_id,
  user_id,
status,
  prev_status,
  status_time,
-- 根據(jù)狀態(tài)類型確定通知方式
CASE
WHENstatusIN('CANCELLED','REFUNDED')THEN'SMS'
WHENstatus='DELIVERED'THEN'APP_PUSH'
ELSE'APP_PUSH'
ENDAS notification_type,
-- 動(dòng)態(tài)生成通知內(nèi)容
CASEstatus
WHEN'PAID'THEN CONCAT('Order ', order_id,' has been paid successfully')
WHEN'SHIPPED'THEN CONCAT('Order ', order_id,' has been shipped')
WHEN'DELIVERED'THEN CONCAT('Order ', order_id,' has been delivered')
WHEN'CANCELLED'THEN CONCAT('Order ', order_id,' has been cancelled: ', cancel_reason_standardized)
WHEN'REFUNDED'THEN CONCAT('Order ', order_id,' has been refunded')
ELSE CONCAT('Order ', order_id,' status updated to ',status)
ENDAS message_content,
-- 設(shè)置通知優(yōu)先級(jí)
CASE
WHENstatusIN('CANCELLED','REFUNDED')THEN1
WHENstatusIN('PAID','DELIVERED')THEN2
ELSE3
ENDAS priority,
CURRENT_TIMESTAMPAS create_time
FROM cleaned_order_status_events
-- 僅對(duì)關(guān)鍵狀態(tài)變更發(fā)送通知
WHEREstatusIN('PAID','SHIPPED','DELIVERED','CANCELLED','REFUNDED');

-- 創(chuàng)建帶狀態(tài)TTL優(yōu)化的訂單狀態(tài)視圖
CREATEVIEW order_status_with_ttl AS
SELECT
  order_id,
  user_id,
status,
  status_time,
  ROW_NUMBER()OVER(PARTITIONBY order_id ORDERBY status_time DESC)AS rn
FROM order_status_events
-- 使用Flink的狀態(tài)TTL功能自動(dòng)清理過(guò)期狀態(tài)
WITH(
'state.ttl'='86400000',-- 狀態(tài)保留1天
'state.cleanup-strategy'='EMBEDDED'
)
WHERE rn =1;-- 只保留最新?tīng)顟B(tài)

-- 優(yōu)化的訂單與庫(kù)存雙流JOIN
CREATEVIEW order_inventory_joined AS
SELECT
/*+ OPTIONS('lookup.join.cache.ttl'='30s', 'lookup.join.cache.size'='10000') */
  o.order_id,
  o.status,
  o.status_time,
  oi.sku_id,
  oi.quantity,
  pi.available_stock,
  pi.frozen_stock,
  pi.total_stock
FROM cleaned_order_status_events o
JOIN order_item_events oi ON o.order_id = oi.order_id
-- 使用緩存優(yōu)化的LOOKUP JOIN
JOIN product_inventory FOR SYSTEM_TIME ASOF o.event_time
ON oi.sku_id = product_inventory.sku_id
WHERE o.status='PAID';

-- 創(chuàng)建規(guī)則配置表 (MySQL)
CREATETABLE order_processing_rules (
  rule_id STRING,
  rule_type STRING,-- INVENTORY_RULE, LOGISTICS_RULE, NOTIFICATION_RULE
  priority INT,
  condition_expr STRING,
  action_expr STRING,
  effective_time TIMESTAMP(3),
  expire_time TIMESTAMP(3),
status STRING,-- ACTIVE, INACTIVE
  create_time TIMESTAMP(3),
  update_time TIMESTAMP(3),
PRIMARYKEY(rule_id)NOT ENFORCED
)WITH(
'connector'='mysql-cdc',
'hostname'='mysql-config',
'port'='3306',
'username'='flink_user',
'password'='flink_password',
'database-name'='config_db',
'table-name'='order_processing_rules'
);

-- 使用動(dòng)態(tài)規(guī)則處理訂單
CREATEVIEW order_processing_with_rules AS
SELECT
  o.*,
  r.rule_id,
  r.action_expr
FROM cleaned_order_status_events o
JOIN order_processing_rules r
ON r.rule_type ='INVENTORY_RULE'
AND r.status='ACTIVE'
AND o.status_time >= r.effective_time
AND(r.expire_time ISNULLOR o.status_time < r.expire_time)
-- 這里可以集成Flink的SQL函數(shù)來(lái)動(dòng)態(tài)評(píng)估規(guī)則條件
AND o.status='PAID';

-- 創(chuàng)建數(shù)據(jù)質(zhì)量監(jiān)控表
CREATETABLE data_quality_metrics (
  metric_time TIMESTAMP(3),
  source_table STRING,
  total_records BIGINT,
  null_order_id_count BIGINT,
  late_records_count BIGINT,
  schema_violation_count BIGINT,
  avg_processing_time DOUBLE,
  error_rate DOUBLE
)WITH(
'connector'='elasticsearch-7',
'hosts'='http://es-node1:9200,http://es-node2:9200',
'index'='data_quality_metrics_{yyyyMMdd}',
'format'='json'
);

-- 監(jiān)控訂單事件流數(shù)據(jù)質(zhì)量
INSERTINTO data_quality_metrics
SELECT
  TUMBLE_START(event_time,INTERVAL'1'MINUTE)AS metric_time,
'order_status_events'AS source_table,
COUNT(*)AS total_records,
COUNT(CASEWHEN order_id ISNULLTHEN1END)AS null_order_id_count,
COUNT(CASEWHEN status_time < event_time -INTERVAL'5'SECONDTHEN1END)AS late_records_count,
0AS schema_violation_count,-- 需要通過(guò)Flink的DDL驗(yàn)證配置獲取
AVG(TIMESTAMPDIFF(MILLISECOND, status_time, PROCTIME()))AS avg_processing_time,
CASEWHENCOUNT(*)=0THEN0
ELSECOUNT(CASEWHEN order_id ISNULLORstatusISNULLTHEN1END)*1.0/COUNT(*)
ENDAS error_rate
FROM order_status_events
GROUPBY TUMBLE(event_time,INTERVAL'1'MINUTE);

-- 啟用Flink的查詢監(jiān)控
SET'execution.profile.enabled'='true';
SET'execution.profile.sample-interval'='1000';
SET'execution.profile.delay'='0';

-- 創(chuàng)建查詢性能監(jiān)控表
CREATETABLE query_performance_metrics (
  query_id STRING,
  job_name STRING,
  start_time TIMESTAMP(3),
  end_time TIMESTAMP(3),
  duration_ms BIGINT,
  rows_read BIGINT,
  rows_written BIGINT,
  peak_memory_usage BIGINT,
  state_size BIGINT,
  backpressure_count INT
)WITH(
'connector'='jdbc',
'url'='jdbc:mysql://mysql-monitor:3306/monitor_db',
'table-name'='query_performance_metrics',
'username'='flink_user',
'password'='flink_password'
);