E-MapReduce：AI Function最佳實務

核心能力

本實踐主要使用以下 AI 內建函數：

方案優勢

• 資料不出庫：所有 AI 處理在 StarRocks 引擎內完成，無需將敏感金融資料匯出到外部系統。

• 純 SQL 操作：分析師無需學習 Python/Java，用熟悉的 SQL 即可完成 AI 分析。

• 批量處理能力：可對百萬級文本記錄批量執行 AI 函數，支援 ETL 管線整合。

• 與分析查詢無縫結合：AI 函數結果可直接參与 GROUP BY、JOIN、視窗函數等分析操作。

環境要求

• EMR Serverless StarRocks 2.1.1及以上版本。

• 已開通AI 中心。

資料準備

本實踐類比金融情境中的三類典型文本資料，建立樣本表並寫入樣本資料。

CREATE TABLE fin_customer_tickets (
    ticket_id       BIGINT,
    customer_id     VARCHAR(20),
    channel         VARCHAR(10)    COMMENT '渠道：app/web/phone/branch',
    product_type    VARCHAR(20)    COMMENT '產品類型：credit_card/loan/deposit/fund/insurance',
    ticket_text     VARCHAR(65533) COMMENT '工單內容',
    created_at      DATETIME
)
DUPLICATE KEY(ticket_id)
DISTRIBUTED BY HASH(ticket_id) BUCKETS 8;

INSERT INTO fin_customer_tickets VALUES
(1001, 'C20240001', 'app', 'credit_card',
 '你們這個信用卡賬單又出錯了！上個月明明還了8000塊，結果賬單顯示還欠5000，這是第三次了！再搞不定我就要投訴到銀保監了！', '2026-05-01 09:15:00'),
(1002, 'C20240002', 'phone', 'loan',
 '想諮詢一下房貸提前還款的流程，需要準備哪些材料？還有提前還款有沒有違約金？', '2026-05-01 10:30:00'),
(1003, 'C20240003', 'web', 'fund',
 '你們推薦的那個穩健型基金，一個月虧了15%，這叫穩健？我要求賠償！基金經理是不是在亂操作？', '2026-05-01 11:45:00'),
(1004, 'C20240004', 'app', 'deposit',
 '大額存單到期自動續存的功能很方便，利率也還行，比之前高了0.1個百分點，整體體驗不錯', '2026-05-01 14:00:00'),
(1005, 'C20240005', 'branch', 'insurance',
 '保險理賠太慢了，住院花了3萬塊，提交材料都兩個月了還沒給結果。催了好幾次客服就說在審核中。', '2026-05-01 15:20:00'),
(1006, 'C20240006', 'app', 'credit_card',
 '剛開通的白金卡權益不錯，機場貴賓廳很舒服，積分兌換也比較划算，推薦給朋友了', '2026-05-01 16:00:00'),
(1007, 'C20240007', 'phone', 'loan',
 '車貸每月還款日能不能改一下？現在是5號扣款，但我工資15號才發，每個月都很緊張', '2026-05-02 09:00:00'),
(1008, 'C20240008', 'web', 'fund',
 '定投了半年的指數基金，收益還可以，目前年化大概8%左右，準備繼續持有', '2026-05-02 10:15:00');

CREATE TABLE fin_regulatory_news (
    news_id         BIGINT,
    source          VARCHAR(50)    COMMENT '來源機構',
    publish_date    DATE,
    title           VARCHAR(500),
    content         VARCHAR(65533) COMMENT '本文內容'
)
DUPLICATE KEY(news_id)
DISTRIBUTED BY HASH(news_id) BUCKETS 4;

INSERT INTO fin_regulatory_news VALUES
(2001, '中國人民銀行', '2026-04-28',
 '關於加強金融消費者權益保護工作的通知',
 '為進一步加強金融消費者權益保護，規範金融機構經營行為，根據《中華人民共和國消費者權益保護法》等法律法規，現就有關事項通知如下：一、金融機構應當建立健全消費者權益保護工作機制，設立專門部門或崗位負責消費者權益保護工作。二、金融機構在營銷宣傳、產品銷售過程中，應當充分揭示產品風險，不得進行虛假或引人誤解的宣傳。三、金融機構應當暢通投訴渠道，及時處理消費者投訴，處理時限不得超過15個工作日。'),
(2002, '中國銀保監會', '2026-04-25',
 '關於規範信貸資金用途的監管要求',
 '近期檢查發現部分銀行信貸資金違規流入房地產市場和股票市場。現要求：各銀行業金融機構應加強貸後管理，確保信貸資金用於約定用途。對於經營性貸款，應核實借款人真實經營需求，防止資金挪用。違規機構將面臨監管處罰，包括但不限於罰款、限制業務範圍等措施。'),
(2003, '中國證監會', '2026-04-20',
 '關於進一步規範上市公司資訊披露的若干意見',
 '為提升上市公司資訊披露品質，保護投資者合法權益，現提出以下意見：上市公司應當真實、準確、完整、及時地披露資訊，不得有虛假記載、誤導性陳述或者重大遺漏。鼓勵上市公司自願披露與投資者決策相關的資訊，提高資訊透明度。'),
(2004, '中國人民銀行', '2026-04-15',
 '2026年第一季度貨幣政策執行報告',
 '2026年第一季度，穩健的貨幣政策精準有力。3月末，廣義貨幣M2餘額同比增長8.2%。人民幣貸款餘額同比增長10.5%，社會融資規模存量同比增長9.8%。利率市場化改革持續推進，企業貸款利率保持在較低水平。下一階段將繼續實施穩健的貨幣政策，保持流動性合理充裕。');

CREATE TABLE fin_contracts (
    contract_id     VARCHAR(30),
    contract_type   VARCHAR(20)   COMMENT '合約類型：loan/guarantee/pledge',
    party_a         VARCHAR(200),
    party_b         VARCHAR(200),
    contract_text   VARCHAR(65533) COMMENT '合約關鍵條款'
)
DUPLICATE KEY(contract_id)
DISTRIBUTED BY HASH(contract_id) BUCKETS 4;

INSERT INTO fin_contracts VALUES
('LOAN-2026-0001', 'loan', '某某銀行股份有限公司杭州分行', '杭州星辰科技有限公司',
 '貸款金額：人民幣伍佰萬元整（¥5,000,000.00）。貸款期限：自2026年1月15日起至2027年1月14日止，共計12個月。貸款利率：按照LPR加60個基點執行，即年利率4.15%。還款方式：等額本息，每月20日為還款日。違約條款：借款人未按時還款的，應按逾期金額每日萬分之五支付罰息。擔保方式：以借款人名下位於杭州市西湖區XX路XX號房產提供抵押擔保。'),
('LOAN-2026-0002', 'loan', '某某銀行股份有限公司上海分行', '上海錦繡貿易有限公司',
 '貸款金額：人民幣壹仟萬元整（¥10,000,000.00）。貸款期限：自2026年3月1日起至2028年2月28日止，共計24個月。貸款利率：固定利率年4.35%。還款方式：先息後本，每季度末支付利息，到期一次性歸還本金。違約條款：借款人連續三期未還息或挪用貸款資金的，貸款人有權宣布貸款提前到期。擔保方式：由上海錦繡集團有限公司提供連帶責任保證擔保。'),
('GUAR-2026-0001', 'guarantee', '某某銀行股份有限公司杭州分行', '浙江鑫達實業集團有限公司',
 '保證方式：連帶責任保證。保證範圍：主債權本金人民幣伍佰萬元及利息、罰息、違約金、實現債權的費用。保證期間：自主債務履行期限屆滿之日起兩年。保證人聲明：保證人具有完全民事行為能力，願意以其全部資產為上述債務提供連帶責任保證。連絡人：張偉，電話：1381234****，社會安全號碼：330102199001010000。');

情感分析——識別客戶情緒

快速判斷每條工單的客戶情緒，優先處理負面反饋：

SELECT
    ticket_id,
    product_type,
    ai_sentiment(ticket_text) AS sentiment,
    LEFT(ticket_text, 40) AS preview
FROM fin_customer_tickets
ORDER BY ticket_id;

結果樣本：

工單智能分類——按業務類型自動歸類

將自由文本工單自動歸入預定義的業務類別：

SELECT
    ticket_id,
    ai_classify(
        ticket_text,
        ARRAY['賬單爭議', '業務諮詢', '投訴建議', '產品體驗', '理賠糾紛']
    ) AS category,
    ai_sentiment(ticket_text) AS sentiment
FROM fin_customer_tickets;

情感分布分析——按產品維度彙總

將 AI 分析結果直接與 SQL 彙總分析結合：

SELECT
    product_type,
    ai_sentiment(ticket_text) AS sentiment,
    COUNT(*) AS ticket_count
FROM fin_customer_tickets
GROUP BY product_type, ai_sentiment(ticket_text)
ORDER BY product_type, ticket_count DESC;

結果樣本：

語義過濾——篩查高風險工單

使用 ai_filter 在 WHERE 子句中按語義條件過濾，找出涉及監管投訴風險的工單：

SELECT
    ticket_id,
    product_type,
    ticket_text,
    ai_sentiment(ticket_text) AS sentiment
FROM fin_customer_tickets
WHERE ai_filter(ticket_text, '客戶威脅要向監管機構投訴或要求賠償');

公告分類

對監管公告按監管領域自動分類：

SELECT
    news_id,
    source,
    title,
    ai_classify(
        content,
        ARRAY['消費者保護', '信貸監管', '資訊披露', '貨幣政策', '反洗錢', '市場准入']
    ) AS reg_category
FROM fin_regulatory_news;

公告摘要

為長篇公告產生簡短摘要，便於管理層快速探索：

SELECT
    news_id,
    title,
    ai_summarize(content) AS summary
FROM fin_regulatory_news
WHERE publish_date >= '2026-04-01';

合規影響分析

使用 ai_complete 進行更深層次的業務分析——判斷公告對本行業務的影響：

SELECT
    news_id,
    title,
    ai_complete(
        content,
        '你是一名銀行合規分析師。請分析這條監管公告對商業銀行零售業務的具體影響，列出需要整改的要點，限100字以內。'
    ) AS compliance_impact
FROM fin_regulatory_news
WHERE ai_filter(content, '涉及銀行業務合規要求或處罰措施');

合約要素提取

從合約文本中自動抽取結構化的關鍵資訊：

SELECT
    contract_id,
    contract_type,
    ai_extract(
        contract_text,
        ARRAY['貸款金額', '貸款期限', '年利率', '還款方式', '擔保方式', '違約條款']
    ) AS key_elements
FROM fin_contracts
WHERE contract_type = 'loan';

結果樣本（JSON 格式）：

{
  "貸款金額": "人民幣伍佰萬元整（¥5,000,000.00）",
  "貸款期限": "2026年1月15日至2027年1月14日，共計12個月",
  "年利率": "LPR加60個基點，即4.15%",
  "還款方式": "等額本息，每月20日還款",
  "擔保方式": "杭州市西湖區房產抵押擔保",
  "違約條款": "逾期按每日萬分之五支付罰息"
}

合約 PII 脫敏

對包含個人敏感資訊的合約文本進行脫敏處理，滿足資料合規要求：

SELECT
    contract_id,
    ai_redact(contract_text) AS redacted_text
FROM fin_contracts
WHERE contract_type = 'guarantee';

脫敏效果：

原文：連絡人：張偉，電話：13812345678，社會安全號碼：330102199001010000
脫敏後：連絡人：張**，電話：138****5678，社會安全號碼：330102********0000

合約風險評估

結合 ai_extract 和 ai_complete 構建合約風險評估管線：

WITH contract_elements AS (
    SELECT
        contract_id,
        party_a,
        party_b,
        contract_text,
        ai_extract(
            contract_text,
            ARRAY['貸款金額', '年利率', '擔保方式', '違約條款']
        ) AS elements
    FROM fin_contracts
    WHERE contract_type = 'loan'
)
SELECT
    contract_id,
    party_b,
    elements,
    ai_complete(
        CONCAT('合約要素：', CAST(elements AS VARCHAR), '\n\n合約原文：', contract_text),
        '你是信貸風控專家。請基於合約要素評估該筆貸款的風險等級（低/中/高），並給出主要風險點，限80字。'
    ) AS risk_assessment
FROM contract_elements;

客戶 360 文本分析看板

將多個 AI 函數組合，構建客戶回函的多維分析視圖：

SELECT
    ticket_id,
    customer_id,
    product_type,
    channel,
    ai_sentiment(ticket_text)  AS sentiment,
    ai_classify(
        ticket_text,
        ARRAY['賬單爭議', '業務諮詢', '投訴建議', '產品體驗', '理賠糾紛']
    ) AS category,
    ai_extract(
        ticket_text,
        ARRAY['涉及金額', '訴求']
    ) AS key_info,
    ai_summarize(ticket_text)  AS summary
FROM fin_customer_tickets
WHERE created_at >= '2026-05-01';

基於 AI 分析結果構建物化視圖

將 AI 分析結果持久化，避免重複計算：

CREATE MATERIALIZED VIEW mv_ticket_analysis AS
SELECT
    ticket_id,
    customer_id,
    product_type,
    channel,
    created_at,
    ai_sentiment(ticket_text) AS sentiment,
    ai_classify(
        ticket_text,
        ARRAY['賬單爭議', '業務諮詢', '投訴建議', '產品體驗', '理賠糾紛']
    ) AS category
FROM fin_customer_tickets;

後續查詢直接基於物化視圖，無需重複調用 AI 函數：

SELECT
    DATE(created_at) AS dt,
    product_type,
    COUNT(*) AS negative_tickets
FROM mv_ticket_analysis
WHERE sentiment = 'negative'
GROUP BY DATE(created_at), product_type
ORDER BY dt, negative_tickets DESC;

ETL 管線整合：INSERT INTO SELECT

將 AI 處理結果寫入下遊分析表，構建完整的資料加工管線：

-- 建立分析結果表
CREATE TABLE fin_ticket_analysis_result (
    ticket_id       BIGINT,
    customer_id     VARCHAR(20),
    product_type    VARCHAR(20),
    sentiment       VARCHAR(20),
    category        JSON,
    key_info        JSON,
    summary         VARCHAR(65533),
    analyzed_at     DATETIME DEFAULT CURRENT_TIMESTAMP
)
DUPLICATE KEY(ticket_id)
DISTRIBUTED BY HASH(ticket_id) BUCKETS 4;

-- 批量 AI 分析並寫入結果表
INSERT INTO fin_ticket_analysis_result
    (ticket_id, customer_id, product_type, sentiment, category, key_info, summary)
SELECT
    ticket_id,
    customer_id,
    product_type,
    ai_sentiment(ticket_text),
    ai_classify(ticket_text, ARRAY['賬單爭議', '業務諮詢', '投訴建議', '產品體驗', '理賠糾紛']),
    ai_extract(ticket_text, ARRAY['涉及金額', '訴求']),
    ai_summarize(ticket_text)
FROM fin_customer_tickets
WHERE ticket_id NOT IN (SELECT ticket_id FROM fin_ticket_analysis_result);