• 分布式执行引擎ray入门--（2）Ray Data

    目录

    一、overview

    基础代码

    核心API：

    二、核心概念

    2.1 加载数据

    从S3上读

    从本地读： 

     其他读取方式

      读取分布式数据（spark）

     从ML libraries 库中读取（不支持并行读取）

    从sql中读取

    2.2 变换数据

    map

    flat_map

    Transforming batches

    Shuffling rows

    Repartitioning data

    2.3 消费数据

    1） 按行遍历

    2）按batch遍历

    3）遍历batch时shuffle

    4）为分布式并行训练分割数据

    2.4 保存数据

    保存文件

    修改分区数

    将数据转换为python对象

    将数据转换为分布式数据（spark）

    今天来带大家一起来学习下ray中对数据的操作，还是非常简洁的。

    一、overview

     

    基础代码

    1. from typing import Dict
    2. import numpy as np
    3. import ray
    4.  
    5. # Create datasets from on-disk files, Python objects, and cloud storage like S3.
    6. ds = ray.data.read_csv("s3://anonymous@ray-example-data/iris.csv")
    7.  
    8. # Apply functions to transform data. Ray Data executes transformations in parallel.
    9. def compute_area(batch: Dict[str, np.ndarray]) -> Dict[str, np.ndarray]:
    10.     length = batch["petal length (cm)"]
    11.     width = batch["petal width (cm)"]
    12.     batch["petal area (cm^2)"] = length * width
    13.     return batch
    14.  
    15. transformed_ds = ds.map_batches(compute_area)
    16.  
    17. # Iterate over batches of data.
    18. for batch in transformed_ds.iter_batches(batch_size=4):
    19.     print(batch)
    20.  
    21. # Save dataset contents to on-disk files or cloud storage.
    22. transformed_ds.write_parquet("local:///tmp/iris/")

    使用ray.data可以方便地从硬盘、python对象、S3上读取文件

    最后写入云端

    核心API：

    二、核心概念

    2.1 加载数据

    • 从S3上读

    1. import ray
    2.  
    3. #加载csv文件
    4. ds = ray.data.read_csv("s3://anonymous@air-example-data/iris.csv")
    5. print(ds.schema())
    6. ds.show(limit=1)
    7.  
    8. #加载parquet文件
    9. ds = ray.data.read_parquet("s3://anonymous@ray-example-data/iris.parquet")
    10.  
    11. #加载image
    12. ds = ray.data.read_images("s3://anonymous@ray-example-data/batoidea/JPEGImages/")
    13.  
    14. # Text
    15. ds = ray.data.read_text("s3://anonymous@ray-example-data/this.txt")
    16.  
    17. # binary
    18. ds = ray.data.read_binary_files("s3://anonymous@ray-example-data/documents")
    19.  
    20. #tfrecords
    21. ds = ray.data.read_tfrecords("s3://anonymous@ray-example-data/iris.tfrecords")

    • 从本地读： 

    ds = ray.data.read_parquet("local:///tmp/iris.parquet")
    • 处理压缩文件
    1. ds = ray.data.read_csv(
    2.     "s3://anonymous@ray-example-data/iris.csv.gz",
    3.     arrow_open_stream_args={"compression": "gzip"},
    4. )

    •  其他读取方式

    1. import ray
    2.  
    3. # 从python对象里获取
    4. ds = ray.data.from_items([
    5.     {"food": "spam", "price": 9.34},
    6.     {"food": "ham", "price": 5.37},
    7.     {"food": "eggs", "price": 0.94}
    8. ])
    9.  
    10.  
    11. ds = ray.data.from_items([1, 2, 3, 4, 5])
    12.  
    13. # 从numpy里获取
    14. array = np.ones((3, 2, 2))
    15. ds = ray.data.from_numpy(array)
    16.  
    17. # 从pandas里获取
    18. df = pd.DataFrame({
    19.     "food": ["spam", "ham", "eggs"],
    20.     "price": [9.34, 5.37, 0.94]
    21. })
    22. ds = ray.data.from_pandas(df)
    23.  
    24. # 从py arrow里获取
    25.  
    26. table = pa.table({
    27.     "food": ["spam", "ham", "eggs"],
    28.     "price": [9.34, 5.37, 0.94]
    29. })
    30. ds = ray.data.from_arrow(table)
    31.  
    1. import ray
    2. import raydp
    3.  
    4. spark = raydp.init_spark(app_name="Spark -> Datasets Example",
    5.                         num_executors=2,
    6.                         executor_cores=2,
    7.                         executor_memory="500MB")
    8. df = spark.createDataFrame([(i, str(i)) for i in range(10000)], ["col1", "col2"])
    9. ds = ray.data.from_spark(df)
    10.  
    11. ds.show(3)

     从ML libraries 库中读取（不支持并行读取）

    1. import ray.data
    2. from datasets import load_dataset
    3.  
    4. # 从huggingface里读取（不支持并行读取）
    5. hf_ds = load_dataset("wikitext", "wikitext-2-raw-v1")
    6. ray_ds = ray.data.from_huggingface(hf_ds["train"])
    7. ray_ds.take(2)
    8.  
    9.  
    10. # 从TensorFlow中读取（不支持并行读取）
    11. import ray
    12. import tensorflow_datasets as tfds
    13.  
    14. tf_ds, _ = tfds.load("cifar10", split=["train", "test"])
    15. ds = ray.data.from_tf(tf_ds)
    16.  
    17. print(ds)

    从sql中读取

    1. import mysql.connector
    2.  
    3. import ray
    4.  
    5. def create_connection():
    6.     return mysql.connector.connect(
    7.         user="admin",
    8.         password=...,
    9.         host="example-mysql-database.c2c2k1yfll7o.us-west-2.rds.amazonaws.com",
    10.         connection_timeout=30,
    11.         database="example",
    12.     )
    13.  
    14. # Get all movies
    15. dataset = ray.data.read_sql("SELECT * FROM movie", create_connection)
    16. # Get movies after the year 1980
    17. dataset = ray.data.read_sql(
    18.     "SELECT title, score FROM movie WHERE year >= 1980", create_connection
    19. )
    20. # Get the number of movies per year
    21. dataset = ray.data.read_sql(
    22.     "SELECT year, COUNT(*) FROM movie GROUP BY year", create_connection
    23. )

    Ray还支持从BigQuery和MongoDB中读取，篇幅问题，不赘述了。

    2.2 变换数据

    变换默认是lazy，直到遍历、保存、检视数据集时才执行

    map

    1. import os
    2. from typing import Any, Dict
    3. import ray
    4.  
    5. def parse_filename(row: Dict[str, Any]) -> Dict[str, Any]:
    6.     row["filename"] = os.path.basename(row["path"])
    7.     return row
    8.  
    9. ds = (
    10.     ray.data.read_images("s3://anonymous@ray-example-data/image-datasets/simple", include_paths=True)
    11.     .map(parse_filename)
    12. )
    13.  
    14.

    flat_map

    1. from typing import Any, Dict, List
    2. import ray
    3.  
    4. def duplicate_row(row: Dict[str, Any]) -> List[Dict[str, Any]]:
    5.     return [row] * 2
    6.  
    7. print(
    8.     ray.data.range(3)
    9.     .flat_map(duplicate_row)
    10.     .take_all()
    11. )
    12.  
    13. # 结果：
    14. # [{'id': 0}, {'id': 0}, {'id': 1}, {'id': 1}, {'id': 2}, {'id': 2}]
    15. # 原先的元素都变成2个

    Transforming batches

    1. from typing import Dict
    2. import numpy as np
    3. import ray
    4.  
    5. def increase_brightness(batch: Dict[str, np.ndarray]) -> Dict[str, np.ndarray]:
    6.     batch["image"] = np.clip(batch["image"] + 4, 0, 255)
    7.     return batch
    8.  
    9.  
    10. # batch_format：指定batch类型，可不加
    11. ds = (
    12.     ray.data.read_images("s3://anonymous@ray-example-data/image-datasets/simple")
    13.     .map_batches(increase_brightness, batch_format="numpy")
    14. )

    如果初始化较贵，使用类而不是函数，这样每次调用类的时候，进行初始化。类有状态，而函数没有状态。

    并行度可以指定(min,max)来自由调整

    Shuffling rows

    1. import ray
    2.  
    3. ds = (
    4.     ray.data.read_images("s3://anonymous@ray-example-data/image-datasets/simple")
    5.     .random_shuffle()
    6. )

    Repartitioning data

    1. import ray
    2.  
    3. ds = ray.data.range(10000, parallelism=1000)
    4.  
    5. # Repartition the data into 100 blocks. Since shuffle=False, Ray Data will minimize
    6. # data movement during this operation by merging adjacent blocks.
    7. ds = ds.repartition(100, shuffle=False).materialize()
    8.  
    9. # Repartition the data into 200 blocks, and force a full data shuffle.
    10. # This operation will be more expensive
    11. ds = ds.repartition(200, shuffle=True).materialize()

    2.3 消费数据

    1） 按行遍历

    1. import ray
    2.  
    3. ds = ray.data.read_csv("s3://anonymous@air-example-data/iris.csv")
    4.  
    5. for row in ds.iter_rows():
    6.     print(row)

    2）按batch遍历

    numpy、pandas、torch、tf使用不同的API遍历batch

    1. # numpy
    2. import ray
    3. ds = ray.data.read_images("s3://anonymous@ray-example-data/image-datasets/simple")
    4. for batch in ds.iter_batches(batch_size=2, batch_format="numpy"):
    5.     print(batch)
    6.  
    7.  
    8. # pandas
    9. import ray
    10. ds = ray.data.read_csv("s3://anonymous@air-example-data/iris.csv")
    11. for batch in ds.iter_batches(batch_size=2, batch_format="pandas"):
    12.     print(batch)
    13.  
    14.  
    15. # torch
    16. import ray
    17. ds = ray.data.read_images("s3://anonymous@ray-example-data/image-datasets/simple")
    18. for batch in ds.iter_torch_batches(batch_size=2):
    19.     print(batch)
    20.  
    21.  
    22. # tf
    23. import ray
    24.  
    25. ds = ray.data.read_csv("s3://anonymous@air-example-data/iris.csv")
    26.  
    27. tf_dataset = ds.to_tf(
    28.     feature_columns="sepal length (cm)",
    29.     label_columns="target",
    30.     batch_size=2
    31. )
    32. for features, labels in tf_dataset:
    33.     print(features, labels)

    3）遍历batch时shuffle

    只需要在遍历batch时增加local_shuffle_buffer_size参数即可。

    非全局洗牌，但性能更好。

    1. import ray
    2.  
    3. ds = ray.data.read_images("s3://anonymous@ray-example-data/image-datasets/simple")
    4.  
    5. for batch in ds.iter_batches(
    6.     batch_size=2,
    7.     batch_format="numpy",
    8.     local_shuffle_buffer_size=250,
    9. ):
    10.     print(batch)

    4）为分布式并行训练分割数据

    1. import ray
    2.  
    3. @ray.remote
    4. class Worker:
    5.  
    6.     def train(self, data_iterator):
    7.         for batch in data_iterator.iter_batches(batch_size=8):
    8.             pass
    9.  
    10. ds = ray.data.read_csv("s3://anonymous@air-example-data/iris.csv")
    11. workers = [Worker.remote() for _ in range(4)]
    12. shards = ds.streaming_split(n=4, equal=True)
    13. ray.get([w.train.remote(s) for w, s in zip(workers, shards)])

    2.4 保存数据

    保存文件

    非常类似pandas保存文件，唯一的区别保存本地文件时需要加入local://前缀。

    注意：如果不加local://前缀，ray则会将不同分区的数据写在不同节点上

    1. import ray
    2.  
    3. ds = ray.data.read_csv("s3://anonymous@ray-example-data/iris.csv")
    4.  
    5. # local
    6. ds.write_parquet("local:///tmp/iris/")
    7.  
    8. # s3
    9. ds.write_parquet("s3://my-bucket/my-folder")
    10.

    修改分区数

    1. import os
    2. import ray
    3.  
    4. ds = ray.data.read_csv("s3://anonymous@ray-example-data/iris.csv")
    5. ds.repartition(2).write_csv("/tmp/two_files/")
    6.  
    7. print(os.listdir("/tmp/two_files/"))

    将数据转换为python对象

    1. import ray
    2.  
    3. ds = ray.data.read_csv("s3://anonymous@ray-example-data/iris.csv")
    4.  
    5. df = ds.to_pandas()
    6. print(df)

    将数据转换为分布式数据（spark）

    1. import ray
    2. import raydp
    3.  
    4. spark = raydp.init_spark(
    5.     app_name = "example",
    6.     num_executors = 1,
    7.     executor_cores = 4,
    8.     executor_memory = "512M"
    9. )
    10.  
    11. ds = ray.data.read_csv("s3://anonymous@ray-example-data/iris.csv")
    12. df = ds.to_spark(spark)

  • 相关阅读:
    【C++】基础强训
    springboot+vue学习用品商店商城系统java毕业设计ucozu
    Python中dataframe.groupby()根据数据属性对数据分组
    sql语句实现查询
    Android Handler 机制解析
    uni-app：如何配置uni.request请求的超时响应时间（全局+局部）
    2023/9/13 -- C++/QT
    【 C++ 】unordered_map和unordered_set的介绍和使用
    Java学习笔记 --- 注解
    零基础Linux_14(基础IO_文件)缓冲区+文件系统inode等
  • 原文地址：https://blog.csdn.net/qq_17246605/article/details/136601935