在数据仓库里用 AI 函数：六大落地场景

用纯 SQL 就能跑的六个 AI 场景——替代你今天在用的 notebook、API 和 pipeline。

by Srikant Das 和 Ismail Makhlouf

在绝大多数组织中，数据仓库存放结构化数据，而非结构化数据则留在数据湖里。这套方式对分析型负载很合适：分析负载大规模消费结构化数据，日复一日地为一批已知报表提供支撑。

但 AI 负载需要不同的输入。AI 模型往往要解析非结构化数据——比如评论、客服工单、PDF——再把它们和结构化数据结合起来，用于训练、构建和部署模型。于是，一个想对客服工单做情感分析的分析师，不得不把这些行数据运到某个外部服务，等预测结果回来，再手工把它们拼回一张表里。这很慢，一旦 schema 变更就会崩，还会引入不必要的安全和治理风险。

AI Functions 的做法是把 AI 直接带到你的数据面前，而不是把你的数据搬到一个独立的 AI 环境里。你在标准 SQL 查询里调用模型，让整个推理过程都留在你既有的 pipeline 和 Unity Catalog 治理体系之内。这种架构从根本上改变了你在数据仓库里使用 AI 的方式：

• 默认即治理：因为 AI Functions 遵守 Unity Catalog 的权限，你的数据始终安全且私密。模型只能访问你明确授权的数据。
• SQL 原生，简单至上：只要你会写 SELECT，你就能用 AI 做开发。Databricks 帮你打理了复杂的部分——规划、并行化、重试，所以你不用操心集群管理或外部编排。对百万行跑推理和对一行跑推理一样简单，同一个查询无需改写即可横向扩展。
• 统一计费：告别对账一堆零散 dashboard 的麻烦。AI 用量会出现在 system.billing.usage 里，和你标准的 Databricks SQL 数据仓库费用并排展示。
• 专用函数：花更少的钱，拿到更好的结果。通过使用面向具体任务的函数——例如 ai_classify、ai_extract、ai_translate、ai_parse_document——你用的是为特定任务量身打造的模型，而不是为通用推理多花冤枉钱。

你可以在 Databricks 的任何地方使用这些 AI 函数，包括 notebook、Lakeflow Spark Declarative Pipelines 和 Workflow。但在这篇文章里，我们要聚焦的是在 Databricks Lakehouse 里专门调用这些函数。下面这些用例会向你展示，如何把这些 AI 函数集成进那些需要把数据仓库里的结构化数据，与数据仓库之外、或者通过 GenAI 赋能的函数自己产出的非结构化数据结合起来的负载中。

用例 1：文档智能，从原始文件到结构化行

ai_parse_document 充当摄取桥梁，把 PDF 或图片这类原始二进制文件内容转换成可读文本。解析完成后，ai_extract 负责做细粒度的键值抽取。这种组合方案，免去了脆弱的自定义 OCR pipeline 或第三方解析服务——后者常常在 schema 变更时崩掉。

在这个用例里，我们把 ai_parse_document 指向一个存放发票的 Databricks volume。这些发票被解析后，AI parse document 会以 JSON 形式产出结果，随后传给 ai_extract 函数，我们在其中定义想要从发票中抽取哪些实体。最终得到一张结构化表，里面是我们想从发票中抽取出的字段。

SELECT
  document_path,
  ai_extract(
    parsed_content,
    '["vendor_name","invoice_no","invoice_date","total_amount","currency","line_items"]'
  ) AS extracted
FROM (
  SELECT
    path AS document_path,
    ai_parse_document(content) AS parsed_content
  FROM read_files('s3://invoices/inbox/', format => 'binaryFile')
);

数据血缘现在从原始 PDF 一直贯通到抽取出的行，全部发生在同一个查询计划里。过去人们手工搭建的那座桥——一个 Python OCR 服务、一次 LLM 调用、再加一步 JSON 展开——全部坍缩进了这条查询。

Demo notebook: Document intelligence

用例 2：客户反馈情感分析

ai_classify 函数执行零样本分类，把自由文本反馈映射到一组用户自定义标签上，无需训练模型。这个过程把混乱、非结构化的文本，转化为受治理、可查询的列，让情感和主题数据能够立即服务于 BI 仪表盘和高管报告。

在这个例子里，我们要把 bronze.nps_responses 表里的客户评价，分类为正面、负面、中性和混合。

SELECT
  response_id,
  customer_id,
  survey_channel,
  ai_classify(
    verbatim_text,
    '["positive","negative","neutral","mixed"]'
  ):response[0]::STRING AS sentiment,
  ai_classify(
    verbatim_text,
    '["pricing_complaint","feature_request","performance_issue","support_experience","cancel_signal"]'
  ):response[0]::STRING AS feedback_topic,
  ai_classify(
    verbatim_text,
    '["high_urgency","medium_urgency","low_urgency"]'
  ):response[0]::STRING AS urgency_level
FROM bronze.nps_responses
WHERE survey_date >= current_date - 30;

Demo notebook: Sentiment analysis

用例 3：多语言数据的行内翻译

借助 ai_translate，你可以直接在查询层把多语言数据归一化到单一目标语言。这避免了数据孤岛和碎片化，让所有下游分析（包括分类和抽取）都能同时作用于整个全球数据集，而不是只处理英文切片。

在这个例子里，我们先从不同客户评价中抽取情感，再把它们翻译成英文。

SELECT
  review_id,
  product_id,
  source_locale,
  ai_translate(review_text, 'en') AS review_en,
  CAST(ai_extract(
    ai_translate(review_text, 'en'),
    '["overall_sentiment","packaging_issue","delivery_issue","customer_service_mentioned","would_repurchase"]'
  ) AS STRING) AS attributes
FROM demo_product_reviews;

Demo notebook: Translation and normalization

用例 4：大规模分类与路由

聚焦运营效率，ai_classify 把自由格式的输入（如客服工单或通话记录）转化为可操作的类别。通过在摄取环节就识别出传入反馈的意图和紧急程度，它能实现自动化的智能路由，把工单派给合适的团队或自动应答系统。

在下面的用例里，我们从一张表里摄取不同的客服工单，然后用 ai_classify 来判断工单的用户意图和紧急程度。

SELECT
  ticket_id,
  customer_segment,
  ai_classify(
    ticket_body,
    '["billing","outage","product_question","cancel_request","praise"]'
  ):response[0]::STRING AS intent,
  ai_classify(
    ticket_body,
    '["high","medium","low"]'
  ):response[0]::STRING AS urgency
FROM bronze.support_tickets;

Demo notebook: Classification and routing

用例 5：用 ai_extract 做销售通话的结构化抽取

ai_extract 函数专为从长文本（如销售通话记录）中挖掘半结构化信息而设计，能把叙述性文本转换成离散的结构化字段。它的价值很大：把定性信息直接送进 BI 工具， effectively 把口头对话变成可查询的指标，比如成交阶段和风险标记。

在这个用例里，我们挖掘一段很长的记录，识别出下一步动作、成交阶段、风险标记和风险原因，这样销售人员就能对产生这段记录的那场会议结果采取行动。

SELECT
  call_id,
  account_id,
  call_date,
  ai_extract(
    transcript,
    '["next_step","owner","deal_stage","risk_flag","risk_reason"]'
  ) AS facts
FROM gold.call_transcripts
WHERE call_date >= current_date - 7;

Demo notebook: Sales-call extraction

用例 6：用 ai_query 做生成式起草

ai_query 是最通用的函数，也是其余所有函数的基础：它允许你向任何你有权访问的、Databricks 托管的基础模型 serving endpoint 发送 prompt，并为每一行返回模型的回答。

在这个用例里，我们可以用 ai_query 为虚构的 gold.renewal_signals 表里每一个客户账户起草一封续约触达邮件——这张表告诉我们哪些账户已临近续约。

SELECT
  account_id,
  account_name,
  account_owner_email,
  ai_query(
    'databricks-gpt-oss-120b',
    CONCAT(
      'You are drafting a renewal-outreach email. Return a JSON object with exactly three fields: subject, body, suggested_send_date. ',
      'subject: one short subject line under 70 characters, no greeting. ',
      'body: the email body, 4 to 8 sentences, addressed to the account owner by first name, referencing the open risks and the recent engagement. ',
      'suggested_send_date: a date string in YYYY-MM-DD format within the next 14 days; pick sooner dates when days_to_renewal is short or open_risk_flags are active. ',
      'Tone: professional, concise, no marketing fluff. ',
      'Account context: name=', account_name,
      ', mrr_usd=', mrr_usd,
      ', days_to_renewal=', days_to_renewal,
      ', open_risk_flags=', open_risk_flags,
      ', last_engagement_summary=', last_engagement_summary
    ),
    responseFormat => '{"type":"json_schema","json_schema":{"name":"renewal_outreach_draft","schema":{"type":"object","properties":{"subject":{"type":"string"},"body":{"type":"string"},"suggested_send_date":{"type":"string"}},"required":["subject","body","suggested_send_date"]},"strict":true}}'
  ) AS outreach_json
FROM gold.renewal_signals
WHERE days_to_renewal <= 60;

因为 prompt 由你来写，它能做模型能做的任何事，所以它能处理那些更专门的函数覆盖不到的场景。

Demo notebook: Generative drafting

生产环境实战建议

• 第一天就给任务打标签：这能让你把 AI Functions 的成本归因到正确的任务上。
• 优先尝试面向任务的函数：只有当 ai_classify、ai_extract、ai_parse_document、ai_translate 都不合适时，才用 ai_query。
• 主动要求结构化输出：对 ai_query，使用 responseFormat 来获取结构化输出。如果你传入一个 DDL STRUCT schema，你会得到带类型的字段而非原始字符串；JSON schema / json_object 格式仍然返回 JSON 字符串。
• 慎重选择模型：每个基础模型都有权衡取舍，包括成本、性能和受支持的输入格式。务必想清楚，针对哪个用例该选哪个模型。
• 先抽样，再规模化：至少先跑 10,000 行，读一遍输出，再跑剩下的。成本与准确率的权衡是随每个用例而定的。
• 把 prompt 当代码对待：给它们做版本管理，在 pull request 里评审、加注释。在这个工作流里，prompt 是一种内嵌了业务逻辑的转换。

这对你的数据仓库战略意味着什么

贯穿这六个用例的主线始终如一：AI 和仓库其余部分跑在同一个地方——一个平台、一套治理模型、一张账单、一套 pipeline。你现有 SQL ETL 里的任何一行，都能接上一个 AI 步骤，而不需要你额外搭建一套系统来托管它；而过去在旁支上做翻译、打分或分类的那些 Python 脚本，都成了「一行代码替换」的候选对象。

所以，从一列开始。挑一个当前服务最脆弱的负载，把它重写成一条 SELECT，在 10,000 行上跑一遍，读读返回了什么。一次快速冲刺之后，你就知道它合不合适了——而且你也就不必再为「只为了用一下数据，就得把数据运出去」额外付费了。

示例 notebook

每个 notebook 都附带内联的样例数据、一步步的 SQL 和你应该期待的输出。

• Use case 1: Generative drafting
• Use case 2: Document intelligence
• Use case 3: Sentiment analysis
• Use case 4: Translation and normalization
• Use case 5: Classification and routing
• Use case 6: Sales-call extraction
• All notebooks