ML Community Day คือวันที่ 9 พฤศจิกายน! ร่วมกับเราสำหรับการปรับปรุงจาก TensorFlow, JAX และอื่น ๆ เรียนรู้เพิ่มเติม

ประสิทธิภาพที่ดีขึ้นด้วย tf.data API

ดูบน TensorFlow.org ทำงานใน Google Colab ดูแหล่งที่มาบน GitHub ดาวน์โหลดโน๊ตบุ๊ค

ภาพรวม

GPU และ TPU สามารถลดเวลาที่ต้องใช้ในการดำเนินการขั้นตอนการฝึกเพียงขั้นตอนเดียวได้อย่างมาก การบรรลุประสิทธิภาพสูงสุดต้องใช้ไปป์ไลน์อินพุตที่มีประสิทธิภาพซึ่งส่งข้อมูลสำหรับขั้นตอนถัดไปก่อนที่ขั้นตอนปัจจุบันจะเสร็จสิ้น tf.data API ช่วยในการสร้างท่อป้อนข้อมูลยืดหยุ่นและมีประสิทธิภาพ เอกสารฉบับนี้แสดงให้เห็นถึงวิธีการใช้ tf.data API เพื่อสร้างท่อป้อนข้อมูล TensorFlow performant สูง

ก่อนที่คุณจะยังคงตรวจสอบ รูปร่าง TensorFlow ท่อป้อนข้อมูล แนวทางในการเรียนรู้วิธีการใช้ tf.data API

ทรัพยากร

ติดตั้ง

import tensorflow as tf

import time

ในคู่มือนี้ คุณจะทำซ้ำในชุดข้อมูลและวัดประสิทธิภาพ การสร้างมาตรฐานประสิทธิภาพที่ทำซ้ำได้อาจเป็นเรื่องยาก ปัจจัยต่างๆ ที่ส่งผลต่อความสามารถในการทำซ้ำ ได้แก่

  • โหลด CPU ปัจจุบัน
  • ปริมาณการใช้เครือข่าย
  • กลไกที่ซับซ้อน เช่น cache

เพื่อให้ได้เกณฑ์มาตรฐานที่ทำซ้ำได้ คุณจะต้องสร้างตัวอย่างเทียม

ชุดข้อมูล

เริ่มต้นด้วยการกำหนดระดับการสืบทอดจาก tf.data.Dataset เรียกว่า ArtificialDataset ชุดข้อมูลนี้:

  • สร้าง num_samples ตัวอย่าง (เริ่มต้นคือ 3)
  • พักสักครู่ก่อนรายการแรกเพื่อจำลองการเปิดไฟล์
  • พักสักครู่ก่อนที่จะสร้างแต่ละรายการเพื่อจำลองการอ่านข้อมูลจากไฟล์
class ArtificialDataset(tf.data.Dataset):
    def _generator(num_samples):
        # Opening the file
        time.sleep(0.03)

        for sample_idx in range(num_samples):
            # Reading data (line, record) from the file
            time.sleep(0.015)

            yield (sample_idx,)

    def __new__(cls, num_samples=3):
        return tf.data.Dataset.from_generator(
            cls._generator,
            output_signature = tf.TensorSpec(shape = (1,), dtype = tf.int64),
            args=(num_samples,)
        )

ชุดนี้มีความคล้ายคลึงกับ tf.data.Dataset.range หนึ่งเพิ่มความล่าช้าคงที่จุดเริ่มต้นของและในระหว่างแต่ละตัวอย่าง

วงการฝึก

ถัดไป เขียนลูปการฝึกจำลองที่วัดระยะเวลาที่ใช้ในการวนซ้ำชุดข้อมูล เวลาฝึกอบรมเป็นแบบจำลอง

def benchmark(dataset, num_epochs=2):
    start_time = time.perf_counter()
    for epoch_num in range(num_epochs):
        for sample in dataset:
            # Performing a training step
            time.sleep(0.01)
    print("Execution time:", time.perf_counter() - start_time)

เพิ่มประสิทธิภาพการทำงาน

จะแสดงวิธีการปฏิบัติงานสามารถเพิ่มประสิทธิภาพคุณจะปรับปรุงประสิทธิภาพการทำงานของ ArtificialDataset

แนวทางที่ไร้เดียงสา

เริ่มต้นด้วยไปป์ไลน์ที่ไร้เดียงสาโดยไม่ต้องใช้ลูกเล่น วนซ้ำชุดข้อมูลตามที่เป็นอยู่

benchmark(ArtificialDataset())
Execution time: 0.26497629899995445

ภายใต้ประทุน นี่คือวิธีการใช้เวลาดำเนินการของคุณ:

พล็อตเวลาดำเนินการข้อมูล - วิธีการที่ไร้เดียงสา

โครงเรื่องแสดงให้เห็นว่าการดำเนินการขั้นตอนการฝึกอบรมเกี่ยวข้องกับ:

  • การเปิดไฟล์หากยังไม่ได้เปิด
  • กำลังดึงข้อมูลรายการจากไฟล์
  • การใช้ข้อมูลในการฝึกอบรม

อย่างไรก็ตาม ในการใช้งานซิงโครนัสแบบไร้เดียงสาเช่นนี้ ในขณะที่ไปป์ไลน์ของคุณกำลังดึงข้อมูล โมเดลของคุณไม่ได้ใช้งาน ในทางกลับกัน ในขณะที่โมเดลของคุณกำลังฝึก ไปป์ไลน์อินพุตจะไม่ได้ใช้งาน เวลาของขั้นตอนการฝึกอบรมจึงเป็นผลรวมของเวลาเปิด การอ่าน และเวลาฝึกอบรม

ส่วนถัดไปสร้างขึ้นบนไปป์ไลน์อินพุตนี้ ซึ่งแสดงให้เห็นแนวทางปฏิบัติที่ดีที่สุดสำหรับการออกแบบไปป์ไลน์อินพุต TensorFlow ที่มีประสิทธิภาพ

กำลังดึงข้อมูลล่วงหน้า

การดึงข้อมูลล่วงหน้าทับซ้อนการประมวลผลล่วงหน้าและการดำเนินการแบบจำลองของขั้นตอนการฝึกอบรม ในขณะที่รูปแบบการดำเนินการฝึกอบรมขั้นตอน s ท่อป้อนข้อมูลที่มีการอ่านข้อมูลสำหรับขั้นตอน s+1 1 การทำเช่นนี้จะลดเวลาของขั้นตอนให้เหลือสูงสุด (เมื่อเทียบกับผลรวม) ของการฝึก และเวลาที่ใช้ในการดึงข้อมูล

tf.data API ให้ tf.data.Dataset.prefetch เปลี่ยนแปลง สามารถใช้เพื่อแยกเวลาที่ข้อมูลถูกสร้างขึ้นจากเวลาที่ข้อมูลถูกใช้ไป โดยเฉพาะอย่างยิ่ง การแปลงจะใช้เธรดพื้นหลังและบัฟเฟอร์ภายในเพื่อดึงองค์ประกอบล่วงหน้าจากชุดข้อมูลอินพุตก่อนเวลาที่ร้องขอ จำนวนองค์ประกอบที่จะดึงข้อมูลล่วงหน้าควรเท่ากับ (หรืออาจมากกว่า) จำนวนแบทช์ที่ใช้โดยขั้นตอนการฝึกเดียว คุณสามารถปรับแต่งด้วยตนเองค่านี้หรือตั้งค่าให้ tf.data.AUTOTUNE ซึ่งจะแจ้งให้ tf.data รันไทม์การปรับแต่งค่าแบบไดนามิกที่รันไทม์

โปรดทราบว่าการแปลงการดึงข้อมูลล่วงหน้าให้ประโยชน์ทุกครั้งที่มีโอกาสทับซ้อนงานของ "ผู้ผลิต" กับงานของ "ผู้บริโภค"

benchmark(
    ArtificialDataset()
    .prefetch(tf.data.AUTOTUNE)
)
Execution time: 0.21731788600027357

พล็อตเวลาดำเนินการข้อมูล - วิธีการดึงข้อมูลล่วงหน้า

ตอนนี้ ตามที่แสดงแผนภาพเวลาการดำเนินการข้อมูล ในขณะที่ขั้นตอนการฝึกอบรมกำลังทำงานสำหรับตัวอย่าง 0 ไปป์ไลน์อินพุตกำลังอ่านข้อมูลสำหรับตัวอย่างที่ 1 เป็นต้น

การแยกข้อมูลแบบขนาน

ในการตั้งค่าจริง ข้อมูลที่ป้อนเข้าอาจถูกจัดเก็บจากระยะไกล (เช่น บน Google Cloud Storage หรือ HDFS) ไปป์ไลน์ชุดข้อมูลที่ทำงานได้ดีเมื่ออ่านข้อมูลในเครื่องอาจกลายเป็นคอขวดบน I/O เมื่ออ่านข้อมูลจากระยะไกลเนื่องจากความแตกต่างระหว่างที่จัดเก็บในเครื่องและระยะไกล:

  • เวลาไปครั้งแรกไบต์: การอ่านไบต์แรกของไฟล์จากการจัดเก็บระยะไกลสามารถใช้คำสั่งของขนาดที่ยาวกว่าจากการจัดเก็บในท้องถิ่น
  • อ่านผ่าน: ในขณะที่การจัดเก็บข้อมูลระยะไกลมักจะมีแบนด์วิดธ์ขนาดใหญ่รวมการอ่านไฟล์เดียวเท่านั้นที่อาจจะสามารถใช้ประโยชน์จากเศษเล็ก ๆ ของแบนด์วิดธ์นี้

นอกจากนี้เมื่อไบต์ดิบถูกโหลดลงในหน่วยความจำก็อาจจะจำเป็นต้อง deserialize และ / หรือถอดรหัสข้อมูล (เช่น protobuf ) ซึ่งจะต้องมีการคำนวณเพิ่มเติม โอเวอร์เฮดนี้มีขึ้นโดยไม่คำนึงว่าข้อมูลจะถูกเก็บไว้ในเครื่องหรือจากระยะไกล แต่อาจแย่กว่านั้นในกรณีระยะไกลหากข้อมูลไม่ได้รับการดึงข้อมูลล่วงหน้าอย่างมีประสิทธิภาพ

เพื่อบรรเทาผลกระทบของค่าโสหุ้ยการสกัดข้อมูลต่างๆที่ tf.data.Dataset.interleave การเปลี่ยนแปลงสามารถนำไปใช้คู่ขนานขั้นตอนการโหลดข้อมูล interleaving เนื้อหาของชุดข้อมูลอื่น ๆ (เช่นผู้อ่านแฟ้มข้อมูล) จำนวนชุดข้อมูลที่จะซ้อนทับกันสามารถระบุได้โดย cycle_length โต้แย้งในขณะที่ระดับของความเท่าเทียมสามารถระบุได้โดย num_parallel_calls โต้แย้ง คล้ายกับ prefetch การเปลี่ยนแปลงการ interleave การเปลี่ยนแปลงสนับสนุน tf.data.AUTOTUNE ซึ่งจะมอบหมายการตัดสินใจเกี่ยวกับระดับของความเท่าเทียมในการใช้งานกับ tf.data รันไทม์

อินเตอร์ลีฟตามลำดับ

ข้อโต้แย้งเริ่มต้นของ tf.data.Dataset.interleave เปลี่ยนแปลงทำให้มันแทรกตัวอย่างเดียวจากสองชุดข้อมูลตามลำดับ

benchmark(
    tf.data.Dataset.range(2)
    .interleave(lambda _: ArtificialDataset())
)
Execution time: 0.4987426460002098

พล็อตเวลาดำเนินการข้อมูล - อินเตอร์ลีฟตามลำดับ

พล็อตข้อมูลเวลาการดำเนินการนี้จะช่วยให้การแสดงพฤติกรรมของ interleave การเปลี่ยนแปลงการเรียกตัวอย่างอีกทางเลือกหนึ่งจากสองชุดข้อมูลที่มีอยู่ อย่างไรก็ตาม ไม่มีการปรับปรุงประสิทธิภาพที่เกี่ยวข้องที่นี่

อินเตอร์ลีฟแบบขนาน

ตอนนี้ใช้ num_parallel_calls อาร์กิวเมนต์ของ interleave การเปลี่ยนแปลง ซึ่งจะโหลดชุดข้อมูลหลายชุดพร้อมกัน ช่วยลดเวลาในการรอเปิดไฟล์

benchmark(
    tf.data.Dataset.range(2)
    .interleave(
        lambda _: ArtificialDataset(),
        num_parallel_calls=tf.data.AUTOTUNE
    )
)
Execution time: 0.283668874000341

พล็อตเวลาดำเนินการข้อมูล - วิธีการแทรกซ้อนแบบขนาน

คราวนี้ ตามที่แสดงพล็อตเวลาการดำเนินการข้อมูล การอ่านชุดข้อมูลทั้งสองแบบจะขนานกัน ช่วยลดเวลาในการประมวลผลข้อมูลทั่วโลก

การแปลงข้อมูลแบบขนาน

เมื่อเตรียมข้อมูล องค์ประกอบอินพุตอาจต้องได้รับการประมวลผลล่วงหน้า ด้วยเหตุนี้การ tf.data ข้อเสนอของ API tf.data.Dataset.map การเปลี่ยนแปลงซึ่งใช้ฟังก์ชั่นที่ผู้ใช้กำหนดองค์ประกอบของชุดข้อมูลที่ป้อนข้อมูลที่แต่ละ เนื่องจากองค์ประกอบอินพุตไม่แยกจากกัน การประมวลผลล่วงหน้าจึงสามารถทำขนานกันในคอร์ CPU หลายคอร์ได้ จะทำให้เรื่องนี้เป็นไปได้คล้าย ๆ กับ prefetch และ interleave แปลงที่ map การเปลี่ยนแปลงให้ num_parallel_calls โต้แย้งเพื่อระบุระดับของความเท่าเทียม

เลือกที่คุ้มค่าที่สุดสำหรับ num_parallel_calls โต้แย้งขึ้นอยู่กับฮาร์ดแวร์ของคุณลักษณะของข้อมูลที่ฝึกอบรมของคุณ (เช่นขนาดและรูปร่าง) ค่าใช้จ่ายของฟังก์ชั่นแผนที่ของคุณและสิ่งที่ประมวลผลอื่น ๆ ที่เกิดขึ้นบน CPU ในเวลาเดียวกัน ฮิวริสติกอย่างง่ายคือการใช้จำนวนคอร์ของ CPU ที่มีอยู่ แต่เป็นสำหรับ prefetch และ interleave การเปลี่ยนแปลงที่ map การเปลี่ยนแปลงสนับสนุน tf.data.AUTOTUNE ซึ่งจะมอบหมายการตัดสินใจเกี่ยวกับระดับของความเท่าเทียมในการใช้งานกับ tf.data รันไทม์

def mapped_function(s):
    # Do some hard pre-processing
    tf.py_function(lambda: time.sleep(0.03), [], ())
    return s

การทำแผนที่ตามลำดับ

เริ่มต้นด้วยการใช้ map การเปลี่ยนแปลงโดยไม่ต้องขนานเป็นตัวอย่างพื้นฐาน

benchmark(
    ArtificialDataset()
    .map(mapped_function)
)
Execution time: 0.4505277170001136

พล็อตเวลาดำเนินการข้อมูล - วิธีการทำแผนที่ตามลำดับ

สำหรับ วิธีการที่ไร้เดียงสา ที่นี่แสดงให้เห็นว่าพล็อตครั้งที่ใช้สำหรับการเปิดอ่านก่อนการประมวลผล (การทำแผนที่) และการฝึกอบรมขั้นตอนสรุปร่วมกันเพื่อย้ำเดียว

การทำแผนที่คู่ขนาน

ตอนนี้ ใช้ฟังก์ชันก่อนการประมวลผลเดียวกันแต่ใช้พร้อมกันกับตัวอย่างหลายรายการ

benchmark(
    ArtificialDataset()
    .map(
        mapped_function,
        num_parallel_calls=tf.data.AUTOTUNE
    )
)
Execution time: 0.2839677860001757

เวลาดำเนินการข้อมูล - การทำแผนที่คู่ขนาน

ตามที่แสดงแผนภาพข้อมูล ขั้นตอนก่อนการประมวลผลจะทับซ้อนกัน ซึ่งช่วยลดเวลาโดยรวมสำหรับการทำซ้ำครั้งเดียว

เก็บเอาไว้

tf.data.Dataset.cache การเปลี่ยนแปลงสามารถแคชชุดข้อมูลทั้งในหน่วยความจำหรือการจัดเก็บในท้องถิ่น การดำเนินการนี้จะช่วยประหยัดการดำเนินการบางอย่าง (เช่น การเปิดไฟล์และการอ่านข้อมูล) ไม่ให้ดำเนินการในแต่ละยุค

benchmark(
    ArtificialDataset()
    .map(  # Apply time consuming operations before cache
        mapped_function
    ).cache(
    ),
    5
)
Execution time: 0.3848854380003104

เวลาดำเนินการข้อมูล - วิธีชุดข้อมูลแคช

นี่คือข้อมูลเวลาดำเนินการแสดงให้เห็นว่าพล็อตที่ว่าเมื่อคุณแคชชุดข้อมูลแปลงก่อน cache หนึ่ง (เช่นการเปิดไฟล์และอ่านข้อมูล) จะดำเนินการเฉพาะในช่วงยุคแรก epochs ต่อไปจะนำมาใช้ข้อมูลที่เก็บไว้ชั่วคราวโดย cache เปลี่ยนแปลง

หากฟังก์ชั่นที่ผู้ใช้กำหนดผ่านเข้าสู่ map การเปลี่ยนแปลงที่มีราคาแพงใช้ cache เปลี่ยนแปลงหลังจากที่ map การเปลี่ยนแปลงตราบใดที่ชุดข้อมูลที่ส่งผลให้ยังคงสามารถใส่ลงในหน่วยความจำหรือจัดเก็บในท้องถิ่น หากฟังก์ชั่นที่ผู้ใช้กำหนดเพิ่มพื้นที่ที่จำเป็นในการจัดเก็บชุดข้อมูลที่เกินความจุแคชทั้งใช้มันหลังจากที่ cache เปลี่ยนแปลงหรือพิจารณาก่อนการประมวลผลข้อมูลของคุณก่อนงานฝึกอบรมของคุณเพื่อลดการใช้ทรัพยากร

การทำแผนที่เวกเตอร์

การเรียกฟังก์ชั่นที่ผู้ใช้กำหนดผ่านเข้าสู่ map การเปลี่ยนแปลงเกี่ยวข้องมีค่าใช้จ่ายในการจัดตารางเวลาและการดำเนินงานฟังก์ชั่นที่ผู้ใช้กำหนด vectorize ฟังก์ชั่นที่ผู้ใช้กำหนด (นั่นคือมีมันทำงานมากกว่าชุดของปัจจัยการผลิตในครั้งเดียว) และใช้ batch การเปลี่ยนแปลงก่อนที่ map การเปลี่ยนแปลง

เพื่อแสดงให้เห็นแนวทางปฏิบัติที่ดีนี้ ชุดข้อมูลเทียมของคุณไม่เหมาะ ความล่าช้าในการจัดตารางเวลาเป็นรอบ 10 ไมโครวินาที (10e-6 วินาที) น้อยกว่าสิบมิลลิวินาทีที่ใช้ในการ ArtificialDataset และทำให้ผลกระทบที่ยากที่จะเห็น

ตัวอย่างนี้ใช้ฐาน tf.data.Dataset.range ฟังก์ชั่นและลดความซับซ้อนห่วงการฝึกอบรมในรูปแบบที่ง่ายที่สุด

fast_dataset = tf.data.Dataset.range(10000)

def fast_benchmark(dataset, num_epochs=2):
    start_time = time.perf_counter()
    for _ in tf.data.Dataset.range(num_epochs):
        for _ in dataset:
            pass
    tf.print("Execution time:", time.perf_counter() - start_time)

def increment(x):
    return x+1

การทำแผนที่สเกลาร์

fast_benchmark(
    fast_dataset
    # Apply function one item at a time
    .map(increment)
    # Batch
    .batch(256)
)
Execution time: 0.2712608739998359

เวลาดำเนินการข้อมูล - วิธีการแมปสเกลาร์

พล็อตด้านบนแสดงสิ่งที่เกิดขึ้น (โดยมีตัวอย่างน้อยกว่า) โดยใช้วิธีการแมปสเกลาร์ แสดงว่ามีการใช้ฟังก์ชันที่แมปสำหรับแต่ละตัวอย่าง แม้ว่าฟังก์ชันนี้จะเร็วมาก แต่ก็มีค่าใช้จ่ายบางอย่างที่ส่งผลต่อประสิทธิภาพของเวลา

การทำแผนที่แบบเวกเตอร์

fast_benchmark(
    fast_dataset
    .batch(256)
    # Apply function on a batch of items
    # The tf.Tensor.__add__ method already handle batches
    .map(increment)
)
Execution time: 0.02737950600021577

เวลาดำเนินการข้อมูล - วิธีการทำแผนที่แบบเวกเตอร์

คราวนี้ ฟังก์ชันที่แมปถูกเรียกหนึ่งครั้งและนำไปใช้กับกลุ่มตัวอย่าง ตามที่พล็อตเวลาดำเนินการข้อมูลแสดง ในขณะที่ฟังก์ชันอาจใช้เวลาในการดำเนินการมากกว่า โอเวอร์เฮดจะปรากฏขึ้นเพียงครั้งเดียว ซึ่งช่วยปรับปรุงประสิทธิภาพของเวลาโดยรวม

ลดรอยเท้าหน่วยความจำ

จำนวนของการเปลี่ยนแปลงรวมทั้ง interleave , prefetch และ shuffle รักษา buffer ภายในขององค์ประกอบ หากฟังก์ชั่นที่ผู้ใช้กำหนดผ่านเข้าสู่ map การเปลี่ยนแปลงการเปลี่ยนแปลงขนาดขององค์ประกอบที่แล้วสั่งซื้อของการเปลี่ยนแปลงแผนที่และการเปลี่ยนแปลงองค์ประกอบ buffer ที่มีผลกระทบต่อการใช้งานหน่วยความจำ โดยทั่วไป เลือกลำดับที่ส่งผลให้พื้นที่หน่วยความจำลดลง เว้นแต่จะต้องการลำดับที่แตกต่างกันสำหรับประสิทธิภาพ

การคำนวณแคชบางส่วน

ก็จะแนะนำให้แคชชุดหลังจากที่ map การเปลี่ยนแปลงยกเว้นถ้าการเปลี่ยนแปลงครั้งนี้ทำให้ข้อมูลที่มีขนาดใหญ่เกินไปให้พอดีในหน่วยความจำ การแลกเปลี่ยนสามารถทำได้หากฟังก์ชั่นที่แมปของคุณแบ่งออกเป็นสองส่วน: ส่วนที่กินเวลาและส่วนที่ใช้หน่วยความจำ ในกรณีนี้ คุณสามารถโยงการเปลี่ยนแปลงของคุณได้ดังนี้:

dataset.map(time_consuming_mapping).cache().map(memory_consuming_mapping)

ด้วยวิธีนี้ ส่วนที่ใช้เวลานานจะถูกดำเนินการในช่วงยุคแรกเท่านั้น และคุณหลีกเลี่ยงการใช้พื้นที่แคชมากเกินไป

สรุปแนวปฏิบัติที่ดีที่สุด

นี่คือบทสรุปของแนวทางปฏิบัติที่ดีที่สุดสำหรับการออกแบบไปป์ไลน์อินพุต TensorFlow ที่มีประสิทธิภาพ:

การจำลองตัวเลข

ไปลึกใน tf.data.Dataset เข้าใจ API คุณสามารถเล่นกับท่อของคุณเอง ด้านล่างนี้คือรหัสที่ใช้ในการพล็อตรูปภาพจากคู่มือนี้ อาจเป็นจุดเริ่มต้นที่ดี โดยแสดงวิธีแก้ไขปัญหาเฉพาะหน้าสำหรับปัญหาทั่วไป เช่น:

  • ความสามารถในการทำซ้ำเวลาดำเนินการ
  • ฟังก์ชั่นที่แมปดำเนินการอย่างกระตือรือร้น
  • interleave callable เปลี่ยนแปลง
import itertools
from collections import defaultdict

import numpy as np
import matplotlib as mpl
import matplotlib.pyplot as plt

ชุดข้อมูล

คล้ายกับ ArtificialDataset คุณสามารถสร้างชุดข้อมูลกลับเวลาที่ใช้ในแต่ละขั้นตอน

class TimeMeasuredDataset(tf.data.Dataset):
    # OUTPUT: (steps, timings, counters)
    OUTPUT_TYPES = (tf.dtypes.string, tf.dtypes.float32, tf.dtypes.int32)
    OUTPUT_SHAPES = ((2, 1), (2, 2), (2, 3))

    _INSTANCES_COUNTER = itertools.count()  # Number of datasets generated
    _EPOCHS_COUNTER = defaultdict(itertools.count)  # Number of epochs done for each dataset

    def _generator(instance_idx, num_samples):
        epoch_idx = next(TimeMeasuredDataset._EPOCHS_COUNTER[instance_idx])

        # Opening the file
        open_enter = time.perf_counter()
        time.sleep(0.03)
        open_elapsed = time.perf_counter() - open_enter

        for sample_idx in range(num_samples):
            # Reading data (line, record) from the file
            read_enter = time.perf_counter()
            time.sleep(0.015)
            read_elapsed = time.perf_counter() - read_enter

            yield (
                [("Open",), ("Read",)],
                [(open_enter, open_elapsed), (read_enter, read_elapsed)],
                [(instance_idx, epoch_idx, -1), (instance_idx, epoch_idx, sample_idx)]
            )
            open_enter, open_elapsed = -1., -1.  # Negative values will be filtered


    def __new__(cls, num_samples=3):
        return tf.data.Dataset.from_generator(
            cls._generator,
            output_types=cls.OUTPUT_TYPES,
            output_shapes=cls.OUTPUT_SHAPES,
            args=(next(cls._INSTANCES_COUNTER), num_samples)
        )

ชุดนี้จะยกตัวอย่างของรูปร่าง [[2, 1], [2, 2], [2, 3]] และประเภท [tf.dtypes.string, tf.dtypes.float32, tf.dtypes.int32] แต่ละตัวอย่างคือ:

(
  [("Open"), ("Read")],
  [(t0, d), (t0, d)],
  [(i, e, -1), (i, e, s)]
)

ที่ไหน:

  • Open และ Read ขั้นตอนตัวระบุ
  • t0 คือการประทับเวลาเมื่อขั้นตอนสอดคล้องกันเริ่มต้น
  • d คือเวลาที่ใช้ในขั้นตอนที่สอดคล้องกัน
  • i เป็นดัชนีอินสแตนซ์
  • e ดัชนียุค (จำนวนครั้งชุดข้อมูลที่ได้รับซ้ำ)
  • s ดัชนีตัวอย่าง

วงวนซ้ำ

ทำให้การวนซ้ำซับซ้อนขึ้นเล็กน้อยเพื่อรวมการกำหนดเวลาทั้งหมด สิ่งนี้จะใช้ได้กับชุดข้อมูลที่สร้างตัวอย่างตามรายละเอียดด้านบนเท่านั้น

def timelined_benchmark(dataset, num_epochs=2):
    # Initialize accumulators
    steps_acc = tf.zeros([0, 1], dtype=tf.dtypes.string)
    times_acc = tf.zeros([0, 2], dtype=tf.dtypes.float32)
    values_acc = tf.zeros([0, 3], dtype=tf.dtypes.int32)

    start_time = time.perf_counter()
    for epoch_num in range(num_epochs):
        epoch_enter = time.perf_counter()
        for (steps, times, values) in dataset:
            # Record dataset preparation informations
            steps_acc = tf.concat((steps_acc, steps), axis=0)
            times_acc = tf.concat((times_acc, times), axis=0)
            values_acc = tf.concat((values_acc, values), axis=0)

            # Simulate training time
            train_enter = time.perf_counter()
            time.sleep(0.01)
            train_elapsed = time.perf_counter() - train_enter

            # Record training informations
            steps_acc = tf.concat((steps_acc, [["Train"]]), axis=0)
            times_acc = tf.concat((times_acc, [(train_enter, train_elapsed)]), axis=0)
            values_acc = tf.concat((values_acc, [values[-1]]), axis=0)

        epoch_elapsed = time.perf_counter() - epoch_enter
        # Record epoch informations
        steps_acc = tf.concat((steps_acc, [["Epoch"]]), axis=0)
        times_acc = tf.concat((times_acc, [(epoch_enter, epoch_elapsed)]), axis=0)
        values_acc = tf.concat((values_acc, [[-1, epoch_num, -1]]), axis=0)
        time.sleep(0.001)

    tf.print("Execution time:", time.perf_counter() - start_time)
    return {"steps": steps_acc, "times": times_acc, "values": values_acc}

วิธีการวางแผน

สุดท้ายกำหนดฟังก์ชั่นสามารถที่จะวางแผนระยะเวลาที่กำหนดค่าที่ส่งกลับโดย timelined_benchmark ฟังก์ชั่น

def draw_timeline(timeline, title, width=0.5, annotate=False, save=False):
    # Remove invalid entries (negative times, or empty steps) from the timelines
    invalid_mask = np.logical_and(timeline['times'] > 0, timeline['steps'] != b'')[:,0]
    steps = timeline['steps'][invalid_mask].numpy()
    times = timeline['times'][invalid_mask].numpy()
    values = timeline['values'][invalid_mask].numpy()

    # Get a set of different steps, ordered by the first time they are encountered
    step_ids, indices = np.stack(np.unique(steps, return_index=True))
    step_ids = step_ids[np.argsort(indices)]

    # Shift the starting time to 0 and compute the maximal time value
    min_time = times[:,0].min()
    times[:,0] = (times[:,0] - min_time)
    end = max(width, (times[:,0]+times[:,1]).max() + 0.01)

    cmap = mpl.cm.get_cmap("plasma")
    plt.close()
    fig, axs = plt.subplots(len(step_ids), sharex=True, gridspec_kw={'hspace': 0})
    fig.suptitle(title)
    fig.set_size_inches(17.0, len(step_ids))
    plt.xlim(-0.01, end)

    for i, step in enumerate(step_ids):
        step_name = step.decode()
        ax = axs[i]
        ax.set_ylabel(step_name)
        ax.set_ylim(0, 1)
        ax.set_yticks([])
        ax.set_xlabel("time (s)")
        ax.set_xticklabels([])
        ax.grid(which="both", axis="x", color="k", linestyle=":")

        # Get timings and annotation for the given step
        entries_mask = np.squeeze(steps==step)
        serie = np.unique(times[entries_mask], axis=0)
        annotations = values[entries_mask]

        ax.broken_barh(serie, (0, 1), color=cmap(i / len(step_ids)), linewidth=1, alpha=0.66)
        if annotate:
            for j, (start, width) in enumerate(serie):
                annotation = "\n".join([f"{l}: {v}" for l,v in zip(("i", "e", "s"), annotations[j])])
                ax.text(start + 0.001 + (0.001 * (j % 2)), 0.55 - (0.1 * (j % 2)), annotation,
                        horizontalalignment='left', verticalalignment='center')
    if save:
        plt.savefig(title.lower().translate(str.maketrans(" ", "_")) + ".svg")

ใช้ wrappers สำหรับฟังก์ชันที่แมป

เมื่อต้องการเรียกใช้ฟังก์ชั่นแมปในบริบทความกระตือรือร้นที่คุณจะต้องห่อไว้ภายใน tf.py_function โทร

def map_decorator(func):
    def wrapper(steps, times, values):
        # Use a tf.py_function to prevent auto-graph from compiling the method
        return tf.py_function(
            func,
            inp=(steps, times, values),
            Tout=(steps.dtype, times.dtype, values.dtype)
        )
    return wrapper

การเปรียบเทียบท่อ

_batch_map_num_items = 50

def dataset_generator_fun(*args):
    return TimeMeasuredDataset(num_samples=_batch_map_num_items)

ไร้เดียงสา

@map_decorator
def naive_map(steps, times, values):
    map_enter = time.perf_counter()
    time.sleep(0.001)  # Time consuming step
    time.sleep(0.0001)  # Memory consuming step
    map_elapsed = time.perf_counter() - map_enter

    return (
        tf.concat((steps, [["Map"]]), axis=0),
        tf.concat((times, [[map_enter, map_elapsed]]), axis=0),
        tf.concat((values, [values[-1]]), axis=0)
    )

naive_timeline = timelined_benchmark(
    tf.data.Dataset.range(2)
    .flat_map(dataset_generator_fun)
    .map(naive_map)
    .batch(_batch_map_num_items, drop_remainder=True)
    .unbatch(),
    5
)
WARNING:tensorflow:From /tmp/ipykernel_23983/64197174.py:36: calling DatasetV2.from_generator (from tensorflow.python.data.ops.dataset_ops) with output_types is deprecated and will be removed in a future version.
Instructions for updating:
Use output_signature instead
WARNING:tensorflow:From /tmp/ipykernel_23983/64197174.py:36: calling DatasetV2.from_generator (from tensorflow.python.data.ops.dataset_ops) with output_shapes is deprecated and will be removed in a future version.
Instructions for updating:
Use output_signature instead
Execution time: 13.13538893499981

เพิ่มประสิทธิภาพ

@map_decorator
def time_consuming_map(steps, times, values):
    map_enter = time.perf_counter()
    time.sleep(0.001 * values.shape[0])  # Time consuming step
    map_elapsed = time.perf_counter() - map_enter

    return (
        tf.concat((steps, tf.tile([[["1st map"]]], [steps.shape[0], 1, 1])), axis=1),
        tf.concat((times, tf.tile([[[map_enter, map_elapsed]]], [times.shape[0], 1, 1])), axis=1),
        tf.concat((values, tf.tile([[values[:][-1][0]]], [values.shape[0], 1, 1])), axis=1)
    )


@map_decorator
def memory_consuming_map(steps, times, values):
    map_enter = time.perf_counter()
    time.sleep(0.0001 * values.shape[0])  # Memory consuming step
    map_elapsed = time.perf_counter() - map_enter

    # Use tf.tile to handle batch dimension
    return (
        tf.concat((steps, tf.tile([[["2nd map"]]], [steps.shape[0], 1, 1])), axis=1),
        tf.concat((times, tf.tile([[[map_enter, map_elapsed]]], [times.shape[0], 1, 1])), axis=1),
        tf.concat((values, tf.tile([[values[:][-1][0]]], [values.shape[0], 1, 1])), axis=1)
    )


optimized_timeline = timelined_benchmark(
    tf.data.Dataset.range(2)
    .interleave(  # Parallelize data reading
        dataset_generator_fun,
        num_parallel_calls=tf.data.AUTOTUNE
    )
    .batch(  # Vectorize your mapped function
        _batch_map_num_items,
        drop_remainder=True)
    .map(  # Parallelize map transformation
        time_consuming_map,
        num_parallel_calls=tf.data.AUTOTUNE
    )
    .cache()  # Cache data
    .map(  # Reduce memory usage
        memory_consuming_map,
        num_parallel_calls=tf.data.AUTOTUNE
    )
    .prefetch(  # Overlap producer and consumer works
        tf.data.AUTOTUNE
    )
    .unbatch(),
    5
)
Execution time: 6.723691489999965
draw_timeline(naive_timeline, "Naive", 15)

png

draw_timeline(optimized_timeline, "Optimized", 15)

png