רכיב Flink אופציונלי של Managed Service for Apache Spark

הרכיב האופציונלי Apache Flink מאפשר ליצור אשכול Managed Service for Apache Spark שמוגדר להריץ עומסי עבודה מבוזרים של עיבוד בסטרימינג ועיבוד באצווה. הפעלת הרכיב הזה מכינה את סביבת זמן הריצה להפעיל צינורות עיבוד נתונים עם שמירת מצב ואפליקציות נתונים בקנה מידה גדול.

אתם יכולים להשתמש באשכול Flink כדי להריץ משימות Flink באמצעות המשאב Jobs ממסוף Google Cloud , מ-Google Cloud CLI או מ-Dataproc API, או להריץ עומסי עבודה באמצעות ה-CLI של flink בצומת הראשי. הרכיב תומך גם במשימות של Apache Beam ובהפעלה באשכולות Kerberized.

כדי ליצור אשכול שרכיב Flink מופעל בו, אתם יכולים להשתמש במסוף Google Cloud , ב-Google Cloud CLI או ב-Dataproc API.

המלצה: השתמשו באשכול מכונות וירטואליות סטנדרטי עם מאסטר אחד ורכיב Flink. אשכולות של Managed Service for Apache Spark במצב זמינות גבוהה (עם 3 מכונות וירטואליות ראשיות) לא תומכים במצב זמינות גבוהה של Flink.

אתם יכולים להריץ משימות Flink באמצעות המשאב Jobs של Managed Service for Apache Spark דרך מסוףGoogle Cloud ,‏ Google Cloud CLI או Dataproc API.

המסוף

כדי לשלוח משימה לדוגמה של ספירת מילים ב-Flink מהמסוף:

  1. פותחים את הדף Submit a job של Managed Service for Apache Spark במסוףGoogle Cloud בדפדפן.

  2. ממלאים את השדות בדף Submit a job:

    1. בוחרים את שם ה-Cluster מרשימת האשכולות.
    2. מגדירים את סוג העבודה לערך Flink.
    3. מגדירים את Main class or jar לערך org.apache.flink.examples.java.wordcount.WordCount.
    4. מגדירים את קובצי Jar לערך file:///usr/lib/flink/examples/batch/WordCount.jar.
      • ‫file:/// מציין קובץ שנמצא באשכול. ‫Managed Service for Apache Spark התקין את WordCount.jar כשהוא יצר את אשכול Flink.
      • בשדה הזה אפשר גם להזין נתיב של Cloud Storage‏ (gs://BUCKET/JARFILE) או נתיב של מערכת קבצים מבוזרת של Hadoop‏ (HDFS) (hdfs://PATH_TO_JAR).

  3. לוחצים על שליחה.

    • הפלט של מנהל ההתקן של המשימה מוצג בדף Job details.
    • המשימות של Flink מופיעות בדף Jobs של Managed Service for Apache Spark במסוף Google Cloud .
    • כדי לעצור או למחוק משימה, לוחצים על Stop או על Delete בדף Jobs או בדף Job details.

gcloud

כדי לשלוח משימת Flink לאשכול Flink ב-Managed Service for Apache Spark, מריצים את הפקודה gcloud dataproc jobs submit של ה-CLI של gcloud באופן מקומי בחלון הטרמינל או ב-Cloud Shell. 

gcloud dataproc jobs submit flink \
    --cluster=CLUSTER_NAME \
    --region=REGION \
    --class=MAIN_CLASS \
    --jar=JAR_FILE \
    -- JOB_ARGS

הערות:

  • ‫CLUSTER_NAME: שם האשכול של Managed Service for Apache Spark Flink שאליו רוצים לשלוח את המשימה.
  • ‫REGION: מציינים את האזור ב-Compute Engine שבו נמצא האשכול.
  • ‫MAIN_CLASS: מציינים את המחלקה main של אפליקציית Flink, למשל:
    • org.apache.flink.examples.java.wordcount.WordCount
  • ‫JAR_FILE: מציינים את קובץ ה-JAR של אפליקציית Flink. אתם יכולים לציין:
    • קובץ jar שמותקן באשכול, באמצעות הקידומת file:///‎:
      • file:///usr/lib/flink/examples/streaming/TopSpeedWindowing.jar
      • file:///usr/lib/flink/examples/batch/WordCount.jar
    • קובץ JAR ב-Cloud Storage: gs://BUCKET/JARFILE
    • קובץ jar ב-HDFS: hdfs://PATH_TO_JAR

  • ‫JOB_ARGS: אם רוצים, מוסיפים ארגומנטים של משימות אחרי המקף הכפול (--).

  • אחרי ששולחים את המשימה, הפלט של מנהל ההתקן של המשימה מוצג בטרמינל המקומי או ב-Cloud Shell.

    Program execution finished
Job with JobID 829d48df4ebef2817f4000dfba126e0f has finished.
Job Runtime: 13610 ms
...
(after,1)
(and,12)
(arrows,1)
(ay,1)
(be,4)
(bourn,1)
(cast,1)
(coil,1)
(come,1)

REST

בקטע הזה מוסבר איך לשלוח משימת Flink לאשכול Flink ב-Managed Service for Apache Spark באמצעות ה-API‏ jobs.submit של Managed Service for Apache Spark.

לפני שמשתמשים בנתוני הבקשה, צריך להחליף את הנתונים הבאים:

  • ‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud
  • ‫REGION: אזור האשכול
  • ‫CLUSTER_NAME: השם של אשכול Flink ב-Managed Service for Apache Spark שאליו רוצים לשלוח את המשימה.

ה-method של ה-HTTP וכתובת ה-URL: 

POST https://dataproc.googleapis.com/v1/projects/PROJECT_ID/regions/REGION/jobs:submit

תוכן בקשת JSON: 

{
  "job": {
    "placement": {
      "clusterName": "CLUSTER_NAME"
    },
    "flinkJob": {
      "mainClass": "org.apache.flink.examples.java.wordcount.WordCount",
      "jarFileUris": [
        "file:///usr/lib/flink/examples/batch/WordCount.jar"
      ]
    }
  }
}

כדי לשלוח את הבקשה צריך להרחיב אחת מהאפשרויות הבאות:

אתם אמורים לקבל תגובת JSON שדומה לזו:

{
  "reference": {
    "projectId": "PROJECT_ID",
    "jobId": "JOB_ID"
  },
  "placement": {
    "clusterName": "CLUSTER_NAME",
    "clusterUuid": "CLUSTER_UUID"
  },
  "flinkJob": {
    "mainClass": "org.apache.flink.examples.java.wordcount.WordCount",
    "args": [
      "1000"
    ],
    "jarFileUris": [
      "file:///usr/lib/flink/examples/batch/WordCount.jar"
    ]
  },
  "status": {
    "state": "PENDING",
    "stateStartTime": "2020-10-07T20:16:21.759Z"
  },
  "jobUuid": "JOB_UUID"
}
  • המשימות של Flink מופיעות בדף Jobs של Managed Service for Apache Spark במסוף Google Cloud .
  • כדי לעצור או למחוק משימה, אפשר ללחוץ על Stop או על Delete בדף Jobs או בדף Job details במסוף Google Cloud .

במקום להריץ משימות Flink באמצעות המשאב Jobs של Managed Service for Apache Spark, אתם יכולים להריץ משימות Flink בצומת הדרייבר של אשכול Flink באמצעות ה-CLI של flink.

בקטעים הבאים מתוארות דרכים שונות להריץ משימה של flink CLI באשכול Flink ב-Managed Service for Apache Spark.

  1. התחברות באמצעות SSH לצומת הראשי: משתמשים בכלי SSH כדי לפתוח חלון טרמינל במכונה הווירטואלית של מאסטר האשכולות.

  2. הגדרת ה-classpath: מפעילים את ה-classpath של Hadoop מחלון הטרמינל של SSH במכונה הווירטואלית של מאסטר האשכולות של Flink:

    export HADOOP_CLASSPATH=$(hadoop classpath)

  3. הפעלת משימות Flink: אתם יכולים להריץ משימות Flink במצבי פריסה שונים ב-YARN: מצב אפליקציה, מצב לכל משימה ומצב סשן.

    1. מצב אפליקציה: מצב אפליקציית Flink נתמך ב-Managed Service for Apache Spark מגרסת אימג' 2.0 ואילך. במצב הזה, השיטה main() של המשימה מופעלת במנהל המשימות של YARN. האשכול נכבה אחרי שהמשימה מסתיימת.

      דוגמה לשליחת משימה:

      flink run-application \
    -t yarn-application \
    -Djobmanager.memory.process.size=1024m \
    -Dtaskmanager.memory.process.size=2048m \
    -Djobmanager.heap.mb=820 \
    -Dtaskmanager.heap.mb=1640 \
    -Dtaskmanager.numberOfTaskSlots=2 \
    -Dparallelism.default=4 \
    /usr/lib/flink/examples/batch/WordCount.jar

      הצגת רשימה של משימות פעילות:

      ./bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY

      ביטול משימה שפועלת:

      ./bin/flink cancel -t yarn-application -Dyarn.application.id=application_XXXX_YY <jobId>

    2. מצב Per-job: במצב הזה של Flink, ה-method‏ main() של המשימה מופעלת בצד הלקוח.

      דוגמה לשליחת משימה:

      flink run \
    -m yarn-cluster \
    -p 4 \
    -ys 2 \
    -yjm 1024m \
    -ytm 2048m \
    /usr/lib/flink/examples/batch/WordCount.jar

    3. מצב סשן: מתחילים סשן Flink YARN ארוך, ואז שולחים לסשן משימה אחת או יותר.

      1. התחלת סשן: אפשר להתחיל סשן של Flink באחת מהדרכים הבאות:

        1. יוצרים אשכול Flink ומוסיפים את הדגל --metadata flink-start-yarn-session=true לפקודה gcloud dataproc clusters create (ראו יצירת אשכול Dataproc Flink). כשהדגל הזה מופעל, אחרי שנוצר האשכול, Managed Service for Apache Spark מריץ את /usr/bin/flink-yarn-daemon כדי להתחיל סשן של Flink באשכול.

          מזהה האפליקציה של YARN בסשן נשמר ב-/tmp/.yarn-properties-${USER}. אפשר להציג את המזהה באמצעות הפקודה yarn application -list.

        2. מריצים את הסקריפט Flink yarn-session.sh, שמותקן מראש במכונה הווירטואלית של מאסטר האשכולות, עם הגדרות בהתאמה אישית:

          דוגמה עם הגדרות מותאמות אישית:

          /usr/lib/flink/bin/yarn-session.sh \
    -s 1 \
    -jm 1024m \
    -tm 2048m \
    -nm flink-dataproc \
    --detached

        3. מריצים את סקריפט ה-wrapper של Flink,‏ /usr/bin/flink-yarn-daemon, עם הגדרות ברירת המחדל:

          . /usr/bin/flink-yarn-daemon

      2. שליחת משימה לסשן: מריצים את הפקודה הבאה כדי לשלוח משימת Flink לסשן.

        flink run -m <var>FLINK_MASTER_URL</var>/usr/lib/flink/examples/batch/WordCount.jar
        • ‫FLINK_MASTER_URL: כתובת ה-URL, כולל המארח והיציאה, של ה-VM הראשי של Flink שבו מופעלים התהליכים. מסירים את http:// prefix מכתובת ה-URL. כתובת ה-URL הזו מופיעה בפלט הפקודה כשמתחילים סשן של Flink. כדי להציג את כתובת ה-URL הזו בשדה Tracking-URL, אפשר להריץ את הפקודה הבאה:
        yarn application -list -appId=<yarn-app-id> | sed 's#http://##'
   ```

      3. הצגת רשימת המשימות בסשן: כדי להציג את רשימת המשימות של Flink בסשן, מבצעים אחת מהפעולות הבאות:

        • מריצים את flink list בלי ארגומנטים. הפקודה מחפשת את מזהה האפליקציה של YARN של הסשן ב-/tmp/.yarn-properties-${USER}.

        • משיגים את מזהה האפליקציה של YARN של הסשן מ-/tmp/.yarn-properties-${USER} או מהפלט של yarn application -list, ואז מריצים את <code>flink list -yid YARN_APPLICATION_ID.

        • מריצים את flink list -m FLINK_MASTER_URL.

      4. הפסקת סשן: כדי להפסיק את הסשן, משיגים את מזהה האפליקציה של YARN של הסשן מ-/tmp/.yarn-properties-${USER} או מהפלט של yarn application -list, ואז מריצים אחת מהפקודות הבאות:

        echo "stop" | /usr/lib/flink/bin/yarn-session.sh -id YARN_APPLICATION_ID

        yarn application -kill YARN_APPLICATION_ID

אתם יכולים להריץ משימות של Apache Beam ב-Managed Service for Apache Spark באמצעות FlinkRunner.

אפשר להריץ משימות Beam ב-Flink בדרכים הבאות:

  1. משימות Java Beam
  2. משימות Portable Beam

משימות Java Beam

אריזת משימות Beam בקובץ JAR. מספקים את קובץ ה-JAR המקובץ עם יחסי התלות שנדרשים להרצת המשימה.

בדוגמה הבאה מריצים משימת Java Beam מהצומת הראשי של אשכול Managed Service for Apache Spark.

  1. יוצרים אשכול Managed Service for Apache Spark עם רכיב Flink מופעל.

    gcloud dataproc clusters create CLUSTER_NAME \
    --optional-components=FLINK \
    --image-version=DATAPROC_IMAGE_VERSION \
    --region=REGION \
    --enable-component-gateway \
    --scopes=https://www.googleapis.com/auth/cloud-platform
    • ‫--optional-components: Flink.
    • ‫--image-version: גרסת קובץ האימג' של האשכול, שקובעת את גרסת Flink שמותקנת באשכול (לדוגמה, ראו את גרסאות הרכיבים של Apache Flink שמופיעות בגרסאות ההפצה של קובץ האימג' 2.0.x האחרונות וארבע הגרסאות הקודמות).
    • ‫--region: אזור נתמך של Managed Service for Apache Spark.
    • ‫--enable-component-gateway: הפעלת הגישה לממשק המשתמש של Flink Job Manager.
    • ‫--scopes: הפעלת הגישה לממשקי ה-API של Google Cloud על ידי האשכול (ראו שיטות מומלצות להיקפים). היקף ההרשאות cloud-platform מופעל כברירת מחדל (אין צורך לכלול את הגדרת הדגל הזו) כשיוצרים אשכול שמשתמש בגרסת אימג' 2.1 או גרסה חדשה יותר של Managed Service for Apache Spark.

  2. משתמשים בכלי SSH כדי לפתוח חלון טרמינל בצומת הראשי של אשכול Flink.

  3. מפעילים סשן של Flink YARN בצומת המאסטר של אשכול Managed Service for Apache Spark.

    . /usr/bin/flink-yarn-daemon

    שימו לב לגרסת Flink באשכול Managed Service for Apache Spark.

    flink --version

  4. במכונה המקומית, יוצרים את הדוגמה הקנונית של ספירת מילים ב-Beam ב-Java.

    בוחרים גרסת Beam שתואמת לגרסת Flink באשכול Managed Service for Apache Spark. בטבלה Flink Version Compatibility מפורטת התאימות בין גרסאות Beam לגרסאות Flink.

    פותחים את קובץ ה-POM שנוצר. בודקים את גרסת ה-Beam Flink runner שצוינה בתג <flink.artifact.name>. אם גרסת ה-Beam Flink runner בשם פריט המידע שנוצר בתהליך הפיתוח (artifact) של Flink לא תואמת לגרסת ה-Flink באשכול, צריך לעדכן את מספר הגרסה כך שיתאים.

    mvn archetype:generate \
    -DarchetypeGroupId=org.apache.beam \
    -DarchetypeArtifactId=beam-sdks-java-maven-archetypes-examples \
    -DarchetypeVersion=BEAM_VERSION \
    -DgroupId=org.example \
    -DartifactId=word-count-beam \
    -Dversion="0.1" \
    -Dpackage=org.apache.beam.examples \
    -DinteractiveMode=false

  5. אריזת הדוגמה של ספירת המילים.

    mvn package -Pflink-runner

  6. מעלים את קובץ ה-uber JAR הארוז, word-count-beam-bundled-0.1.jar (כ-135MB) לצומת הראשי של אשכול Managed Service for Apache Spark. אתם יכולים להשתמש ב-gcloud storage cp כדי להעביר קבצים מהר יותר מ-Cloud Storage לאשכול של Managed Service for Apache Spark.

    1. במסוף המקומי, יוצרים קטגוריה של Cloud Storage ומעלים את קובץ ה-JAR של uber.

      gcloud storage buckets create BUCKET_NAME

gcloud storage cp target/word-count-beam-bundled-0.1.jar gs://BUCKET_NAME/

    2. בצומת הראשי של Managed Service for Apache Spark, מורידים את קובץ ה-JAR של uber.

      gcloud storage cp gs://BUCKET_NAME/word-count-beam-bundled-0.1.jar .

  7. מריצים את משימת Java Beam בצומת הדרייבר של אשכול Managed Service for Apache Spark.

    flink run -c org.apache.beam.examples.WordCount word-count-beam-bundled-0.1.jar \
    --runner=FlinkRunner \
    --output=gs://BUCKET_NAME/java-wordcount-out

  8. בודקים שהתוצאות נכתבו בקטגוריה של Cloud Storage.

    gcloud storage cat gs://BUCKET_NAME/java-wordcount-out-SHARD_ID

  9. עוצרים את הסשן של Flink YARN.

    yarn application -list

yarn application -kill YARN_APPLICATION_ID

משימות Beam ניידות

כדי להריץ משימות Beam שכתובות ב-Python, ב-Go ובשפות נתמכות אחרות, אפשר להשתמש ב-FlinkRunner וב-PortableRunner, כפי שמתואר בדף Flink Runner של Beam (ראו גם את מפת הדרכים של Portability Framework).

בדוגמה הבאה מריצים משימת Beam ניידת ב-Python מצומת המאסטר של אשכול Managed Service for Apache Spark.

  1. יוצרים אשכול Managed Service for Apache Spark שבו הרכיבים Flink ו-Docker מופעלים.

    gcloud dataproc clusters create CLUSTER_NAME \
    --optional-components=FLINK,DOCKER \
    --image-version=DATAPROC_IMAGE_VERSION \
    --region=REGION \
    --enable-component-gateway \
    --scopes=https://www.googleapis.com/auth/cloud-platform

    הערות:

    • ‫--optional-components: Flink ו-Docker.
    • ‫--image-version: גרסת קובץ האימג' של האשכול, שקובעת את גרסת Flink שמותקנת באשכול (לדוגמה, ראו את גרסאות הרכיבים של Apache Flink שמופיעות בגרסאות ההפצה של קובץ האימג' 2.0.x האחרונות וארבע הגרסאות הקודמות).
    • ‫--region: אזור זמין של Managed Service for Apache Spark.
    • ‫--enable-component-gateway: הפעלת הגישה לממשק המשתמש של Flink Job Manager.
    • ‫--scopes: הפעלת הגישה לממשקי ה-API של Google Cloud באשכול (ראו שיטות מומלצות להיקפים). היקף ההרשאות cloud-platform מופעל כברירת מחדל (לא צריך לכלול את הגדרת הדגל הזו) כשיוצרים אשכול שמשתמש בגרסת אימג' 2.1 או גרסה חדשה יותר של Managed Service for Apache Spark.

  2. משתמשים ב-CLI של gcloud באופן מקומי או ב-Cloud Shell כדי ליצור קטגוריה של Cloud Storage. תציינו את BUCKET_NAME כשתריצו תוכנית לדוגמה של ספירת מילים.

    gcloud storage buckets create BUCKET_NAME

  3. בחלון הטרמינל של המכונה הווירטואלית של האשכול, מתחילים סשן של Flink YARN. שימו לב לכתובת ה-URL של ה-Flink master, הכתובת של ה-Flink master שבה מופעלות המשימות. תציינו את FLINK_MASTER_URL כשתריצו תוכנית לדוגמה של ספירת מילים.

    . /usr/bin/flink-yarn-daemon

    מציגים את גרסת Flink שמריצה את האשכול של Managed Service for Apache Spark ורושמים אותה בצד. תציינו את FLINK_VERSION כשתריצו תוכנית לדוגמה של ספירת מילים.

    flink --version

  4. מתקינים את ספריות Python שנדרשות למשימה בצומת הראשי של האשכול.

  5. מתקינים גרסת Beam שתואמת לגרסת Flink באשכול.

    python -m pip install apache-beam[gcp]==BEAM_VERSION

  6. מריצים את הדוגמה של ספירת המילים בצומת של מאסטר האשכולות (cluster master).

    python -m apache_beam.examples.wordcount \
    --runner=FlinkRunner \
    --flink_version=FLINK_VERSION \
    --flink_master=FLINK_MASTER_URL
    --flink_submit_uber_jar \
    --output=gs://BUCKET_NAME/python-wordcount-out

    הערות:

    • --runner: FlinkRunner.
    • ‫--flink_version: FLINK_VERSION, שצוין קודם.
    • ‫--flink_master: FLINK_MASTER_URL, שצוין קודם.
    • ‫--flink_submit_uber_jar: משתמשים ב-uber JAR כדי להריץ את משימת Beam.
    • ‫--output: BUCKET_NAME, שנוצר קודם.

  7. מוודאים שהתוצאות נכתבו בקטגוריה.

    gcloud storage cat gs://BUCKET_NAME/python-wordcount-out-SHARD_ID

  8. עוצרים את הסשן של Flink YARN.

    1. מאתרים את מזהה האפליקציה.
    yarn application -list

1. Insert the <var>YARN_APPLICATION_ID</var>, then stop the session.

yarn application -kill

רכיב ה-Flink של Managed Service for Apache Spark תומך באשכולות Kerberized. כדי לשלוח ולשמור משימת Flink או להפעיל אשכול Flink, צריך כרטיס Kerberos תקין. כברירת מחדל, טיקט של Kerberos תקף למשך שבעה ימים.

ממשק האינטרנט של Flink Job Manager זמין בזמן שפועלים אשכול סשן של Flink או משימת Flink. כדי להשתמש בממשק האינטרנט:

  1. יוצרים אשכול Managed Service for Apache Spark Flink.
  2. אחרי שיוצרים את האשכול, לוחצים על הקישור YARN ResourceManager של Component Gateway בכרטיסייה ממשק אינטרנט בדף Cluster details במסוף Google Cloud .
  3. בממשק המשתמש של YARN Resource Manager, מזהים את הרשומה של אפליקציית אשכול Flink. בהתאם לסטטוס ההשלמה של המשימה, יופיע קישור ל-ApplicationMaster או ל-History.
  4. במשימת סטרימינג שפועלת לאורך זמן, לוחצים על הקישור ApplicationManager כדי לפתוח את לוח הבקרה של Flink. במשימה שהושלמה, לוחצים על הקישור History כדי לראות את פרטי המשימה.