27---工作流调度

超、凢脫俗 2023-09-27 19:38 7阅读 0赞

工作流是一种简单但功能强大的构造，它接受可调用元素并返回元素。他们正在流式传输并批量处理数据，从而可以高效地处理大量数据。使用流数据时，工作流会持续运行，直到数据流耗尽。

工作流也可以安排运行。在这种情况下，一组静态元素会动态扩展。例如，返回项目的 API 服务端点，或轮询包含进出文件的目录。

本文将介绍如何在 txtai 中使用工作流调度。

安装依赖  
安装txtai和所有依赖项。  
pip install txtai\[workflow\]  
创建工作流操作  
工作流运行一系列任务来转换和处理数据。此部分创建可用作工作流操作的可调用对象。该对象遍历数据集，返回一批数据。  
from datasets import load\_dataset

class Stream:  
def **init**(self):  
self.dataset = load\_dataset(“ag\_news”, split=“train”)  
self.index, self.size = 0, 2500

def **call**(self, fields):  
outputs = \[\]  
for field in fields:  
output = \[\]  
for row in self.dataset.select(range(self.index, self.index+self.size)):  
output.append((self.index, row\[field\], None))  
self.index += 1

outputs.append(output)
    
    return outputs

建立工作流程  
接下来我们将创建工作流。该工作流从流中读取成批数据并将其加载到嵌入索引中。我们将按计划的时间间隔运行此工作流四次，以演示计划的工作流。  
from txtai.app import Application

## Run up to every 5 seconds 4 times ##

workflow = “”"  
writable: true  
embeddings:  
path: sentence-transformers/nli-mpnet-base-v2  
content: true

workflow:  
index:  
schedule:  
cron: ‘\* \* \* \* \* 0/5’  
elements:  
\- text  
iterations: 4  
tasks:  
\- **main**.Stream  
\- upsert  
“”"  
app = API(workflow)  
app.wait()  
2022-02-03 02:12:06,751 \[INFO\] schedule: ‘index’ scheduler started with schedule \* \* \* \* \* 0/5  
2022-02-03 02:12:06,757 \[INFO\] schedule: ‘index’ next run scheduled for 2022-02-03T02:12:10+00:00  
2022-02-03 02:12:34,937 \[INFO\] schedule: ‘index’ next run scheduled for 2022-02-03T02:12:35+00:00