עבודה עם פונקציות בהגדרת המשתמש ב-Python

פונקציה בהגדרת המשתמש (UDF) ב-Python מאפשרת להטמיע פונקציה סקלרית ב-Python ולהשתמש בה בשאילתת SQL. פונקציות מוגדרות על ידי המשתמש ב-Python דומות לפונקציות מוגדרות על ידי המשתמש ב-SQL וב-JavaScript, אבל יש להן יכולות נוספות. פונקציות UDF ב-Python מאפשרות לכם להתקין ספריות של צד שלישי מ-Python Package Index‏ (PyPI) ולגשת לשירותים חיצוניים באמצעות קישור למשאבים ב-Cloud.

פונקציות UDF ב-Python נוצרות ומופעלות במשאבים מנוהלים של BigQuery.

מגבלות

• ‫python-3.11 היא סביבת זמן הריצה היחידה שנתמכת.
• אי אפשר ליצור פונקציה זמנית מוגדרת על ידי המשתמש (UDF) ב-Python.
• אי אפשר להשתמש בפונקציה מוגדרת על ידי המשתמש (UDF) בשפת Python עם תצוגה מגובה בחומר.
• התוצאות של שאילתה שקוראת לפונקציית Python UDF לא נשמרות במטמון, כי ערך ההחזרה של פונקציית Python UDF תמיד נחשב כלא דטרמיניסטי.
• אין תמיכה ב-Assured Workloads.
• סוגי הנתונים הבאים לא נתמכים: JSON,‏ RANGE,‏ INTERVAL ו-GEOGRAPHY.
• אפשר להגדיר קונטיינרים שמריצים פונקציות מוגדרות על ידי המשתמש (UDF) ב-Python עד 4 vCpu ו-16‎ GiB.
• אין תמיכה בהצפנת קוד של פונקציות מוגדרות על ידי המשתמש (UDF) ב-Python באמצעות מפתחות הצפנה בניהול הלקוח (CMEK).
• פונקציות UDF ב-Python תומכות ב-VPC Service Controls, אבל לא תומכות ברשתות VPC.

התפקידים הנדרשים

תפקידי ה-IAM הנדרשים תלויים בשאלה אם אתם הבעלים של UDF ב-Python או משתמשים ב-UDF ב-Python.

בעלי פונקציות UDF

בדרך כלל, הבעלים של פונקציית UDF ב-Python יוצר או מעדכן את הפונקציה. נדרשים תפקידים נוספים גם אם יוצרים פונקציית UDF ב-Python שמפנה לחיבור למשאב ב-Cloud. החיבור הזה נדרש רק אם ה-UDF משתמש בסעיף WITH CONNECTION כדי לגשת לשירות חיצוני.

כדי לקבל את ההרשאות שדרושות ליצירה או לעדכון של פונקציה מוגדרת על ידי המשתמש (UDF) ב-Python, צריך לבקש מהאדמין להקצות לכם את תפקידי ה-IAM הבאים:

• עריכה של נתוני BigQuery (roles/bigquery.dataEditor) במערך הנתונים
• BigQuery Job User (roles/bigquery.jobUser) בפרויקט
• BigQuery Connection Admin (roles/bigquery.connectionAdmin) on the project

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

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

יכול להיות שתקבלו את ההרשאות האלה באמצעות תפקידים בהתאמה אישית או תפקידים מוגדרים מראש אחרים.

מידע נוסף על תפקידים ב-BigQuery זמין במאמר תפקידים מוגדרים מראש ב-IAM.

משתמשי UDF

משתמש ב-Python UDF מפעיל UDF שנוצר על ידי מישהו אחר. נדרשים גם תפקידים נוספים אם מפעילים פונקציית UDF של Python שמפנה לחיבור למשאב Cloud.

כדי לקבל את ההרשאות שדרושות להפעלת פונקציית UDF ב-Python שנוצרה על ידי מישהו אחר, צריך לבקש מהאדמין להקצות לכם את תפקידי ה-IAM הבאים:

• ‫BigQuery User (roles/bigquery.user) בפרויקט
• ‫BigQuery Data Viewer (roles/bigquery.dataViewer) במערך הנתונים
• BigQuery Connection User (roles/bigquery.connectionUser) on the connection

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

התפקידים המוגדרים מראש האלה מכילים את ההרשאות שנדרשות להפעלת פונקציית UDF של Python שנוצרה על ידי מישהו אחר. כדי לראות בדיוק אילו הרשאות נדרשות, אפשר להרחיב את הקטע ההרשאות הנדרשות:

יכול להיות שתקבלו את ההרשאות האלה באמצעות תפקידים בהתאמה אישית או תפקידים מוגדרים מראש אחרים.

מידע נוסף על תפקידים ב-BigQuery זמין במאמר תפקידים מוגדרים מראש ב-IAM.

יצירת פונקציה מתמידה מוגדרת על ידי המשתמש (UDF) ב-Python

כשיוצרים פונקציית UDF ב-Python, צריך לפעול לפי הכללים הבאים:

• הגוף של פונקציית Python UDF חייב להיות מחרוזת מילולית בתוך מרכאות שמייצגת את קוד Python. מידע נוסף על פורמטים של מחרוזות ליטרליות במירכאות

• גוף הפונקציה המוגדרת על ידי המשתמש ב-Python חייב לכלול פונקציית Python שמשמשת בארגומנט entry_point ברשימת האפשרויות של הפונקציה המוגדרת על ידי המשתמש ב-Python.

• צריך לציין גרסת זמן ריצה של Python באפשרות runtime_version. גרסת זמן הריצה היחידה של Python שנתמכת היא python-3.11. רשימה מלאה של האפשרויות הזמינות מופיעה ברשימת האפשרויות של הפונקציה עבור ההצהרה CREATE FUNCTION.

כדי ליצור פונקציית UDF מתמשכת ב-Python, משתמשים בהצהרה CREATE FUNCTION בלי מילות המפתח TEMP או TEMPORARY. כדי למחוק פונקציית Python UDF מתמידה, משתמשים בהצהרה DROP FUNCTION.

דוגמה

כדי לראות דוגמה ליצירת פונקציית UDF מתמשכת ב-Python, בוחרים באחת מהאפשרויות הבאות:

import bigframes.pandas as bpd

# Set BigQuery DataFrames options
bpd.options.bigquery.project = your_gcp_project_id
bpd.options.bigquery.location = "US"

# BigQuery DataFrames gives you the ability to turn your custom functions
# into a BigQuery Python UDF. One can find more details about the usage and
# the requirements via `help` command.
help(bpd.udf)

# Read a table and inspect the column of interest.
df = bpd.read_gbq("bigquery-public-data.ml_datasets.penguins")
df["body_mass_g"].peek(10)

# Define a custom function, and specify the intent to turn it into a
# BigQuery Python UDF. Let's try a `pandas`-like use case in which we want
# to apply a user defined function to every value in a `Series`, more
# specifically bucketize the `body_mass_g` value of the penguins, which is a
# real number, into a category, which is a string.
@bpd.udf(
    dataset=your_bq_dataset_id,
    name=your_bq_routine_id,
)
def get_bucket(num: float) -> str:
    if not num:
        return "NA"
    boundary = 4000
    return "at_or_above_4000" if num >= boundary else "below_4000"

# Then we can apply the udf on the `Series` of interest via
# `apply` API and store the result in a new column in the DataFrame.
df = df.assign(body_mass_bucket=df["body_mass_g"].apply(get_bucket))

# This will add a new column `body_mass_bucket` in the DataFrame. You can
# preview the original value and the bucketized value side by side.
df[["body_mass_g", "body_mass_bucket"]].peek(10)

# The above operation was possible by doing all the computation on the
# cloud through an underlying BigQuery Python UDF that was created to
# support the user's operations in the Python code.

# The BigQuery Python UDF created to support the BigQuery DataFrames
# udf can be located via a property `bigframes_bigquery_function`
# set in the udf object.
print(f"Created BQ Python UDF: {get_bucket.bigframes_bigquery_function}")

# If you have already defined a custom function in BigQuery, either via the
# BigQuery Google Cloud Console or with the `udf` decorator,
# or otherwise, you may use it with BigQuery DataFrames with the
# `read_gbq_function` method. More details are available via the `help`
# command.
help(bpd.read_gbq_function)

existing_get_bucket_bq_udf = get_bucket.bigframes_bigquery_function

# Here is an example of using `read_gbq_function` to load an existing
# BigQuery Python UDF.
df = bpd.read_gbq("bigquery-public-data.ml_datasets.penguins")
get_bucket_function = bpd.read_gbq_function(existing_get_bucket_bq_udf)

df = df.assign(body_mass_bucket=df["body_mass_g"].apply(get_bucket_function))
df.peek(10)

# Let's continue trying other potential use cases of udf. Let's say we
# consider the `species`, `island` and `sex` of the penguins sensitive
# information and want to redact that by replacing with their hash code
# instead. Let's define another scalar custom function and decorate it
# as a udf. The custom function in this example has external package
# dependency, which can be specified via `packages` parameter.
@bpd.udf(
    dataset=your_bq_dataset_id,
    name=your_bq_routine_id,
    packages=["cryptography"],
)
def get_hash(input: str) -> str:
    from cryptography.fernet import Fernet

    # handle missing value
    if input is None:
        input = ""

    key = Fernet.generate_key()
    f = Fernet(key)
    return f.encrypt(input.encode()).decode()

# We can use this udf in another `pandas`-like API `map` that
# can be applied on a DataFrame
df_redacted = df[["species", "island", "sex"]].map(get_hash)
df_redacted.peek(10)

# If the BigQuery routine is no longer needed, we can clean it up
# to free up any cloud quota
session = bpd.get_global_session()
session.bqclient.delete_routine(f"{your_bq_dataset_id}.{your_bq_routine_id}")

סטטוס ה-build של הקונטיינר

כשיוצרים פונקציה מוגדרת על ידי המשתמש (UDF) ב-Python באמצעות ההצהרה CREATE FUNCTION,‏ BigQuery יוצר או מעדכן קובץ אימג' של קונטיינר שמבוסס על קובץ אימג' בסיסי. הקונטיינר נוצר על בסיס תמונת הבסיס באמצעות הקוד שלכם ותלות החבילות שצוינו.

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

הגודל של כל קובצי ה-UDF של Python בכל פרויקט ובכל אזור מוגבל לסכום כולל של 10GiB. מידע נוסף זמין במאמר בנושא מגבלות על פונקציות בהגדרת משתמש (UDF) עבור פונקציות מתמידות בהגדרת משתמש. אם הפרויקט הגיע למכסה, ה-build של הקונטיינר ייכשל.

כדי לראות את הסטטוס של בניית מאגר התגים, בוחרים באחת מהאפשרויות הבאות:

יצירת פונקציה מוגדרת על ידי המשתמש (UDF) ב-Python עם וקטוריזציה

אפשר להטמיע את פונקציית ה-UDF של Python כדי לעבד קבוצה של שורות במקום שורה אחת, באמצעות וקטוריזציה. הפיכת נתונים לווקטורים יכולה לשפר את ביצועי השאילתות. אפשר ליצור פונקציה בהגדרת המשתמש (UDF) עם וקטוריזציה באמצעות Pandas או Apache Arrow.

כדי לשלוט בהתנהגות של העיבוד באצווה, מציינים את המספר המקסימלי של שורות בכל אצווה באמצעות האפשרות max_batching_rows ברשימת האפשרויות CREATE OR REPLACE FUNCTION. אם מציינים את max_batching_rows, ‏ BigQuery קובע את מספר השורות בחבילה, עד למגבלה של max_batching_rows. אם לא מציינים את max_batching_rows, מספר השורות שיצורפו לחבילה נקבע באופן אוטומטי.

שימוש ב-Pandas

לפונקציית UDF וקטורית של Python יש ארגומנט יחיד pandas.DataFrame שחייב להיות עם הערה. לארגומנט pandas.DataFrame יש אותו מספר עמודות כמו לפרמטרים של פונקציית ה-UDF ב-Python שהוגדרו בהצהרה CREATE FUNCTION. לשמות העמודות בארגומנט pandas.DataFrame יש שמות זהים לפרמטרים של הפונקציה המוגדרת על ידי המשתמש.

הפונקציה צריכה להחזיר pandas.Series או pandas.DataFrame עם מספר שורות זהה למספר השורות של הקלט.

בדוגמה הבאה נוצרת פונקציית UDF וקטורית של Python בשם multiplyInputs עם שני פרמטרים – x ו-y:

1. עוברים לדף BigQuery.

   כניסה ל-BigQuery

2. בעורך השאילתות, מזינים את ההצהרה הבאה של CREATE FUNCTION:

   CREATE FUNCTION `PROJECT_ID.DATASET_ID`.multiplyVectorized(x FLOAT64, y FLOAT64)
   RETURNS FLOAT64
   LANGUAGE python
   OPTIONS(runtime_version="python-3.11", entry_point="vectorized_multiply")
   AS r'''
   import pandas as pd
   
   def vectorized_multiply(df: pd.DataFrame):
     return df['x'] * df['y']
   
   ''';

   החלפה של PROJECT_ID.‫DATASET_ID במזהה הפרויקט ובמזהה מערך הנתונים.

   הקריאה לפונקציה UDF זהה לקריאה בדוגמה הקודמת.

3. לוחצים על play_circle_filled הפעלה.

שימוש ב-Apache Arrow

בדוגמה הבאה נעשה שימוש בממשק RecordBatch של Apache Arrow. כשמשתמשים בממשק RecordBatch, הפונקציה מעבירה לנקודת הכניסה אצווה של שורות של עמודות באורך שווה. בדוגמה הבאה נעשה שימוש ב-Apache Arrow כדי ליצור פונקציית UDF וקטורית של Python בשם multiplyVectorizedArrow.

1. עוברים לדף BigQuery.

   כניסה ל-BigQuery

2. בעורך השאילתות, מזינים את ההצהרה הבאה של CREATE FUNCTION:

   CREATE FUNCTION `PROJECT_ID.DATASET_ID`.multiplyVectorizedArrow(x FLOAT64, y FLOAT64)
   RETURNS FLOAT64
   LANGUAGE python
   OPTIONS(
     runtime_version="python-3.11",
     entry_point="vectorized_multiply_arrow"
   )
   AS r'''
   import pyarrow as pa
   import pyarrow.compute as pc
   
   def vectorized_multiply_arrow(batch: pa.RecordBatch):
       # Access columns directly from the Arrow RecordBatch
       x = batch.column('x')
       y = batch.column('y')
   
       # Use pyarrow.compute for vectorized operations
       return pc.multiply(x, y)
   ''';

   החלפה של PROJECT_ID.‫DATASET_ID במזהה הפרויקט ובמזהה מערך הנתונים.

   הקריאה לפונקציה המוגדרת על ידי המשתמש זהה לדוגמאות הקודמות.

3. לוחצים על play_circle_filled הפעלה.

קריאה לפונקציה מוגדרת על ידי המשתמש (UDF) ב-Python

אם יש לכם הרשאה להפעיל UDF של Python, אתם יכולים לקרוא לה כמו לכל פונקציה אחרת. כדי להשתמש בפונקציה שמוגדרת בפרויקט אחר, צריך להשתמש בשם המלא של הפונקציה. לדוגמה, כדי להפעיל פונקציה לחילוץ XML בשם cw_xml_extract בפרויקט אחר, מבצעים את השלבים הבאים.

import textwrap
from typing import Tuple

import bigframes.pandas as bpd
import pandas as pd
import pyarrow as pa


# Using partial ordering mode enables more efficient query optimizations.
bpd.options.bigquery.ordering_mode = "partial"


def call_python_udf(
    project_id: str, location: str,
) -> Tuple[pd.Series, bpd.Series]:
    # Set the billing project to use for queries. This step is optional, as the
    # project can be inferred from your environment in many cases.
    bpd.options.bigquery.project = project_id  # "your-project-id"

    # Since this example works with local data, set a processing location.
    bpd.options.bigquery.location = location  # "US"

    # Create a sample series.
    xml_series = pd.Series(
        [
            textwrap.dedent(
                """
                <book id="1">
                    <title>The Great Gatsby</title>
                    <author>F. Scott Fitzgerald</author>
                </book>
                """
            ),
            textwrap.dedent(
                """
                <book id="2">
                    <title>1984</title>
                    <author>George Orwell</author>
                </book>
                """
            ),
            textwrap.dedent(
                """
                <book id="3">
                    <title>Brave New World</title>
                    <author>Aldous Huxley</author>
                </book>
                """
            ),
        ],
        dtype=pd.ArrowDtype(pa.string()),
    )
    df = pd.DataFrame({"xml": xml_series})

    # Use the BigQuery Accessor, which is automatically registered on pandas
    # DataFrames when you import bigframes.  This example uses a function that
    # has been deployed to bigquery-utils for demonstration purposes. To use in
    # production, deploy the function at
    # https://github.com/GoogleCloudPlatform/bigquery-utils/blob/master/udfs/community/cw_xml_extract.sqlx
    # to your own project.
    titles_pandas = df.bigquery.sql_scalar(
        "`bqutil`.`fn`.cw_xml_extract({xml}, '//title/text()')",
    )

    # Alternatively, call read_gbq_function to get a pointer to the function
    # that can be applied on BigQuery DataFrames objects.
    cw_xml_extract = bpd.read_gbq_function("bqutil.fn.cw_xml_extract")
    xml_bigframes = bpd.read_pandas(xml_series)

    xpath_query = "//title/text()"
    titles_bigframes = xml_bigframes.apply(cw_xml_extract, args=(xpath_query,))
    return titles_pandas, titles_bigframes

סוגי נתונים נתמכים של פונקציות מוגדרות על ידי המשתמש (UDF) ב-Python

בטבלה הבאה מוגדר המיפוי בין סוגי נתונים ב-BigQuery, סוגי נתונים ב-Python וסוגי נתונים ב-Pandas:

גרסאות זמן ריצה נתמכות

פונקציות UDF ב-Python ב-BigQuery תומכות בסביבת זמן הריצה python-3.11. גרסת Python הזו כוללת כמה חבילות נוספות שהותקנו מראש. בספריות מערכת, בודקים את תמונת הבסיס של זמן הריצה.

שימוש בחבילות של צד שלישי

אפשר להשתמש ברשימת האפשרויות CREATE FUNCTION כדי להשתמש במודולים אחרים מאלה שסופקו על ידי ספריית התקנים של Python וחבילות שהותקנו מראש. אפשר להתקין חבילות מאינדקס החבילות של Python ‏ (PyPI), או לייבא קובצי Python מ-Cloud Storage.

התקנת חבילה מאינדקס חבילות Python

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

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

בדוגמה הבאה אפשר לראות איך יוצרים פונקציית UDF ב-Python שמתקינה את חבילת scipy באמצעות רשימת האפשרויות CREATE OR REPLACE FUNCTION:

1. עוברים לדף BigQuery.

   כניסה ל-BigQuery

2. בעורך השאילתות, מזינים את ההצהרה הבאה של CREATE FUNCTION:

   CREATE FUNCTION `PROJECT_ID.DATASET_ID`.area(radius FLOAT64)
   RETURNS FLOAT64 LANGUAGE python
   OPTIONS (entry_point='area_handler', runtime_version='python-3.11', packages=['scipy==1.15.3'])
   AS r"""
   import scipy
   
   def area_handler(radius):
     return scipy.constants.pi*radius*radius
   """;
   
   SELECT `PROJECT_ID.DATASET_ID`.area(4.5);

   החלפה של PROJECT_ID.‫DATASET_ID במזהה הפרויקט ובמזהה מערך הנתונים.

3. לוחצים על play_circle_filled הפעלה.

ייבוא קובצי Python נוספים כספריות

אפשר להרחיב את הפונקציות המוגדרות על ידי המשתמש ב-Python באמצעות רשימת האפשרויות של הפונקציה על ידי ייבוא קובצי Python מ-Cloud Storage.

בקוד Python של פונקציית UDF, אפשר לייבא את קובצי Python מ-Cloud Storage כמודולים באמצעות הצהרת הייבוא, ואחריה הנתיב לאובייקט Cloud Storage. לדוגמה, אם מייבאים את gs://BUCKET_NAME/path/to/lib1.py, הצהרת הייבוא תהיה import path.to.lib1.

שם הקובץ ב-Python צריך להיות מזהה Python. כל שם של folder בשם האובייקט (אחרי /) צריך להיות מזהה תקין של Python. בטווח ASCII ‏ (U+0001..U+007F), אפשר להשתמש בתווים הבאים במזהים:

• אותיות רישיות וקטנות באנגלית, מ-A עד Z.
• קווים תחתונים.
• הספרות אפס עד תשע, אבל מספר לא יכול להופיע כתו הראשון במזהה.

בדוגמה הבאה מוצג איך ליצור פונקציית UDF ב-Python שמייבאת את חבילת ספריית הלקוח lib1.py מקטגוריה של Cloud Storage בשם my_bucket:

1. עוברים לדף BigQuery.

   כניסה ל-BigQuery

2. בעורך השאילתות, מזינים את ההצהרה הבאה של CREATE FUNCTION:

   CREATE FUNCTION `PROJECT_ID.DATASET_ID`.myFunc(a FLOAT64, b STRING)
   RETURNS STRING LANGUAGE python
   OPTIONS (
   entry_point='compute', runtime_version='python-3.11',
   library=['gs://BUCKET_NAME/PATH/lib1.py'])
   AS r"""
   import path.to.lib1 as lib1
   
   def compute(a, b):
     # doInterestingStuff is a function defined in
     # gs://BUCKET_NAME/PATH/lib1.py
     return lib1.doInterestingStuff(a, b);
   
   """;

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

   • PROJECT_ID: מזהה הפרויקט.
   • ‫DATASET_ID: מזהה מערך הנתונים.
   • ‫BUCKET_NAME: השם של קטגוריית Cloud Storage שמכילה את lib1.py.
   • ‫PATH: הנתיב לקטגוריה של Cloud Storage.

3. לוחצים על play_circle_filled הפעלה.

הגדרת מגבלות קונטיינר לפונקציות מוגדרות על ידי משתמש (UDF) ב-Python

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

כברירת מחדל, המשאבים הבאים מוקצים למאגרי תגים:

• הזיכרון שהוקצה הוא 512Mi.
• המעבד שהוקצה הוא 1.0 vCPU.
• מגבלת הבו-זמניות (concurrency) של בקשות למאגר היא 80.

בדוגמה הבאה נוצרת פונקציה מוגדרת על ידי המשתמש (UDF) ב-Python באמצעות האפשרות CREATE FUNCTION list כדי לציין מגבלות של קונטיינר:

1. עוברים לדף BigQuery.

   כניסה ל-BigQuery

2. בעורך השאילתות, מזינים את ההצהרה הבאה של CREATE FUNCTION:

   CREATE FUNCTION `PROJECT_ID.DATASET_ID`.square_area(length FLOAT64)
   RETURNS FLOAT64 LANGUAGE python
   OPTIONS (entry_point='square_area', runtime_version='python-3.11',
   container_memory='CONTAINER_MEMORY', container_cpu=CONTAINER_CPU, container_request_concurrency=CONTAINER_REQUEST_CONCURRENCY)
   AS r"""
   def square_area(length):
     return length*length
   """;
   
   SELECT `PROJECT_ID.DATASET_ID`.square_area(4.5);

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

   • ‫PROJECT_ID.DATASET_ID: מזהה הפרויקט ומזהה מערך הנתונים.
   • ‫CONTAINER_MEMORY: ערך הזיכרון בפורמט הבא: <integer_number><unit>. היחידה חייבת להיות אחת מהערכים הבאים: Mi (MiB),‏ M (MB),‏ Gi (GiB) או G (GB). לדוגמה, 2Gi.
   • ‫CONTAINER_CPU: ערך ה-CPU. פונקציות UDF ב-Python תומכות בערכי מעבד חלקיים בין 0.33 ל-1.0 ובערכי מעבד לא חלקיים של 1,‏ 2 ו-4.
   • ‫CONTAINER_REQUEST_CONCURRENCY: המספר המקסימלי של בקשות בו-זמניות לכל מופע של קונטיינר של פונקציה מוגדרת על ידי המשתמש (UDF) ב-Python. הערך צריך להיות מספר שלם בין 1 ל-1000.

3. לוחצים על play_circle_filled הפעלה.

ערכי מעבד נתמכים

פונקציות UDF ב-Python תומכות בערכי CPU חלקיים בין 0.33 ל-1.0 ובערכי CPU לא חלקיים של 1,‏ 2 ו-4. אפשר להגדיר קונטיינרים שמריצים פונקציות מוגדרות על ידי המשתמש (UDF) ב-Python עד 4 vCpu. ערך ברירת המחדל הוא 1.0. ערכי קלט חלקיים מעוגלים לשתי ספרות אחרי הנקודה העשרונית לפני שהם מוחלים על מאגר התגים.

ערכי זיכרון נתמכים

קונטיינרים של פונקציות מוגדרות על ידי המשתמש ב-Python תומכים בערכי זיכרון בפורמט הבא: <integer_number><unit>. היחידה צריכה להיות אחת מהערכים הבאים: Mi, ‏ M, ‏ Gi,‏ G. כמות הזיכרון המינימלית שאפשר להגדיר היא 256Mi. הכמות המקסימלית של הזיכרון שאפשר להגדיר היא 16Gi.

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

לחלופין, אם קבעתם את כמות ה-CPU שהקציתם, תוכלו להשתמש בטבלה הבאה כדי לקבוע את טווח הזיכרון המתאים:

התקשרות Cloud de Confiance by S3NS לשירותים באינטרנט בקוד Python

פונקציית UDF ב-Python ניגשת ל Cloud de Confiance by S3NS שירות או לשירות חיצוני באמצעות חשבון השירות של חיבור למשאב ב-Cloud. צריך להעניק לחשבון השירות של החיבור הרשאות גישה לשירות. ההרשאות הנדרשות משתנות בהתאם לשירות שאליו ניגשים ולממשקי ה-API שמבוצעות להם קריאות מקוד Python.

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

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

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

בדוגמה הבאה אפשר לראות איך ניגשים לשירות Cloud Translation מפונקציית UDF של Python. בדוגמה הזו יש שני פרויקטים – פרויקט בשם my_query_project שבו יוצרים את הפונקציה המוגדרת על ידי המשתמש ואת קישור למשאבים ב-Cloud, ופרויקט שבו מפעילים את Cloud Translation בשם my_translate_project.

יצירת קישור למשאבים ב-Cloud

קודם כול, יוצרים קישור למשאבים ב-Cloud ב-my_query_project. כדי ליצור את החיבור למשאב בענן, פועלים לפי השלבים הבאים.

import google.api_core.exceptions
from google.cloud import bigquery_connection_v1

client = bigquery_connection_v1.ConnectionServiceClient()


def create_connection(
    project_id: str,
    location: str,
    connection_id: str,
):
    """Creates a BigQuery connection to a Cloud Resource.

    Cloud Resource connection creates a service account which can then be
    granted access to other Google Cloud resources for federated queries.

    Args:
        project_id: The Google Cloud project ID.
        location: The location of the connection (for example, "us-central1").
        connection_id: The ID of the connection to create.
    """

    parent = client.common_location_path(project_id, location)

    connection = bigquery_connection_v1.Connection(
        friendly_name="Example Connection",
        description="A sample connection for a Cloud Resource.",
        cloud_resource=bigquery_connection_v1.CloudResourceProperties(),
    )

    try:
        created_connection = client.create_connection(
            parent=parent, connection_id=connection_id, connection=connection
        )
        print(f"Successfully created connection: {created_connection.name}")
        print(f"Friendly name: {created_connection.friendly_name}")
        print(
            f"Service Account: {created_connection.cloud_resource.service_account_id}"
        )

    except google.api_core.exceptions.AlreadyExists:
        print(f"Connection with ID '{connection_id}' already exists.")
        print("Please use a different connection ID.")
    except Exception as e:
        print(f"An unexpected error occurred while creating the connection: {e}")

const {ConnectionServiceClient} =
  require('@google-cloud/bigquery-connection').v1;
const {status} = require('@grpc/grpc-js');

const client = new ConnectionServiceClient();

/**
 * Creates a new BigQuery connection to a Cloud Resource.
 *
 * A Cloud Resource connection creates a service account that can be granted access
 * to other Google Cloud resources.
 *
 * @param {string} projectId The Google Cloud project ID. for example, 'example-project-id'
 * @param {string} location The location of the project to create the connection in. for example, 'us-central1'
 * @param {string} connectionId The ID of the connection to create. for example, 'example-connection-id'
 */
async function createConnection(projectId, location, connectionId) {
  const parent = client.locationPath(projectId, location);

  const connection = {
    friendlyName: 'Example Connection',
    description: 'A sample connection for a Cloud Resource',
    // The service account for this cloudResource will be created by the API.
    // Its ID will be available in the response.
    cloudResource: {},
  };

  const request = {
    parent,
    connectionId,
    connection,
  };

  try {
    const [response] = await client.createConnection(request);

    console.log(`Successfully created connection: ${response.name}`);
    console.log(`Friendly name: ${response.friendlyName}`);

    console.log(`Service Account: ${response.cloudResource.serviceAccountId}`);
  } catch (err) {
    if (err.code === status.ALREADY_EXISTS) {
      console.log(`Connection '${connectionId}' already exists.`);
    } else {
      console.error(`Error creating connection: ${err.message}`);
    }
  }
}

# This queries the provider for project information.
data "google_project" "default" {}

# This creates a cloud resource connection in the US region named my_cloud_resource_connection.
# Note: The cloud resource nested object has only one output field - serviceAccountId.
resource "google_bigquery_connection" "default" {
  connection_id = "my_cloud_resource_connection"
  project       = data.google_project.default.project_id
  location      = "US"
  cloud_resource {}
}

הענקת גישה לחשבון השירות של החיבור

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

כדי לתת לחשבון השירות של Cloud resource connection גישה לפרויקטים שלכם, צריך להקצות לחשבון השירות את התפקיד 'צרכן של שימוש בשירות' (roles/serviceusage.serviceUsageConsumer) ב-my_query_project ואת התפקיד 'משתמש ב-Cloud Translation API' (roles/cloudtranslate.user) ב-my_translate_project.

יצירת פונקציית UDF ב-Python שקוראת לשירות Cloud Translation

ב-my_query_project, יוצרים פונקציית UDF ב-Python שקוראת לשירות Cloud Translation באמצעות הקישור למשאבים ב-Cloud.

1. במסוף Cloud de Confiance , עוברים לדף BigQuery.

   כניסה ל-BigQuery

2. מזינים את הצהרת CREATE FUNCTION הבאה בעורך השאילתות:

   CREATE FUNCTION `PROJECT_ID.DATASET_ID`.translate_to_es(x STRING)
   RETURNS STRING LANGUAGE python
   WITH CONNECTION `PROJECT_ID.REGION.CONNECTION_ID`
   OPTIONS (entry_point='do_translate', runtime_version='python-3.11', packages=['google-cloud-translate>=3.11', 'google-api-core'])
   AS r"""
   
   from google.api_core.retry import Retry
   from google.cloud import translate
   
   project = "my_translate_project"
   translate_client = translate.TranslationServiceClient()
   
   def do_translate(x : str) -> str:
   
       response = translate_client.translate_text(
           request={
               "parent": f"projects/PROJECT_ID/locations/us-central1",
               "contents": [x],
               "target_language_code": "es",
               "mime_type": "text/plain",
           },
           retry=Retry(),
       )
       return response.translations[0].translated_text
   
   """;
   
   -- Call the UDF.
   WITH text_table AS
     (SELECT "Hello" AS text
     UNION ALL
     SELECT "Good morning" AS text
     UNION ALL
     SELECT "Goodbye" AS text)
   SELECT text,
   `PROJECT_ID.DATASET_ID`.translate_to_es(text) AS translated_text
   FROM text_table;

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

   • ‫PROJECT_ID: מזהה הפרויקט.
   • ‫DATASET_ID: מזהה קבוצת הנתונים.
   • ‫REGION: האזור של החיבור.
   • ‫CONNECTION_ID: מזהה החיבור.

3. לוחצים על play_circle_filled הפעלה.

   הפלט אמור להיראות כך:

   +--------------------------+-------------------------------+
   | text                     | translated_text               |
   +--------------------------+-------------------------------+
   | Hello                    | Hola                          |
   | Good morning             | Buen dia                      |
   | Goodbye                  | Adios                         |
   +--------------------------+-------------------------------+
   

שימוש ב-VPC Service Controls

פונקציות UDF ב-Python מקבלות בירושה את גבולות הגזרה של VPC Service Controls של הפרויקט שמריץ את עבודת השאילתה. גבולות גזרה אלה מגנים על העבודות שלכם מפני חילוץ נתונים, ומבטיחים שהאינטראקציות עם השירות יהיו מאובטחות.

כשמפעילים פונקציית UDF של Python בתוך גבולות הגזרה של VPC Service Controls, יש לה את קישוריות הרשת הבאה:

• פונקציות UDF ב-Python שלא משתמשות בקישור למשאבים ב-Cloud מבודדות לחלוטין. כל התנועה היוצאת חסומה.
• פונקציות UDF ב-Python שמשתמשות בקישור למשאבים ב-Cloud חסומות לגישה לאינטרנט הציבורי. פונקציות UDF של Python יכולות לגשת רק לשירותים של Cloud de Confiance by S3NS תומכים ב-VPC Service Controls. תנועה יוצאת לכל יעד אחר חוץ מ-restricted.googleapis.com נחסמת.

הגדרת פונקציות מוגדרות על ידי המשתמש (UDF) ב-Python כדי לגשת לשירותים בצורה מאובטחת בתוך VPC Service Controls Cloud de Confiance by S3NS

כדי לגשת לשירותים מ-UDF של Python תוך אכיפה של VPC Service Controls, פועלים לפי השלבים הבאים: Cloud de Confiance by S3NS

1. יוצרים את פונקציית ה-UDF של Python באמצעות הסעיף WITH CONNECTION של הצהרת CREATE FUNCTION.
2. כוללים את פרויקט BigQuery שבו מופעלת משימת השאילתה ואת פרויקט השירות של היעד בגבולות גזרה לשירות. אפשרות אחרת היא להגדיר גשר בין גבולות גזרה.
3. מוסיפים את ה-API של שירות היעד להגדרת גבול הגזרה. לדוגמה, translate.googleapis.com אם מתחברים ל-Cloud Translation API.

פרטים נוספים על הגדרת גבולות גזרה של VPC Service Controls זמינים במאמרים הבאים:

• VPC Service Controls ל-BigQuery
• סקירה כללית על VPC Service Controls

שיטות מומלצות

כשיוצרים פונקציות UDF ב-Python, מומלץ לפעול לפי השיטות המומלצות הבאות:

• אופטימיזציה של הלוגיקה של השאילתות כדי לאגד אותן. מבני שאילתות מורכבים יכולים להשבית את האפשרות של הוספה לאוסף. הפעולה הזו גורמת לעיבוד איטי של כל שורה בנפרד, מה שמגדיל באופן משמעותי את זמן האחזור במערכי נתונים גדולים.
• אופטימיזציה של מטען הנתונים. גודל השורות הבודדות יכול להשפיע על היעילות של התכונה 'הוספה של כמה שורות בבת אחת'. כדאי להקפיד שכל שורה תהיה קטנה ככל האפשר כדי למקסם את מספר השורות שאפשר לעבד באצווה אחת.
• הגדרת מגבלות על קונטיינרים בצורה יעילה. מדרגיות היא פונקציה של מעבד (CPU), זיכרון ובו-זמניות בקשות. כדאי לבדוק את מדדי המעקב כדי לשנות את הגדרות הקונטיינר. אם ניצול המעבד גבוה, אפשר להגדיל את הקצאת המעבד באמצעות המגבלה container_cpu או להקטין את מקבילות הבקשות של הקונטיינר באמצעות המגבלה container_request_concurrency.
• כשמשתמשים בכוונון איטרטיבי, מתחילים עם ערכי ברירת מחדל. אם הביצועים לא אופטימליים, כדאי לנתח את מדדי הניטור כדי לזהות צווארי בקבוק ספציפיים.
• הטמעה של פסק זמן (timeout) ב-API. כשפונקציית Python UDF ניגשת לאינטרנט, צריך להגדיר זמן קצוב לתפוגה לקריאה ל-API כדי למנוע התנהגות לא צפויה. דוגמה לגישה לאינטרנט היא קריאה מקטגוריה של Cloud Storage.

הצגת מדדים של פונקציות מוגדרות על ידי המשתמש (UDF) ב-Python

פונקציות מוגדרות על ידי המשתמש (UDF) ב-Python מייצאות מדדים אל Cloud Monitoring. המדדים האלה עוזרים לכם לעקוב אחרי היבטים שונים של תקינות הפעולה וצריכת המשאבים של פונקציות מוגדרות על ידי המשתמש, ומספקים תובנות לגבי הביצועים וההתנהגות של מופעי הפונקציות המוגדרות על ידי המשתמש.

סוג המשאב למעקב

המדדים של פונקציות UDF ב-Python מדווחים בסוג המשאב הבא ב-Cloud Monitoring:

• Type (סוג): bigquery.googleapis.com/ManagedRoutineInvocation
• שם לתצוגה: BigQuery Managed Routine Invocation
• תוויות:
  • ‫resource_container: מזהה הפרויקט שבו הופעלה משימת השאילתה.
  • ‫location: המיקום שבו משימת השאילתה הופעלה.
  • ‫query_job_id: המזהה של עבודת השאילתה שהפעילה את ה-UDF של Python.
  • ‫routine_project_id: מזהה הפרויקט שבו מאוחסנת הפרוצדורה שהופעלה.
  • ‫routine_dataset_id: מזהה מערך הנתונים שבו מאוחסנת הפרוצדורה שהופעלה.
  • ‫routine_id: המזהה של השגרה שהופעלה.

מדדים

המדדים הבאים זמינים עבור סוג המשאב bigquery.googleapis.com/ManagedRoutineInvocation:

הצגת המדדים

כדי לראות את המדדים של פונקציות UDF ב-Python, בוחרים באחת מהאפשרויות בקטעים הבאים.

פרטי משרה

כדי לראות את מדדי ה-UDF של Python עבור עבודת שאילתה ספציפית, פועלים לפי השלבים הבאים:

1. עוברים לדף BigQuery.

   כניסה ל-BigQuery

2. לוחצים על היסטוריית משרות.

3. בעמודה מזהה משימה, לוחצים על מזהה משימת השאילתה.

4. בדף פרטי עבודת השאילתה, לוחצים על לוח הבקרה של Cloud Monitoring. הקישור הזה מציג לוח בקרה עם סינון של מדדי פונקציית Python UDF של העבודה.

Metrics Explorer

כדי לראות את מדדי פונקציות UDF של Python ב-Metrics Explorer:

1. עוברים לדף Metrics Explorer ב-Cloud Monitoring.

   כניסה לדף Metrics Explorer

2. לוחצים על Select a metric (בחירת מדד), ובשדה Filter (מסנן) מקלידים BigQuery Managed Routine Invocation או bigquery.googleapis.com/ManagedRoutineInvocation.

3. בוחרים באפשרות Bigquery Managed Routine > Managed_routine.

4. לוחצים על אחד מהמדדים הזמינים, כמו:

   • ניצול המעבד של מופע
   • ניצול הזיכרון של המופע
   • מספר מקסימלי של בקשות בו-זמנית

5. לוחצים על אישור.

   כברירת מחדל, המדדים מוצגים בתרשים.

6. אפשר לסנן ולקבץ את המדדים באמצעות התוויות שמוגדרות בסוגי משאבים של Monitoring. כדי לסנן את המדדים:

   1. בשדה Filter בוחרים סוג משאב כמו query_job_id או routine_id.

   2. בשדה ערך, מזינים את מזהה המשימה או מזהה השגרה, או בוחרים אחד מהם מהרשימה.

לוחות בקרה של Cloud Monitoring

כדי לראות את מדדי הפונקציות המוגדרות על ידי המשתמש ב-Python באמצעות לוחות הבקרה של המעקב:

1. נכנסים לדף Dashboards ב-Cloud Monitoring.

   מעבר למרכזי השליטה

2. לוחצים על מרכז הבקרה BigQuery Managed Routine Query Monitoring (מעקב אחרי שאילתות של שגרות מנוהלות ב-BigQuery).

   לוח הבקרה הזה מספק סקירה כללית של מדדי מפתח בפונקציות שהוגדרו על ידי המשתמש.

3. כדי לסנן את מרכז הבקרה הזה:

   1. לוחצים על addסינון.

   2. ברשימה Filter by resource, בוחרים באפשרות כמו מזהה פרויקט, מיקום, מזהה שגרה או מזהה משימה.

מיקומים נתמכים

פונקציות UDF ב-Python נתמכות בכל המיקומים במספר אזורים ובאזורים ב-BigQuery.

תמחור

החיוב על פונקציות UDF ב-Python מתבצע באמצעות מק"ט שירותי BigQuery.

החיובים כוללים את הפריטים הבאים:

• יצירה או יצירה מחדש של קובץ האימג' של הקונטיינר של ה-UDF. החיוב הזה הוא יחסי למשך הזמן שנדרש לבניית התמונה המתאימה עם קוד הלקוח והתלות.

  • אם אתם משתמשים ב-Routines API, משך הבנייה האחרון מופיע בשדה BuildStatus. אפשר גם לראות את משך הבנייה בעמודה BuildStatus בתצוגה INFORMATION_SCHEMA.ROUTINES.
  • כדי לראות את העלות הכוללת של בנייה לכל פרויקט, אפשר לסנן את דוח החיוב באמצעות האפשרויות הבאות:
    • מקרא: goog-bq-feature-type
    • ערך: MANAGED_ROUTINE_BUILD

• לקוחות שמשתמשים בפונקציות מוגדרות על ידי המשתמש (UDF) ב-Python מחויבים גם בעלות של הפעלת פונקציות כאלה. החיוב הזה הוא יחסי לכמות המחשוב והזיכרון שנצרכו כשמפעילים את פונקציית Python UDF.

  • כדי לראות את העלויות של פונקציות UDF ב-Python לכל שאילתה, אפשר להריץ שאילתה בשדה ExternalServiceCosts באמצעות Job API. אפשר גם לראות את העלויות לכל שאילתה בעמודה external_service_costs בתצוגה INFORMATION_SCHEMA.JOBS ולהחיל את המסנן הבא: 'external_service_costs.external_service="MANAGED_ROUTINE_EXECUTION"'.
  • כדי לראות את העלות הכוללת של הפעלת פונקציות מוגדרות על ידי המשתמש ב-Python לכל פרויקט, אפשר לסנן את דוח החיוב באמצעות הפרמטרים הבאים:
    • מקרא: goog-bq-feature-type
    • ערך: MANAGED_ROUTINE_EXECUTION

• אם פונקציות UDF של Python גורמות לתעבורת נתונים יוצאת ברשת חיצונית או באינטרנט, תראו גם חיוב על תעבורת נתונים יוצאת באינטרנט במסלול פרימיום, על סמך מק"טים של תעבורת נתונים יוצאת ב-BigQuery.

מכסות

מכסות ומגבלות של UDF