处理文本#

使用 Ray Data，您可以轻松读取和转换大量文本数据。

本指南将向您展示如何

读取文本文件#

Ray Data 可以读取文本行和 JSONL。此外，您可以读取原始二进制文件并手动解码数据。

要读取文本行，请调用 read_text()。Ray Data 会为每行文本创建一个行。

import ray

ds = ray.data.read_text("s3://anonymous@ray-example-data/this.txt")

ds.show(3)

{'text': 'The Zen of Python, by Tim Peters'}
{'text': 'Beautiful is better than ugly.'}
{'text': 'Explicit is better than implicit.'}

JSON Lines 是一种用于结构化数据的文本格式。它通常用于一次处理一条记录的数据。

要读取 JSON Lines 文件，请调用 read_json()。Ray Data 会为每个 JSON 对象创建一个行。

import ray

ds = ray.data.read_json("s3://anonymous@ray-example-data/logs.json")

ds.show(3)

{'timestamp': datetime.datetime(2022, 2, 8, 15, 43, 41), 'size': 48261360}
{'timestamp': datetime.datetime(2011, 12, 29, 0, 19, 10), 'size': 519523}
{'timestamp': datetime.datetime(2028, 9, 9, 5, 6, 7), 'size': 2163626}

要读取其他文本格式，请调用 read_binary_files()。然后，调用 map() 来解码您的数据。

from typing import Any, Dict
from bs4 import BeautifulSoup
import ray

def parse_html(row: Dict[str, Any]) -> Dict[str, Any]:
    html = row["bytes"].decode("utf-8")
    soup = BeautifulSoup(html, features="html.parser")
    return {"text": soup.get_text().strip()}

ds = (
    ray.data.read_binary_files("s3://anonymous@ray-example-data/index.html")
    .map(parse_html)
)

ds.show()

{'text': 'Batoidea\nBatoidea is a superorder of cartilaginous fishes...'}

有关读取文件的更多信息，请参阅加载数据。

转换文本#

要转换文本，请在函数或可调用类中实现您的转换。然后，调用 Dataset.map() 或 Dataset.map_batches()。Ray Data 会并行转换您的文本。

from typing import Any, Dict
import ray

def to_lower(row: Dict[str, Any]) -> Dict[str, Any]:
    row["text"] = row["text"].lower()
    return row

ds = (
    ray.data.read_text("s3://anonymous@ray-example-data/this.txt")
    .map(to_lower)
)

ds.show(3)

{'text': 'the zen of python, by tim peters'}
{'text': 'beautiful is better than ugly.'}
{'text': 'explicit is better than implicit.'}

有关转换数据的更多信息，请参阅转换数据。

对文本执行推理#

要使用预训练模型对文本数据执行推理，请实现一个可调用类，该类设置并调用模型。然后，调用 Dataset.map_batches()。

from typing import Dict

import numpy as np
from transformers import pipeline

import ray

class TextClassifier:
    def __init__(self):

        self.model = pipeline("text-classification")

    def __call__(self, batch: Dict[str, np.ndarray]) -> Dict[str, list]:
        predictions = self.model(list(batch["text"]))
        batch["label"] = [prediction["label"] for prediction in predictions]
        return batch

ds = (
    ray.data.read_text("s3://anonymous@ray-example-data/this.txt")
    .map_batches(TextClassifier, concurrency=2)
)

ds.show(3)

{'text': 'The Zen of Python, by Tim Peters', 'label': 'POSITIVE'}
{'text': 'Beautiful is better than ugly.', 'label': 'POSITIVE'}
{'text': 'Explicit is better than implicit.', 'label': 'POSITIVE'}

有关处理大语言模型的更多信息，请参阅处理 LLMs。

有关执行推理的更多信息，请参阅端到端：离线批量推理和有状态转换。

保存文本#

要保存文本，请调用诸如 write_parquet() 之类的方法。Ray Data 可以以多种格式保存文本。

要查看支持的文件格式完整列表，请参阅输入/输出参考。

import ray

ds = ray.data.read_text("s3://anonymous@ray-example-data/this.txt")

ds.write_parquet("local:///tmp/results")

有关保存数据的更多信息，请参阅保存数据。