הפעלת משימת Dataflow בקונטיינר בהתאמה אישית

במאמר הזה מוסבר איך להריץ צינור Dataflow באמצעות קונטיינר בהתאמה אישית.

מידע על יצירת אימג' של קונטיינר זמין במאמר יצירת אימג'ים מותאמים אישית של קונטיינרים ל-Dataflow.

כשמריצים את צינור הנתונים, מפעילים אותו באמצעות Apache Beam SDK עם אותה גרסה וגרסת שפה כמו ה-SDK בתמונת הקונטיינר המותאם אישית. השלב הזה עוזר למנוע שגיאות לא צפויות כתוצאה מתלות לא תואמת או מ-SDK לא תואם.

בדיקה מקומית

לפני שמריצים את צינור עיבוד הנתונים ב-Dataflow, מומלץ לבדוק את תמונת הקונטיינר באופן מקומי, כדי לאפשר בדיקה וניפוי באגים מהירים יותר.

מידע נוסף על שימוש ספציפי ב-Apache Beam זמין במדריך Apache Beam בנושא הרצת צינורות עם תמונות קונטיינר בהתאמה אישית.

בדיקה בסיסית באמצעות PortableRunner

כדי לוודא שאפשר למשוך תמונות של קונטיינרים מרחוק ולהריץ צינור בסיסי, משתמשים ב-Apache Beam PortableRunner. כשמשתמשים ב-PortableRunner, שליחת העבודה מתבצעת בסביבה המקומית, וההרצה של DoFn מתבצעת בסביבת Docker.

כשמשתמשים במעבדי GPU, יכול להיות שלקונטיינר Docker לא תהיה גישה למעבדי ה-GPU. כדי לבדוק את הקונטיינר עם יחידות GPU, משתמשים בdirect runner ופועלים לפי השלבים לבדיקת תמונת קונטיינר במכונה וירטואלית עצמאית עם יחידות GPU בקטע ניפוי באגים באמצעות מכונה וירטואלית עצמאית בדף 'שימוש ביחידות GPU'.

הפקודה הבאה מריצה צינור עיבוד נתונים לדוגמה:

Java

mvn compile exec:java -Dexec.mainClass=com.example.package.MyClassWithMain \
    -Dexec.args="--runner=PortableRunner \
    --jobEndpoint=REGION \
    --defaultEnvironmentType=DOCKER \
    --defaultEnvironmentConfig=IMAGE_URI \
    --inputFile=INPUT_FILE \
    --output=OUTPUT_FILE"

Python

python path/to/my/pipeline.py \
  --runner=PortableRunner \
  --job_endpoint=REGION \
  --environment_type=DOCKER \
  --environment_config=IMAGE_URI \
  --input=INPUT_FILE \
  --output=OUTPUT_FILE

המשך

go path/to/my/pipeline.go \
  --runner=PortableRunner \
  --job_endpoint=REGION \
  --environment_type=DOCKER \
  --environment_config=IMAGE_URI \
  --input=INPUT_FILE \
  --output=OUTPUT_FILE

מחליפים את מה שכתוב בשדות הבאים:

• ‫REGION: האזור של שירות המשימות שבו רוצים להשתמש, בפורמט של כתובת ויציאה. לדוגמה: localhost:3000. אפשר להשתמש ב-embed כדי להריץ שירות עבודות בתהליך.
• ‫IMAGE_URI: ה-URI של קובץ האימג' של הקונטיינר בהתאמה אישית.
• ‫INPUT_FILE: קובץ קלט שאפשר לקרוא כקובץ טקסט. ערכת ה-SDK צריכה להיות מסוגלת לגשת לקובץ הזה
  בתוך תמונת המאגר, או שהקובץ צריך להיות טעון מראש בתמונת המאגר או להיות קובץ מרוחק.
• ‫OUTPUT_FILE: נתיב לכתיבת הפלט. הנתיב הזה הוא נתיב מרוחק או נתיב מקומי במאגר.

כשהתהליך מסתיים בהצלחה, בודקים את היומנים של המסוף כדי לוודא שהתהליך הסתיים בהצלחה ושהתמונה המרוחקת שצוינה על ידי IMAGE_URI נמצאת בשימוש.

אחרי שמריצים את צינור העיבוד, הקבצים שנשמרו במאגר לא נמצאים במערכת הקבצים המקומית, והמאגר מופסק. אפשר להעתיק קבצים ממערכת הקבצים של קובץ המאגר שהופסק באמצעות docker cp.

אפשרות אחרת:

• לספק פלטים למערכת קבצים מרוחקת כמו Cloud Storage. יכול להיות שתצטרכו להגדיר גישה באופן ידני למטרות בדיקה, כולל קובצי פרטי כניסה או Application Default Credentials.
• כדי לבצע ניפוי באגים במהירות, מוסיפים רישום זמני ביומן.

שימוש ב-Direct Runner

כדי לבצע בדיקה מקומית מעמיקה יותר של קובץ האימג' של הקונטיינר ושל צינור הנתונים, אפשר להשתמש ב-Direct Runner של Apache Beam.

אפשר לבדוק את צינור העיבוד בנפרד מהקונטיינר. לשם כך, אפשר לבצע בדיקה בסביבה מקומית שתואמת לקובץ האימג' של הקונטיינר, או להפעיל את צינור העיבוד בקונטיינר פעיל.

Java

docker run -it --entrypoint "/bin/bash" IMAGE_URI
...
# On docker container:
root@4f041a451ef3:/#  mvn compile exec:java -Dexec.mainClass=com.example.package.MyClassWithMain ...

Python

docker run -it --entrypoint "/bin/bash" IMAGE_URI
...
# On docker container:
root@4f041a451ef3:/#  python path/to/my/pipeline.py ...

המשך

docker run -it --entrypoint "/bin/bash" IMAGE_URI
...
# On docker container:
root@4f041a451ef3:/#  go path/to/my/pipeline.go ...

מחליפים את IMAGE_URI במזהה המשאבים האחיד (URI) של קובץ האימג' בקונטיינר בהתאמה אישית.

בדוגמאות מניחים שכל קובצי הצינור, כולל הצינור עצמו, נמצאים במאגר המותאם אישית, הועברו ממערכת קבצים מקומית או שהם מרוחקים ונגישים ל-Apache Beam ולמאגר. לדוגמה, כדי להשתמש ב-Maven ‏ (mvn) כדי להריץ את דוגמת ה-Java הקודמת, צריך להכין את Maven ואת התלויות שלו בקונטיינר. מידע נוסף זמין במאמרים בנושא אחסון וdocker run במאמרי העזרה של Docker.

המטרה של בדיקה ב-Direct Runner היא לבדוק את צינור עיבוד הנתונים בסביבת הקונטיינר המותאם אישית, ולא לבדוק את הפעלת הקונטיינר עם ENTRYPOINT ברירת המחדל שלו. משנים את ENTRYPOINT (לדוגמה, ENTRYPOINT) כדי להריץ את צינור הנתונים ישירות או כדי לאפשר הרצה ידנית של פקודות בקונטיינר.docker run --entrypoint ...

אם אתם מסתמכים על הגדרה ספציפית שמבוססת על הרצת הקונטיינר ב-Compute Engine, אתם יכולים להריץ את הקונטיינר ישירות במכונה וירטואלית של Compute Engine. מידע נוסף זמין במאמר בנושא קונטיינרים ב-Compute Engine.

הפעלת המשימה ב-Dataflow

כשמפעילים את צינור Apache Beam ב-Dataflow, מציינים את הנתיב לקובץ האימג' של הקונטיינר. אל תשתמשו בתג :latest עם תמונות מותאמות אישית. מתייגים את הגרסאות עם תאריך או מזהה ייחודי. אם משהו משתבש, שימוש בסוג התג הזה עשוי לאפשר להחזיר את ההפעלה של צינור העיבוד להגדרה קודמת שידוע שהיא תקינה, ולאפשר בדיקה של השינויים.

Java

משתמשים ב---sdkContainerImage כדי לציין קובץ אימג' של קונטיינר SDK עבור זמן הריצה של Java.

כדי להפעיל את Portable Runner, משתמשים בניסוי הבא בהתאם לגרסת ה-SDK:

• ‫Apache Beam SDK 2.74 ואילך: --experiments=enable_portable_runner.
• ‫Apache Beam SDK 2.73 או גרסה מוקדמת יותר: --experiments=use_runner_v2.

Python

אם משתמשים בגרסת SDK‏ 2.30.0 ואילך, צריך להשתמש באפשרות הצינור --sdk_container_image כדי לציין תמונה של מאגר SDK.

בגרסאות קודמות של ה-SDK, משתמשים באפשרות הצינור --worker_harness_container_image כדי לציין את המיקום של קובץ האימג' בקונטיינר שבו רוצים להשתמש עבור ה-worker harness.

מאגרי תגים בהתאמה אישית נתמכים רק ב-Dataflow Portable Runner. אם מפעילים צינור Python לעיבוד נתונים באצווה, צריך להגדיר את דגל הניסוי בהתאם לגרסת ה-SDK:

• ‫Apache Beam SDK 2.74 ואילך: --experiments=enable_portable_runner.
• ‫Apache Beam SDK 2.73 או גרסה מוקדמת יותר: --experiments=use_runner_v2.

אם מפעילים צינור Python לסטרימינג, אין צורך לציין את הניסוי, כי צינורות Python לסטרימינג משתמשים ב-Portable Runner כברירת מחדל.

המשך

אם משתמשים בגרסת SDK ‏2.40.0 ואילך, צריך להשתמש באפשרות הצינור --sdk_container_image כדי לציין תמונת מאגר של SDK.

בגרסאות קודמות של ה-SDK, משתמשים באפשרות הצינור --worker_harness_container_image כדי לציין את המיקום של קובץ האימג' בקונטיינר שבו רוצים להשתמש עבור ה-worker harness.

מאגרי תגים בהתאמה אישית נתמכים בכל הגרסאות של Go SDK כי הם משתמשים ב-Dataflow Portable Runner כברירת מחדל.

בדוגמה הבאה מוסבר איך להפעיל את הדוגמה WordCount של אצווה עם קונטיינר בהתאמה אישית.

Java

הערה: אם אתם משתמשים ב-Apache Beam SDK 2.74 ואילך, צריך להשתמש בדגל המעודכן --experiments=enable_portable_runner כדי להפעיל את Portable Runner.

mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \
   -Dexec.args="--runner=DataflowRunner \
                --inputFile=INPUT_FILE \
                --output=OUTPUT_FILE \
                --project=PROJECT_ID \
                --region=REGION \
                --gcpTempLocation=TEMP_LOCATION \
                --diskSizeGb=DISK_SIZE_GB \
                --experiments=use_runner_v2 \
                --sdkContainerImage=IMAGE_URI"

Python

הערה: אם אתם משתמשים ב-Apache Beam SDK 2.74 ואילך, צריך להשתמש בדגל המעודכן --experiments=enable_portable_runner כדי להפעיל את Portable Runner.

שימוש ב-Apache Beam SDK ל-Python בגרסה 2.30.0 ואילך:

python -m apache_beam.examples.wordcount \
  --input=INPUT_FILE \
  --output=OUTPUT_FILE \
  --project=PROJECT_ID \
  --region=REGION \
  --temp_location=TEMP_LOCATION \
  --runner=DataflowRunner \
  --disk_size_gb=DISK_SIZE_GB \
  --experiments=use_runner_v2 \
  --sdk_container_image=IMAGE_URI

המשך

wordcount --input gs://dataflow-samples/shakespeare/kinglear.txt \
          --output gs://<your-gcs-bucket>/counts \
          --runner dataflow \
          --project your-gcp-project \
          --region your-gcp-region \
          --temp_location gs://<your-gcs-bucket>/tmp/ \
          --staging_location gs://<your-gcs-bucket>/binaries/ \
          --sdk_container_image=IMAGE_URI

מחליפים את מה שכתוב בשדות הבאים:

• ‫INPUT_FILE: נתיב הקלט של Cloud Storage שנקרא על ידי Dataflow כשמריצים את הדוגמה.
• ‫OUTPUT_FILE: נתיב הפלט של Cloud Storage שאליו נכתב צינור הנתונים לדוגמה. הקובץ הזה מכיל את ספירת המילים.
• ‫PROJECT_ID: המזהה של הפרויקט ב- Google Cloud.
• ‫REGION: האזור שבו רוצים לפרוס את משימת Dataflow.
• ‫TEMP_LOCATION: הנתיב ב-Cloud Storage שבו Dataflow יאחסן קבצים זמניים של משימות שנוצרו במהלך ההרצה של צינור עיבוד הנתונים.
• ‫DISK_SIZE_GB: אופציונלי. אם הקונטיינר גדול, כדאי להגדיל את גודל דיסק האתחול שמוגדר כברירת מחדל כדי למנוע מצב של חוסר מקום בדיסק.
• ‫IMAGE_URI: ה-URI של תמונת מאגר מותאמת אישית של SDK. תמיד משתמשים ב-SHA או בתג של מאגר תגים עם גרסה. אל תשתמשו בתג :latest או בתג שניתן לשינוי.

סטרימינג של קובץ אימג' של קונטיינר

כדי לשפר את זמן האחזור של ההפעלה וההתאמה לעומס (autoscaling) של צינור Dataflow, אפשר להפעיל סטרימינג של תמונות. התכונה הזו שימושית אם מאגר התגים המותאם אישית מכיל תוכן מיותר או אם לא נעשה שימוש בכל התוכן שלו בכל שלב. לדוגמה, יכול להיות שהקונטיינר יכיל תוכן מקרי כמו קוד של ספרייה מבוססת-CPU להסקת מסקנות מבוססת-GPU. באופן דומה, יכול להיות שיש לכם קונטיינר שמריץ צינורות ML עם כמה מודלים שמשתמשים רק במודל אחד בכל שלב, כך שאין צורך בתוכן שלו בבת אחת. הפעלת סטרימינג של תמונות יכולה לעזור לשפר את זמן האחזור במקרים האלה.

Java

--dataflowServiceOptions=enable_image_streaming

Python

--dataflow_service_options=enable_image_streaming

המשך

--dataflow_service_options=enable_image_streaming

הזרמת תמונות תאחזר חלקים מהקונטיינר המותאם אישית לפי הצורך של קוד צינור הנתונים, במקום להוריד את כל הקונטיינר מראש. אין צורך להוריד חלקים במאגר התגים שלא נעשה בהם שימוש.

כדי ליהנות מהזרמת תמונות, צריך להפעיל את Container File System API.