企业官网建设流程全解析

@daft.udf 已经0.7.0正式标记 deprecated。daft提供非常灵活的函数定义形式。

1对1 row-rise

1 row in -> 1 value out 很多算子的组织形式

importdaft@daft.funcdefadd(a:int,b:int)->int:returna+b df=daft.from_pydict({"x":[1,2],"y":[10,20]})df=df.with_column("z",add(df["x"],df["y"]))

1对多

1 row in -> N rows out
可以这么写

fromtypingimportIterator@daft.funcdefsplit_into_sentences(text:str)->Iterator[str]:importre sentences=re.split(r'(?<=[.!?])\s+',text.strip())forsentenceinsentences:ifsentence:yieldsentence

输入

输出

这里

Daft 会自动把原来的 ticket_id 复制到每个生成出来的新行，不需要手动 explode()。

这里的关键是 Iterator[T] 和 yield。返回类型是 Iterator[str]。这告诉 Daft：
这个函数不是返回一个普通字符串，而是会连续 yield 多个字符串。
比如：

yield"chunk 1"yield"chunk 2"yield"chunk 3"

Daft 就知道：每个 yield 出来的值，都应该变成一行。

传统的做法 list + explode()

@daft.funcdefsplit_into_sentences(text:str)->list[str]:return["A.","B.","C."]df=df.with_column("sentences",split_into_sentences(df["body"]))df=df.explode("sentences")

不用先把所有结果收集成一个大 list，再 explode。对于长文档、音频切片、日志拆分、视频帧采样这种场景，Generator 更自然，也更省内存。

访问外部接口 （使用异步方式）

访问外部的时候使用异步的方式。适合 I/O，比如 HTTP API、对象存储、小文件下载、远程 embedding 服务。

@daft.func(max_concurrency=10)asyncdeffetch_url(url:str)->str:importaiohttpasyncwithaiohttp.ClientSession()assession:asyncwithsession.get(url)asresponse:returnawaitresponse.text()df=df.with_column("html",fetch_url(df["url"]))

max_concurrency=10 表示限制并发请求数，避免打爆 API。避免内存/网络压力过大。

异步函数：

asyncdeffetch(url):...

调用后不是马上把结果算出来，而是返回一个 coroutine，需要被事件循环调度执行。

在 Daft 里，async def 的意义是：

这个 UDF 可以并发执行多行，而不是一行一行阻塞等待。

比如 100 个 URL 请求：

• 普通同步函数：请求 1 完成，再请求 2，再请求 3；
• async 函数：可以同时发多个请求，谁先返回就先处理谁。

有状态的访问

有昂贵初始化时，用 @daft.cls。比如加载模型、初始化 tokenizer、创建数据库连接。

@daft.clsclassTextClassifier:def__init__(self,model_path:str):self.model=load_model(model_path)def__call__(self,text:str)->str:returnself.model.predict(text)classifier=TextClassifier("model.pkl")df=df.with_column("label",classifier(df["text"]),)

不会立刻真的加载模型。Daft 执行任务时，会在 worker 上初始化实例，并复用它处理多行。

@daft.cls 里可以有多个方法

@daft.clsclassTextProcessor:def__init__(self,prefix:str):self.prefix=prefixdef__call__(self,text:str)->str:returnself.prefix+textdeflowercase(self,text:str)->str:returntext.lower()deflength(self,text:str)->int:returnlen(text)processor=TextProcessor(">> ")df=df.select(processor(df["text"]).alias("prefixed"),processor.lowercase(df["text"]).alias("lower"),processor.length(df["text"]).alias("length"),)

call可以直接：

processor(df["text"])

普通方法要：

processor.lowercase(df["text"])

在 @daft.cls 里，如果某个方法需要指定返回类型，用 @daft.method。

fromdaftimportDataType@daft.clsclassTextProcessor:@daft.method(return_dtype=DataType.list(DataType.string()))defsplit_words(self,text:str):returntext.split()

或者返回 struct：

@daft.clsclassAnalyzer:@daft.method(return_dtype=daft.DataType.struct({"word_count":daft.DataType.int64(),"char_count":daft.DataType.int64(),}),unnest=True,)defanalyze(self,text:str):return{"word_count":len(text.split()),"char_count":len(text),}

调用：

df=df.select(Analyzer().analyze(df["text"]))

如果 unnest=True，struct 会展开成多列：word_count /char_count

批处理

@daft.func.batch
它和普通 @daft.func 的区别是：

• 普通 UDF：一行一行处理
• Batch UDF：一批一批处理

@daft.func.batch(return_dtype=daft.DataType.int64())defword_count_batch(texts:daft.Series)->list:"""Count words in each text -- operating on the entire batch at once."""return[len(text.split())fortextintexts.to_pylist()]

这里函数收到的不是一个 str，而是一批文本：texts: daft.Series
可以理解成：

["hello world","this is a ticket","daft udf batch example"]

然后函数返回同样长度的结果：

[2,4,4]

也就是：Series[str] -> list[int]
主要是为了减少 Python 调用开销，并且方便用向量化库。
普通 UDF：每一行调用一次 Python 函数。比如 100 万行，可能要调用 100 万次。
Batch UDF ：每一批调用一次 Python 函数 。比如每批 1024 行，100 万行大约调用 1000 次。调用次数少很多。

适合 NumPy

@daft.func.batch(return_dtype=daft.DataType.float64())defnormalize(values:daft.Series)->list:importnumpyasnp arr=np.array(values.to_pylist())arr=(arr-arr.mean())/arr.std()returnarr.tolist()

适合 pandas

@daft.func.batch(return_dtype=daft.DataType.string())defclean_texts(texts:daft.Series)->list:importpandasaspd s=pd.Series(texts.to_pylist())s=s.str.lower().str.strip()returns.tolist()

适合批量 API
比如 embedding API 通常支持一次传多个文本：

@daft.func.batch(return_dtype=daft.DataType.embedding(daft.DataType.float32(),1536))defembed_texts(texts:daft.Series)->list:client=...response=client.embeddings.create(input=texts.to_pylist(),model="text-embedding-3-small",)return[item.embeddingforiteminresponse.data]

这比每行单独请求一次 API 高效很多。

参考：https://docs.daft.ai/en/stable/examples/udf-patterns/#pattern-2-generator-one-input-becomes-many-rows