调用远程函数

Dask 是一个用于 Python 中的并行计算的开源库。此部分说明了以下内容。

如何 DataFrame 将 Dask 传递给你的远程函数

Dask DataFrames 通常用于处理大型数据集，因为它们可以容纳需要比可用内存更多的数据集。这是因为 Dask DataFrame 不会将您的本地数据加载到内存中。如果您将 Dask DataFrame 作为函数参数传递给远程函数，Dask 可能会传递对本地磁盘或云存储中数据的引用，而不是数据本身。以下代码显示了在远程函数中传递一个 Dask DataFrame 的示例，该函数将在空 DataFrame函数上运行。

#Do not pass a Dask DataFrame  to your remote function as follows
def clean(df: dask.DataFrame ):
    cleaned = df[] \ ...                            

只有当你使用时，Dask 才会将 Dask 中的数据加载 DataFrame 到内存中。 DataFrame 如果要在远程函数中使用 Dask DataFrame ，请提供数据的路径。之后，Dask 将直接从您在代码运行时指定的数据路径中读取数据集。

以下代码示例显示了如何在远程函数clean中使用 Dask DataFrame。在代码示例中，raw_data_path传递给 clean 而不是 Dask DataFrame。在代码运行时，直接从 raw_data_path 中指定的 Amazon S3 存储桶的位置读取数据集。然后，该persist函数将数据集保存在内存中以方便后续random_split函数，并使用 Dask DataFrame API 函数将数据集写回 S3 存储桶中的输出数据路径。

import dask.dataframe as dd

@remote(
   instance_type='ml.m5.24xlarge',
   volume_size=300, 
   keep_alive_period_in_seconds=600)
#pass the data path to your remote function rather than the Dask DataFrame  itself
def clean(raw_data_path: str, output_data_path: str: split_ratio: list[float]):
    df = dd.read_parquet(raw_data_path) #pass the path to your DataFrame 
    cleaned = df[(df.column_a >= 1) & (df.column_a < 5)]\
        .drop(['column_b', 'column_c'], axis=1)\
        .persist() #keep the data in memory to facilitate the following random_split operation

    train_df, test_df = cleaned.random_split(split_ratio, random_state=10)

    train_df.to_parquet(os.path.join(output_data_path, 'train')
    test_df.to_parquet(os.path.join(output_data_path, 'test'))
    
clean("s3://amzn-s3-demo-bucket/raw/", "s3://amzn-s3-demo-bucket/cleaned/", split_ratio=[0.7, 0.3])                            

如何将汇总统计数据从 Dask DataFrame 转换为 Pandas DataFrame

DataFrame 通过调用以下示例代码所示compute的方法， DataFrame 可以将来自 Dask 的汇总统计数据转换为 Pandas。在示例中，S3 存储桶包含一个无法放入内存或 Pandas 数据框的大型 Dask DataFrame 。在以下示例中，远程函数扫描数据集，并将 DataFrame包含输出统计信息的 Dask 返回describe到 Pandas DataFrame。

executor = RemoteExecutor(
    instance_type='ml.m5.24xlarge',
    volume_size=300, 
    keep_alive_period_in_seconds=600)

future = executor.submit(lambda: dd.read_parquet("s3://amzn-s3-demo-bucket/raw/").describe().compute())

future.result()                            

DMatrix是用于加载数据的内部数据结构。XGBoost不能为了在计算会话之间轻松移动而对DMatrix对象进行封存。直接传递DMatrix实例将失败，并显示为SerializationError。

如何将数据对象传递给远程函数并使用它进行训练 XGBoost

要将 Pandas DataFrame 转换为DMatrix实例并在远程函数中使用它进行训练，请将其直接传递给远程函数，如以下代码示例所示。

import xgboost as xgb

@remote
def train(df, params):
    #Convert a pandas dataframe into a DMatrix DataFrame and use it for training
    dtrain = DMatrix(df) 
    return xgb.train(dtrain, params)                        

TensorFlow 数据集和子类是训练期间 TensorFlow 用来加载数据的内部对象。 TensorFlow 不能为了在计算会话之间轻松移动而对数据集和子类进行封存。直接传递 Tensorflow 数据集或子类将失败，并显示 SerializationError。使用 Tensorflow I/O APIs 从存储中加载数据，如以下代码示例所示。

import tensorflow as tf
import tensorflow_io as tfio

@remote
def train(data_path: str, params):
    
    dataset = tf.data.TextLineDataset(tf.data.Dataset.list_files(f"{data_path}/*.txt"))
    ...
    
train("s3://amzn-s3-demo-bucket/data", {})                        

PyTorch 模型是可序列化的，可以在本地环境和远程函数之间传递。如果您的本地环境和远程环境具有不同的设备类型，例如（GPUs和CPUs），则无法将经过训练的模型返回到本地环境。例如，如果以下代码是在本地环境中开发的，GPUs但没有但在带的实例中运行GPUs，则直接返回经过训练的模型将导致DeserializationError。

# Do not return a model trained on GPUs to a CPU-only environment as follows

@remote(instance_type='ml.g4dn.xlarge')
def train(...):
    if torch.cuda.is_available():
        device = torch.device("cuda")
    else:
        device = torch.device("cpu") # a device without GPU capabilities
    
    model = Net().to(device)
    
    # train the model
    ...
    
    return model
    
model = train(...) #returns a DeserializationError if run on a device with GPU                        

要将在一个GPU环境中训练的模型返回到仅包含CPU功能的模型，请APIs直接使用 PyTorch 模型 I/O，如下面的代码示例所示。

import s3fs

model_path = "s3://amzn-s3-demo-bucket/folder/"

@remote(instance_type='ml.g4dn.xlarge')
def train(...):
    if torch.cuda.is_available():
        device = torch.device("cuda")
    else:
        device = torch.device("cpu")
    
    model = Net().to(device)
    
    # train the model
    ...
    
    fs = s3fs.FileSystem()
    with fs.open(os.path.join(model_path, 'model.pt'), 'wb') as file:
        torch.save(model.state_dict(), file) #this writes the model in a device-agnostic way (CPU vs GPU)
    
train(...) #use the model to train on either CPUs or GPUs

model = Net()
fs = s3fs.FileSystem()with fs.open(os.path.join(model_path, 'model.pt'), 'rb') as file:
    model.load_state_dict(torch.load(file, map_location=torch.device('cpu')))