Tf.data API के साथ बेहतर प्रदर्शन

TensorFlow.org पर देखें

Google Colab में चलाएं

GitHub पर स्रोत देखें

नोटबुक डाउनलोड करें

अवलोकन

जीपीयू और टीपीयू मौलिक रूप से एकल प्रशिक्षण कदम को निष्पादित करने के लिए आवश्यक समय को कम कर सकते हैं। चोटी के प्रदर्शन को प्राप्त करने के लिए एक कुशल इनपुट पाइपलाइन की आवश्यकता होती है जो मौजूदा चरण समाप्त होने से पहले अगले चरण के लिए डेटा वितरित करती है। tf.data API लचीला और कुशल इनपुट पाइपलाइन बनाने में मदद करता है। यह दस्तावेज़ दर्शाता है कि अत्यधिक प्रदर्शन करने वाले TensorFlow इनपुट पाइपलाइनों के निर्माण के लिए tf.data API का उपयोग कैसे करें।

जारी रखने से पहले, tf.data API का उपयोग करने का तरीका जानने के लिए बिल्ड TensorFlow इनपुट पाइपलाइन गाइड की जाँच करें।

संसाधन

• TensorFlow इनपुट पाइपलाइनों का निर्माण
• tf.data.Dataset एपीआई
• TF प्रोफाइलर के साथ tf.data प्रदर्शन का विश्लेषण करें

सेट अप

import tensorflow as tf

import time

इस गाइड के दौरान, आप एक डेटासेट में पुनरावृति करेंगे और प्रदर्शन को मापेंगे। प्रतिलिपि प्रस्तुत करने योग्य प्रदर्शन बेंचमार्क बनाना मुश्किल हो सकता है। प्रजनन क्षमता को प्रभावित करने वाले विभिन्न कारकों में शामिल हैं:

• वर्तमान सीपीयू लोड
• नेटवर्क यातायात
• जटिल तंत्र, जैसे कि कैश

एक प्रतिलिपि प्रस्तुत करने योग्य बेंचमार्क प्राप्त करने के लिए, आप एक कृत्रिम उदाहरण का निर्माण करेंगे।

डेटासेट

tf.data.Dataset नामक ArtificialDataset से विरासत में मिली कक्षा को परिभाषित करने के साथ शुरू करें। यह डेटासेट:

• num_samples नमूने 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.2541472299999441

हुड के तहत, इस तरह से आपके निष्पादन का समय व्यतीत हुआ:

कथानक से पता चलता है कि प्रशिक्षण चरण में शामिल करना शामिल है:

• यदि यह अभी तक नहीं खोला गया है तो एक फ़ाइल खोलना
• फ़ाइल से डेटा प्रविष्टि प्राप्त करना
• प्रशिक्षण के लिए डेटा का उपयोग करना

हालांकि, यहां एक भोले तुल्यकालिक कार्यान्वयन में, जबकि आपकी पाइपलाइन डेटा प्राप्त कर रही है, आपका मॉडल बेकार बैठा है। इसके विपरीत, जब आपका मॉडल प्रशिक्षण दे रहा होता है, तो इनपुट पाइपलाइन बेकार बैठी होती है। प्रशिक्षण चरण का समय इस प्रकार खुलने, पढ़ने और प्रशिक्षण के समय का योग है।

अगले खंड इस इनपुट पाइपलाइन पर निर्माण करते हैं, प्रदर्शन करने वाले TensorFlow इनपुट पाइपलाइनों को डिजाइन करने के लिए सर्वोत्तम प्रथाओं का चित्रण करते हैं।

प्रीफेटिंग

प्रीफेटिंग एक प्रशिक्षण चरण के प्रीप्रोसेसिंग और मॉडल निष्पादन को ओवरलैप करता है। जबकि मॉडल प्रशिक्षण चरण s निष्पादित कर रहा है, इनपुट पाइपलाइन चरण s+1 लिए डेटा पढ़ रही है। ऐसा करने से प्रशिक्षण के अधिकतम (योग के विपरीत) कदम समय कम हो जाता है और डेटा निकालने में समय लगता है।

tf.data API tf.data.Dataset.prefetch रूपांतरण प्रदान करता है। इसका उपयोग उस समय को कम करने के लिए किया जा सकता है जब डेटा का उपभोग किया जाता है। विशेष रूप से, परिवर्तन समय-समय पर इनपुट डेटासेट से तत्वों को प्रीफ़ैच करने के लिए बैकग्राउंड थ्रेड और आंतरिक बफर का उपयोग करता है। प्रीफ़ैच करने वाले तत्वों की संख्या एकल प्रशिक्षण चरण द्वारा उपभोग किए गए बैचों की संख्या के बराबर (या संभवतः अधिक से अधिक) होनी चाहिए। आप या तो मैन्युअल रूप से इस मान को ट्यून कर सकते हैं, या इसे tf.data.AUTOTUNE सेट कर tf.data.AUTOTUNE , जो रनटाइम पर गतिशील रूप से मान को ट्यून करने के लिए tf.data रनटाइम का संकेत देगा।

ध्यान दें कि प्रीफ़ैच परिवर्तन किसी भी समय लाभ प्रदान करता है "उपभोक्ता" के काम के साथ "निर्माता" के काम को ओवरलैप करने का अवसर है।

benchmark(
    ArtificialDataset()
    .prefetch(tf.data.AUTOTUNE)
)

Execution time: 0.20805208699994182

अब, जैसा कि डेटा निष्पादन समय प्लॉट दिखाता है, जबकि प्रशिक्षण चरण नमूना 0 के लिए चल रहा है, इनपुट पाइपलाइन नमूना 1 के लिए डेटा पढ़ रहा है, और इसी तरह।

डेटा निष्कर्षण को समानांतर करना

वास्तविक दुनिया की सेटिंग में, इनपुट डेटा को दूरस्थ रूप से संग्रहीत किया जा सकता है (उदाहरण के लिए, Google क्लाउड स्टोरेज या एचडीएफएस पर)। एक डेटासेट पाइपलाइन जो स्थानीय रूप से डेटा पढ़ने के दौरान अच्छी तरह से काम करती है, स्थानीय और दूरस्थ भंडारण के बीच निम्नलिखित अंतरों के कारण दूरस्थ रूप से डेटा पढ़ते समय I / O पर अड़चन बन सकती है:

• टाइम-टू-फ़र्स्ट-बाइट : रिमोट स्टोरेज से किसी फ़ाइल के पहले बाइट को पढ़ना लोकल स्टोरेज की तुलना में अधिक समय तक के लिए ऑर्डर ले सकता है।
• थ्रूपुट पढ़ें : जबकि रिमोट स्टोरेज आमतौर पर बड़ी एग्रीगेट बैंडविड्थ प्रदान करता है, एक फाइल को पढ़ना केवल इस बैंडविड्थ के एक छोटे से हिस्से का उपयोग करने में सक्षम हो सकता है।

इसके अलावा, एक बार जब कच्ची बाइट्स को मेमोरी में लोड किया जाता है, तो डेटा को डिस्क्राइबलाइज और / या डिक्रिप्ट करना भी आवश्यक हो सकता है (जैसे प्रोटोबुफ़ ), जिसमें अतिरिक्त कम्प्यूटेशन की आवश्यकता होती है। यह ओवरहेड मौजूद है चाहे डेटा स्थानीय रूप से या दूरस्थ रूप से संग्रहीत किया गया हो, लेकिन यदि दूरस्थ रूप से डेटा को प्रभावी ढंग से पूर्व निर्धारित नहीं किया गया है तो यह खराब हो सकता है।

विभिन्न डेटा निष्कर्षण ओवरहेड्स के प्रभाव को कम करने के लिए, tf.data.Dataset.interleave परिवर्तन का उपयोग डेटा लोडिंग चरण को समानांतर करने के लिए किया जा सकता है, अन्य डेटासेट (जैसे डेटा फ़ाइल रीडर) की सामग्री को tf.data.Dataset.interleave किया जा सकता है। ओवरलैप के डेटासेट की संख्या को cycle_length तर्क द्वारा निर्दिष्ट किया जा सकता है, जबकि 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.4883518669998921

यह डेटा निष्पादन टाइम प्लॉट interleave ट्रांसफॉर्मेशन के व्यवहार को प्रदर्शित करने की अनुमति देता है, जो उपलब्ध दो डेटासेट से वैकल्पिक रूप से नमूने ले रहा है। हालांकि, यहां कोई प्रदर्शन सुधार शामिल नहीं है।

समानांतर इंटरलेव

अब, interleave परिवर्तन के num_parallel_calls तर्क का उपयोग करें। यह समानांतर में कई डेटासेट लोड करता है, जिससे फाइलों के खुलने का समय कम हो जाता है।

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

Execution time: 0.26920967700016263

इस बार, जैसा कि डेटा एक्जीक्यूशन टाइम प्लॉट दिखाता है, दो डेटासेट की रीडिंग को समानांतर किया जाता है, जिससे ग्लोबल डेटा प्रोसेसिंग टाइम कम हो जाता है।

डेटा परिवर्तन को समानांतर करना

डेटा तैयार करते समय, इनपुट तत्वों को पूर्व-संसाधित करने की आवश्यकता हो सकती है। यह अंत करने के लिए, tf.data API tf.data.Dataset.map परिवर्तन प्रदान करता है, जो इनपुट डेटासेट के प्रत्येक तत्व के लिए एक उपयोगकर्ता-परिभाषित फ़ंक्शन लागू करता है। इनपुट तत्व एक दूसरे से स्वतंत्र होने के कारण, पूर्व-प्रसंस्करण को कई सीपीयू कोर के पार समानांतर किया जा सकता है। इसी तरह करने के लिए, यह संभव बनाने के prefetch और interleave परिवर्तनों, map परिवर्तन प्रदान करता है num_parallel_calls समानांतरवाद के स्तर को निर्दिष्ट करने के तर्क।

num_parallel_calls तर्क के लिए सर्वोत्तम मूल्य चुनना आपके हार्डवेयर, आपके प्रशिक्षण डेटा की विशेषताओं (जैसे इसका आकार और आकार), आपके मानचित्र फ़ंक्शन की लागत और एक ही समय में 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.4379127629999857

भोले दृष्टिकोण के लिए , यहाँ, जैसा कि कथानक से पता चलता है, खोलने, पढ़ने, पूर्व-प्रसंस्करण (मैपिंग) और प्रशिक्षण के चरणों के लिए खर्च किया गया समय एक साथ एक पुनरावृत्ति के लिए होता है।

समानांतर मानचित्रण

अब, समान प्री-प्रोसेसिंग फ़ंक्शन का उपयोग करें लेकिन इसे कई नमूनों पर समानांतर में लागू करें।

benchmark(
    ArtificialDataset()
    .map(
        mapped_function,
        num_parallel_calls=tf.data.AUTOTUNE
    )
)

Execution time: 0.2747970279999663

जैसा कि डेटा प्लॉट प्रदर्शित करता है, पूर्व-प्रसंस्करण चरण ओवरलैप करते हैं, एकल पुनरावृत्ति के लिए समग्र समय को कम करते हैं।

कैशिंग

tf.data.Dataset.cache परिवर्तन किसी डेटासेट को कैश कर सकता है, या तो मेमोरी में या लोकल स्टोरेज पर। यह कुछ संचालन (जैसे फ़ाइल खोलने और डेटा पढ़ने) को प्रत्येक युग के दौरान निष्पादित होने से बचाएगा।

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

Execution time: 0.3715158390000397

यहां, डेटा निष्पादन टाइम प्लॉट दिखाता है कि जब आप कोई डेटासेट कैश करते हैं, तो cache एक (फ़ाइल खोलने और डेटा रीडिंग की तरह) से पहले रूपांतरण केवल पहली अवधि के दौरान निष्पादित होते हैं। अगले युग cache परिवर्तन द्वारा cache किए गए डेटा का पुन: उपयोग करेगा।

यदि उपयोगकर्ता-निर्धारित फ़ंक्शन map परिवर्तन में पारित हो गया है, तो महंगा है, cache ट्रांसफ़ॉर्मेशन को map परिवर्तन के बाद लागू करें जब तक कि डेटासेट अभी भी मेमोरी या लोकल स्टोरेज में फिट हो सके। यदि उपयोगकर्ता-परिभाषित फ़ंक्शन, कैश क्षमता से परे डेटासेट को संग्रहीत करने के लिए आवश्यक स्थान को बढ़ाता है, तो इसे या तो cache परिवर्तन के बाद लागू cache या संसाधन उपयोग को कम करने के लिए अपने प्रशिक्षण कार्य से पहले अपने डेटा को पूर्व-प्रसंस्करण पर विचार करें।

वेक्टरिंग मैपिंग

map परिवर्तन में पारित उपयोगकर्ता-परिभाषित फ़ंक्शन को शामिल करने से उपयोगकर्ता-निर्धारित फ़ंक्शन को शेड्यूल करने और निष्पादित करने से संबंधित ओवरहेड होता है। उपयोगकर्ता-परिभाषित फ़ंक्शन को वेक्टर करें (अर्थात, एक बार में इनपुट के एक बैच पर काम करें) और map परिवर्तन से पहले batch परिवर्तन लागू करें।

इस अच्छे अभ्यास का वर्णन करने के लिए, आपका कृत्रिम डेटासेट उपयुक्त नहीं है। शेड्यूलिंग में देरी 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.9082538790000854

ऊपर दिए गए कथानक से पता चलता है कि स्केलर मैपिंग पद्धति का उपयोग करके (कम नमूनों के साथ) क्या चल रहा है। यह दर्शाता है कि प्रत्येक नमूने के लिए मैप किए गए फ़ंक्शन को लागू किया जाता है। जबकि यह फ़ंक्शन बहुत तेज़ है, इसमें कुछ ओवरहेड हैं जो समय प्रदर्शन को प्रभावित करते हैं।

सदिश मानचित्रण

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.03624614399996062

इस बार, मैप किए गए फ़ंक्शन को एक बार बुलाया जाता है और नमूने के एक बैच पर लागू होता है। जैसा कि डेटा निष्पादन समय प्लॉट दिखाता है, जबकि फ़ंक्शन को निष्पादित करने में अधिक समय लग सकता है, ओवरहेड केवल एक बार दिखाई देता है, समग्र समय प्रदर्शन में सुधार करता है।

स्मृति पदचिह्न कम करना

interleave , prefetch और shuffle सहित कई रूपांतरण, तत्वों के आंतरिक बफर को बनाए रखते हैं। यदि उपयोगकर्ता-परिभाषित फ़ंक्शन map परिवर्तन में पारित हो जाता है, तो तत्वों का आकार बदल जाता है, फिर मानचित्र परिवर्तन का क्रम और रूपांतरण जो बफर तत्वों को स्मृति उपयोग को प्रभावित करते हैं। सामान्य तौर पर, उस आदेश का चयन करें जिसके परिणामस्वरूप कम मेमोरी फ़ुटप्रिंट होता है, जब तक कि अलग-अलग ऑर्डर प्रदर्शन के लिए वांछनीय न हों।

आंशिक अभिकलन कैशिंग

map परिवर्तन के बाद डेटासेट को कैश करने की सिफारिश की जाती है, यदि यह परिवर्तन डेटा को स्मृति में फिट करने के लिए बहुत बड़ा बनाता है। यदि आपके मैप किए गए फ़ंक्शन को दो भागों में विभाजित किया जा सकता है तो एक ट्रेड-ऑफ प्राप्त किया जा सकता है: एक समय लेने वाला और एक मेमोरी लेने वाला हिस्सा। इस स्थिति में, आप नीचे दिए गए परिवर्तनों की श्रृंखला बना सकते हैं:

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

इस तरह, समय लेने वाला हिस्सा केवल पहले युग के दौरान निष्पादित किया जाता है, और आप बहुत अधिक कैश स्थान का उपयोग करने से बचते हैं।

सबसे अच्छा अभ्यास सारांश

यहाँ प्रदर्शनकारी TensorFlow इनपुट पाइपलाइनों को डिजाइन करने के लिए सर्वोत्तम प्रथाओं का सारांश दिया गया है:

• एक निर्माता और उपभोक्ता के काम को ओवरलैप करने के लिए prefetch ट्रांसफ़ॉर्म का उपयोग करें
• डेटा पढ़ने परिवर्तन parallelize का उपयोग कर interleave परिवर्तन
• num_parallel_calls तर्क सेट करके map परिवर्तन को समानांतर करें
• पहले युग के दौरान मेमोरी में कैशे डेटा में cache परिवर्तन का उपयोग करें
• map परिवर्तन में पारित उपयोगकर्ता-परिभाषित कार्यों को वेक्टर करें
• interleave , prefetch और shuffle परिवर्तनों को लागू करते समय मेमोरी का उपयोग कम करें

आंकड़ों को फिर से प्रस्तुत करना

tf.data.Dataset एपीआई समझ में गहराई तक जाने के लिए, आप अपनी खुद की पाइपलाइन के साथ खेल सकते हैं। नीचे इस गाइड से छवियों को प्लॉट करने के लिए उपयोग किया गया कोड है। यह एक अच्छा शुरुआती बिंदु हो सकता है, जैसे कि सामान्य कठिनाइयों के लिए कुछ वर्कअराउंड दिखा रहा है:

• निष्पादन का समय प्रजनन
• मैप किए गए कार्यों को उत्सुक निष्पादन
• interleave परिवर्तन कॉल करने योग्य

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")

मैप किए गए फ़ंक्शन के लिए रैपर का उपयोग करें

एक उत्सुक संदर्भ में मैप किए गए फ़ंक्शन को चलाने के लिए, आपको उन्हें एक 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 <ipython-input-1-c85330a00c6e>: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 <ipython-input-1-c85330a00c6e>: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: 12.445692234000035

अनुकूलित किया गया

@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.326935971000012

draw_timeline(naive_timeline, "Naive", 15)

draw_timeline(optimized_timeline, "Optimized", 15)