מחברת זו מכילה דוגמאות כיצד ליצור ולהפעיל רכיבי פונקציית Python בתוך TFX InteractiveContext ובצינור TFX מתוזמר מקומי.
במשך יותר בהקשר ומידע, לראות את רכיבי פונקצית Python Custom הדף באתר תיעוד TFX.
להכין
נתקין תחילה TFX ונייבא את המודולים הדרושים. TFX דורש Python 3.
בדוק את גרסת מערכת Python
import sys
sys.version
'3.7.5 (default, Feb 23 2021, 13:22:40) \n[GCC 8.4.0]'
שדרוג פיפ
כדי להימנע משדרוג Pip במערכת בעת הפעלה מקומית, בדוק כדי לוודא שאנו פועלים ב-Colab. ניתן כמובן לשדרג מערכות מקומיות בנפרד.
try:
import colab
!pip install --upgrade pip
except:
pass
התקן TFX
pip install -U tfx
הפעלת מחדש את זמן הריצה?
אם אתה משתמש ב-Google Colab, בפעם הראשונה שאתה מפעיל את התא שלמעלה, עליך להפעיל מחדש את זמן הריצה (Runtime > Restart runtime...). זה בגלל האופן שבו קולאב טוען חבילות.
ייבוא חבילות
אנו מייבאים TFX ובודקים את הגרסה שלו.
# Check version
from tfx import v1 as tfx
tfx.__version__
'1.4.0'
רכיבי פונקציית Python מותאמים אישית
בחלק זה, ניצור רכיבים מפונקציות של Python. לא נעשה שום בעיית ML אמיתית - הפונקציות הפשוטות הללו משמשות רק כדי להמחיש את תהליך הפיתוח של רכיבי פונקציית Python.
ראה מדריך רכיב מבוסס פונקצית Python לתיעוד נוסף.
צור רכיבים מותאמים אישית של Python
אנו מתחילים בכתיבת פונקציה שיוצרת כמה נתוני דמה. זה נכתב לקובץ מודול Python משלו.
%%writefile my_generator.py
import os
import tensorflow as tf # Used for writing files.
from tfx import v1 as tfx
# Non-public APIs, just for showcase.
from tfx.types.experimental.simple_artifacts import Dataset
@tfx.dsl.components.component
def MyGenerator(data: tfx.dsl.components.OutputArtifact[Dataset]):
"""Create a file with dummy data in the output artifact."""
with tf.io.gfile.GFile(os.path.join(data.uri, 'data_file.txt'), 'w') as f:
f.write('Dummy data')
# Set metadata and ensure that it gets passed to downstream components.
data.set_string_custom_property('my_custom_field', 'my_custom_value')
Writing my_generator.py
לאחר מכן, נכתוב רכיב שני המשתמש בנתוני הדמה שנוצרו. אנחנו רק נחשב את ה-hash של הנתונים ונחזיר אותו.
%%writefile my_consumer.py
import hashlib
import os
import tensorflow as tf
from tfx import v1 as tfx
# Non-public APIs, just for showcase.
from tfx.types.experimental.simple_artifacts import Dataset
from tfx.types.standard_artifacts import String
@tfx.dsl.components.component
def MyConsumer(data: tfx.dsl.components.InputArtifact[Dataset],
hash: tfx.dsl.components.OutputArtifact[String],
algorithm: tfx.dsl.components.Parameter[str] = 'sha256'):
"""Reads the contents of data and calculate."""
with tf.io.gfile.GFile(
os.path.join(data.uri, 'data_file.txt'), 'r') as f:
contents = f.read()
h = hashlib.new(algorithm)
h.update(tf.compat.as_bytes(contents))
hash.value = h.hexdigest()
# Read a custom property from the input artifact and set to the output.
custom_value = data.get_string_custom_property('my_custom_field')
hash.set_string_custom_property('input_custom_field', custom_value)
Writing my_consumer.py
הפעל במחברת עם InteractiveContext
כעת, נדגים את השימוש ברכיבים החדשים שלנו ב-TFX InteractiveContext.
לקבלת מידע נוסף על מה אתה יכול לעשות עם מחברת TFX InteractiveContext, לראות את-מחברת TFX Keras רכיב הדרכה .
from my_generator import MyGenerator
from my_consumer import MyConsumer
בנה את ה-InteractiveContext
# Here, we create an InteractiveContext using default parameters. This will
# use a temporary directory with an ephemeral ML Metadata database instance.
# To use your own pipeline root or database, the optional properties
# `pipeline_root` and `metadata_connection_config` may be passed to
# InteractiveContext. Calls to InteractiveContext are no-ops outside of the
# notebook.
from tfx.orchestration.experimental.interactive.interactive_context import InteractiveContext
context = InteractiveContext()
WARNING:absl:InteractiveContext pipeline_root argument not provided: using temporary directory /tmp/tfx-interactive-2021-12-05T10_37_04.715534-3q0k1y0m as root for pipeline outputs. WARNING:absl:InteractiveContext metadata_connection_config not provided: using SQLite ML Metadata database at /tmp/tfx-interactive-2021-12-05T10_37_04.715534-3q0k1y0m/metadata.sqlite.
הפעל הרכיב שלך באופן אינטראקטיבי עם context.run()
הבא, אנחנו רצים הרכיבים שלנו אינטראקטיבי בתוך המחברת עם context.run()
. הרכיב הצרכני שלנו משתמש ביציאות של רכיב הגנרטור.
generator = MyGenerator()
context.run(generator)
WARNING: Logging before InitGoogleLogging() is written to STDERR I1205 10:37:04.765872 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type
consumer = MyConsumer(
data=generator.outputs['data'],
algorithm='md5')
context.run(consumer)
I1205 10:37:04.808555 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type
לאחר הביצוע, נוכל לבדוק את התוכן של חפץ הפלט "hash" של רכיב הצרכן בדיסק.
tail -v {consumer.outputs['hash'].get()[0].uri}
==> /tmp/tfx-interactive-2021-12-05T10_37_04.715534-3q0k1y0m/MyConsumer/hash/2/value <== 0015fe7975d1a2794b59aa12635703f1
זהו, ועכשיו כתבת וביצעת רכיבים מותאמים אישית משלך!
כתוב הגדרת צינור
לאחר מכן, נחבר צינור באמצעות אותם רכיבים. בעת שימוש InteractiveContext
בתוך מחברת עובד היטב עבור ניסויים, המגדיר צינור מאפשר לך לפרוס הצינור שלך על רצים מקומיים או מרוחקים עבור שימוש ייצור.
כאן, נדגים את השימוש ב-LocalDagRunner הפועל באופן מקומי במחשב שלך. לביצוע ייצור, רצי Airflow או Kubeflow עשויים להתאים יותר.
בניית צינור
import os
import tempfile
from tfx import v1 as tfx
# Select a persistent TFX root directory to store your output artifacts.
# For demonstration purposes only, we use a temporary directory.
PIPELINE_ROOT = tempfile.mkdtemp()
# Select a pipeline name so that multiple runs of the same logical pipeline
# can be grouped.
PIPELINE_NAME = "function-based-pipeline"
# We use a ML Metadata configuration that uses a local SQLite database in
# the pipeline root directory. Other backends for ML Metadata are available
# for production usage.
METADATA_CONNECTION_CONFIG = tfx.orchestration.metadata.sqlite_metadata_connection_config(
os.path.join(PIPELINE_ROOT, 'metadata.sqlite'))
def function_based_pipeline():
# Here, we construct our generator and consumer components in the same way.
generator = MyGenerator()
consumer = MyConsumer(
data=generator.outputs['data'],
algorithm='md5')
return tfx.dsl.Pipeline(
pipeline_name=PIPELINE_NAME,
pipeline_root=PIPELINE_ROOT,
components=[generator, consumer],
metadata_connection_config=METADATA_CONNECTION_CONFIG)
my_pipeline = function_based_pipeline()
הפעל הצינור שלך עם LocalDagRunner
tfx.orchestration.LocalDagRunner().run(my_pipeline)
I1205 10:37:04.983860 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type I1205 10:37:04.990442 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type I1205 10:37:04.996665 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type I1205 10:37:05.003470 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type I1205 10:37:05.013659 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type I1205 10:37:05.031374 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type I1205 10:37:05.048280 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type I1205 10:37:05.067972 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type
אנו יכולים לבדוק את חפצי הפלט שנוצרו על ידי ביצוע צינור זה.
find {PIPELINE_ROOT}
/tmp/tmpydmun02b /tmp/tmpydmun02b/metadata.sqlite /tmp/tmpydmun02b/MyConsumer /tmp/tmpydmun02b/MyConsumer/.system /tmp/tmpydmun02b/MyConsumer/.system/executor_execution /tmp/tmpydmun02b/MyConsumer/.system/executor_execution/2 /tmp/tmpydmun02b/MyConsumer/hash /tmp/tmpydmun02b/MyConsumer/hash/2 /tmp/tmpydmun02b/MyConsumer/hash/2/value /tmp/tmpydmun02b/MyGenerator /tmp/tmpydmun02b/MyGenerator/data /tmp/tmpydmun02b/MyGenerator/data/1 /tmp/tmpydmun02b/MyGenerator/data/1/data_file.txt /tmp/tmpydmun02b/MyGenerator/.system /tmp/tmpydmun02b/MyGenerator/.system/executor_execution /tmp/tmpydmun02b/MyGenerator/.system/executor_execution/1
כעת כתבת רכיבים מותאמים אישית משלך ותיזרת את הביצוע שלהם ב-LocalDagRunner! לקבלת השלבים הבאים, צפו בסרטוני הדרכה ומדריכים נוספים על אתר TFX .