기능 뷰 API 참조

액세스 제어

기능은 관리 가능한 Unity 카탈로그 개체입니다. 기능에 대한 액세스는 , READ FEATURE및 MANAGE Unity 카탈로그 권한에 의해 CREATE FEATURE제어됩니다. 자세한 내용은 Unity 카탈로그 권한 참조를 참조하세요.

• CREATE FEATURE — 스키마에서 기능을 만드는 데 필요합니다. create_feature 및 register_feature 부모 스키마에 필요합니다 CREATE FEATURE . 최소 권한 원칙에 따라 스키마 수준에서 부여 CREATE FEATURE 합니다. 카탈로그에 부여하여 해당 카탈로그의 모든 스키마에서 기능을 만들 수 있도록 허용할 수도 있습니다.
• READ FEATURE - 기능 및 해당 데이터를 읽는 데 필요합니다. get_feature, create_training_set및 학습 또는 서비스를 위한 구체화된 기능 데이터를 읽는 데는 기능에 대한 요구가 필요합니다 READ FEATURE . READ FEATURE 스키마 또는 카탈로그에 부여된 내용은 해당 스키마 또는 카탈로그에 포함된 모든 현재 및 미래의 기능에 적용됩니다.
• MANAGE - 기능의 수명 주기 및 부여를 관리하는 데 필요합니다. 기능을 사용하여 기능을 delete_feature삭제하고 기능을 사용하거나 delete_materialized_feature해당 기능을 materialize_features 구체화하려면 이 기능이 필요합니다MANAGE.

모든 기능 작업에는 부모 카탈로그 및 USE SCHEMA 부모 스키마에도 필요합니다USE CATALOG. 구체화 방법 MANAGE 및 READ FEATURE 적용 방법은 사용 권한을 참조하세요.

기능 보기 API

Feature 생성자 및 register_feature()

권장되는 방법은 개체를 로컬로 Feature 생성하고 Unity 카탈로그에 유지하는 데 사용하는 register_feature 것입니다. 이 2단계 워크플로를 사용하면 기능을 등록하기 전에 기능(포함 create_training_set)을 실험할 수 있습니다.

Feature(
    source: DataSource,                                    # Required: DeltaTableSource, StreamSource, or RequestSource
    function: Union[AggregationFunction, ColumnSelection], # Required: Aggregation or column selection
    entity: Optional[List[str]] = None,                    # Required for aggregation: entity columns
    timeseries_column: Optional[str] = None,               # Required for aggregation: timestamp column
    name: Optional[str] = None,                            # Optional: Feature name (auto-generated if omitted)
    description: Optional[str] = None,                     # Optional: Feature description
)

FeatureEngineeringClient.register_feature() 는 Unity 카탈로그에서 로컬로 생성된 Feature 등록합니다.

FeatureEngineeringClient.register_feature(
    feature: Feature,       # Required: A Feature instance (not already registered)
    catalog_name: str,      # Required: UC catalog name
    schema_name: str,       # Required: UC schema name
) -> Feature

from databricks.feature_engineering.entities import Feature, DeltaTableSource, AggregationFunction, Sum, RollingWindow
from datetime import timedelta

# Step 1: Construct the feature locally
feature = Feature(
    source=DeltaTableSource(catalog_name="main", schema_name="store", table_name="transactions"),
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

# Step 2: Register in Unity Catalog
fe = FeatureEngineeringClient()
registered_feature = fe.register_feature(
    feature=feature,
    catalog_name="main",
    schema_name="store",
)

create_feature()

FeatureEngineeringClient.create_feature() 단일 단계에서 Unity 카탈로그에서 기능의 유효성을 검사하고, 생성하고, 즉시 등록합니다. 이 기능은 먼저 로컬에서 실험할 필요가 없는 경우에 사용합니다.

FeatureEngineeringClient.create_feature(
    source: DataSource,                                    # Required: DeltaTableSource, StreamSource, or RequestSource
    function: Union[AggregationFunction, ColumnSelection], # Required: Aggregation or column selection
    catalog_name: str,                                     # Required: The catalog name for the feature
    schema_name: str,                                      # Required: The schema name for the feature
    entity: Optional[List[str]] = None,                    # Required for aggregation: entity columns
    timeseries_column: Optional[str] = None,               # Required for aggregation: timestamp column
    name: Optional[str] = None,                            # Optional: Feature name (auto-generated if omitted)
    description: Optional[str] = None,                     # Optional: Feature description
) -> Feature

매개 변수:

• source: 기능 계산(DeltaTableSource또는StreamSourceRequestSource)에 사용되는 데이터 원본입니다.
• function AggregationFunction: 연산자(예Sum(input="amount"): ), 입력 열 및 시간 창을 함께 번들로 묶는 값입니다. 또는 ColumnSelection("column_name") 통과 기능의 경우.
• catalog_name: 기능의 Unity 카탈로그 이름입니다.
• schema_name: 기능의 Unity 카탈로그 스키마 이름입니다.
• entity: 집계 수준(기본 키)을 정의하는 열 이름 목록입니다. 집계 기능에 필요합니다. 예를 들어 ["user_id"] 사용자당 집계입니다.
• timeseries_column: 시간 창 집계에 사용되는 타임스탬프 열입니다. 집계 기능에 필요합니다.
• name: 선택적 기능 이름입니다. 생략하면 입력 열, 함수 및 창에서 자동으로 생성됩니다(예: amount_avg_rolling_7d).
• description: 기능에 대한 선택적 설명입니다.

반환: 유효성이 검사된 기능 인스턴스

발생 시킵니다: 유효성 검사에 실패하는 경우 ValueError

delete_feature()

정규화된 이름으로 Unity 카탈로그에서 기능을 삭제합니다.

FeatureEngineeringClient.delete_feature(
    full_name: str,  # Required: '<catalog>.<schema>.<feature_name>'
) -> None

fe.delete_feature(full_name="main.store.amount_sum_rolling_7d")

기능을 삭제하기 전에 해당 기능을 참조하는 모델 또는 기능 사양을 제거하거나 업데이트합니다. 기능이 구체화된 경우 먼저 구체화된 기능을 삭제합니다. 구체화된 기능을 삭제하는 방법을 참조하세요.

자동 생성된 이름

name 생략하면 이름이 자동으로 생성됩니다. 생성된 이름은 다음과 같은 패턴을 {column}_{function}_{window}따릅니다. 다음은 그 예입니다.

• price_avg_rolling_1h (1시간 평균 가격)
• transaction_count_rolling_30d_1d (이벤트 타임스탬프에서 1d 지연이 있는 30일 트랜잭션 수)

지원되는 함수

집계 함수

ColumnSelection (통과)

ColumnSelection 는 집계를 적용하지 않고 원본에서 단일 열을 선택합니다. 내부가 아닌 function매개 변수에 AggregationFunction 직접 래핑됩니다. 반환 형식은 원본 스키마에서 유추됩니다.

ColumnSelection 는 모든 데이터 원본과 함께 사용할 수 있습니다.

• DeltaTableSource: 지정 시간 조인을 통해 엔터티 키당 최신 값을 반환합니다(조회 창 집계 없음).
• StreamSource: 스트림에서 엔터티 키당 최신 값을 반환합니다(조회 창 집계 없음).
• RequestSource: 유추 시간에 제공되거나 학습 시 레이블이 지정된 DataFrame에서 추출된 값을 전달합니다.

from databricks.feature_engineering.entities import (
    ColumnSelection, DeltaTableSource, Feature, FieldDefinition,
    RequestSource, ScalarDataType,
)

delta_source = DeltaTableSource(
    catalog_name="main", schema_name="feature_store", table_name="transactions",
)

request_source = RequestSource(
    schema=[
        FieldDefinition(name="session_duration", data_type=ScalarDataType.DOUBLE),
    ]
)

# ColumnSelection from a Delta table
latest_amount = Feature(
    source=delta_source,
    function=ColumnSelection("amount"),
    entity=["user_id"],
    timeseries_column="transaction_time",
    name="latest_transaction_amount",
)

# ColumnSelection from a RequestSource
session_feature = Feature(
    source=request_source,
    function=ColumnSelection("session_duration"),
    name="session_duration",
)

예: 집계 및 열 선택 기능

다음 예제에서는 동일한 데이터 원본에 대해 정의된 기능을 보여줍니다.

from databricks.feature_engineering.entities import (
    AggregationFunction, Feature, Sum, Avg, ApproxCountDistinct,
    ColumnSelection, RollingWindow,
)
from datetime import timedelta

window = RollingWindow(window_duration=timedelta(days=7))

sum_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="event_time",
    function=AggregationFunction(Sum(input="amount"), window),
)

avg_feature = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="event_time",
    function=AggregationFunction(Avg(input="amount"), window),
)

distinct_count = Feature(
    source=source,
    entity=["user_id"],
    timeseries_column="event_time",
    function=AggregationFunction(ApproxCountDistinct(input="product_id", relativeSD=0.01), window),
)

# Column selection (no aggregation, no time window)
latest_amount = Feature(
    source=source,
    function=ColumnSelection("amount"),
    entity=["user_id"],
    timeseries_column="event_time",
    name="latest_amount",
)

필터 조건이 있는 기능

이 filter_condition 매개 변수를 사용하면 집계를 계산 하기 전에 원본 테이블의 행을 필터링할 수 있습니다. 이 함수는 데이터를 그룹화하고 집계하기 전에 적용되는 SQL WHERE 절로 작동합니다.

필터는 기능 계산에 필요한 데이터의 상위 집합을 포함하는 큰 원본 테이블로 작업할 때 유용하며 이러한 테이블 위에 별도의 뷰를 만들 필요가 최소화됩니다.

from databricks.feature_engineering.entities import AggregationFunction, Sum, Count, RollingWindow, DeltaTableSource
from datetime import timedelta

# Source with filter applied at the source level
high_value_transactions = DeltaTableSource(
    catalog_name="main",
    schema_name="ecommerce",
    table_name="transactions",
    filter_condition="amount > 100",  # Only transactions over $100
)

high_value_sales = Feature(
    source=high_value_transactions,
    entity=["user_id"],
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=30))),
)

# Multiple conditions
completed_orders_source = DeltaTableSource(
    catalog_name="main",
    schema_name="ecommerce",
    table_name="orders",
    filter_condition="status = 'completed' AND payment_method = 'credit_card'",
)

completed_orders = Feature(
    source=completed_orders_source,
    entity=["user_id"],
    timeseries_column="order_time",
    function=AggregationFunction(Count(input="order_id"), RollingWindow(window_duration=timedelta(days=7))),
)

# Filter on a StreamSource
from databricks.feature_engineering.entities import StreamSource

purchase_stream = StreamSource(
    full_name="main.ecommerce.transactions_stream",
    filter_condition="value.event_type = 'purchase'",
)

purchase_total = Feature(
    source=purchase_stream,
    entity=["value.user_id"],
    timeseries_column="value.event_time",
    function=AggregationFunction(Sum(input="value.amount"), RollingWindow(window_duration=timedelta(hours=1))),
)

데이터 원본

DeltaTableSource

DeltaTableSource는 원본 테이블에서 기능을 계산하는 방법을 정의하는 데 사용되는 임시 Python 개체입니다. 새 테이블을 만들지 않습니다. 데이터를 읽고 기능을 집계하기 위한 구성을 지정합니다.

DeltaTableSource(
    catalog_name: str,                              # Required: Catalog name
    schema_name: str,                               # Required: Schema name
    table_name: str,                                # Required: Table name
    filter_condition: Optional[str] = None,         # Optional: SQL WHERE clause to filter source data
    transformation_sql: Optional[str] = None,       # Optional: SQL SELECT expression for column transformations
    dataframe_schema: Optional[str] = None,         # Required if transformation_sql is set: schema of the resulting DataFrame
)

매개 변수:

• catalog_name, schema_nametable_name: Unity 카탈로그에서 원본 델타 테이블을 식별합니다.
• filter_condition: 집계 전에 적용된 SQL WHERE 절입니다. 예: "status = 'completed'".
• transformation_sql: 원본 테이블에 적용되는 SQL SELECT 식입니다. 집계 전에 열, 캐스트 형식 또는 컴퓨팅 파생 열의 이름을 바꾸는 데 사용합니다. 생략하면 모든 열이 선택됩니다(*). 예: "user_id, CAST(amount AS DOUBLE) AS amount, event_time".
• dataframe_schema: 변환 후 Spark StructType JSON 형식(원본 df.schema.json())의 결과 DataFrame 스키마입니다. 가 제공된 경우 transformation_sql 필요합니다. 이렇게 하면 변환으로 인해 발생하는 열 이름과 형식이 시스템에 표시됩니다.

둘 다 filter_conditiontransformation_sql 설정되면 결과 쿼리는 다음과 SELECT {transformation_sql} FROM {table} WHERE {filter_condition}같습니다.

예제: 열 변환에 사용 transformation_sql

source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="raw_events",
    transformation_sql="user_id, CAST(price_cents AS DOUBLE) / 100 AS price, event_time",
    filter_condition="event_type = 'purchase'",
    dataframe_schema=spark.sql(
        "SELECT user_id, CAST(price_cents AS DOUBLE) / 100 AS price, event_time FROM main.analytics.raw_events LIMIT 0"
    ).schema.json(),
)

예: PySpark DataFrame에서 파생 transformation_sql 및 dataframe_schema 파생

변환을 PySpark 쿼리로 작성한 다음 결과 DataFrame에서 스키마를 추출할 수 있습니다.

df = spark.sql(f"""
  SELECT user_id, CAST(amount AS DOUBLE) / 100 AS amount_dollars, event_time
  FROM main.analytics.events
  WHERE event_date >= date_sub(current_date(), 7)
  LIMIT 0
""")

# Use df.schema.json() as the dataframe_schema
source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="events",
    transformation_sql="user_id, CAST(amount AS DOUBLE) / 100 AS amount_dollars, event_time",
    filter_condition="event_date >= date_sub(current_date(), 7)",
    dataframe_schema=df.schema.json(),
)

DeltaTableSource.from_sql()

편의를 위해 SQL 쿼리에서 만들 DeltaTableSource 수 있습니다. 이 메서드는 쿼리를 구문 분석하여 테이블 이름을 transformation_sql자동으로 추출합니다 filter_condition.

DeltaTableSource.from_sql(
    sql: str,                           # Required: SQL SELECT query
    spark: SparkSession,                # Required: active SparkSession (for schema inference)
) -> DeltaTableSource

간단한 SELECT ... FROM ... [WHERE ...] 쿼리만 지원됩니다. 복합 SQL(JOIN, 하위 쿼리, CTE, UNION)이 거부됩니다. 복잡한 쿼리의 경우 다음을 사용하여 직접 DeltaTableSource 생성 transformation_sql 합니다filter_condition.

from databricks.feature_engineering.entities import (
    AggregationFunction,
    DeltaTableSource,
    Feature,
    Sum,
    TumblingWindow,
)

source = DeltaTableSource.from_sql(
    spark=spark,
    sql=f"SELECT customer_id, event_ts, amount * 2 AS doubled_amount, amount FROM {CATALOG}.{SCHEMA}.{TABLE}",
)

feature = Feature(
    source=source,
    function=AggregationFunction(Sum(input="doubled_amount"), time_window=TumblingWindow(window_duration=timedelta(days=7))),
    entity=["customer_id"], timeseries_column="event_ts",
)

다음을 사용하여 반복 to_dataframe()

기능 계산에 사용할 데이터를 미리 보는 데 사용합니다 source.to_dataframe() . 이는 예상된 결과를 생성할 때까지 반복하는 filter_conditiontransformation_sql 데 유용합니다.

source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="events",
    filter_condition="event_type = 'purchase'",
)

# Preview the filtered source data
source.to_dataframe().display()

엔터티 이해

엔터티 열은 기능에 대한 집계 수준을 정의합니다. 정의에는 Feature 지정되지 않고 .DeltaTableSource 엔터티는 다음을 결정합니다.

• 데이터 그룹화 방법: 기능은 엔터티 값의 고유한 조합별로 집계됩니다(SQL과 유사 GROUP BY ).
• 기본 키 구조: 각 고유 엔터티 조합은 계산된 기능의 한 행을 생성합니다.

예: 고객 수준 기능

다음 코드는 고객 수준에서 기능을 집계합니다(고객당 한 행).

from databricks.feature_engineering.entities import DeltaTableSource

source = DeltaTableSource(
    catalog_name="main",
    schema_name="analytics",
    table_name="user_events",
)

Feature(
    source=source,
    entity=["user_id"],                # Features aggregated per user
    timeseries_column="event_time",    # Timestamp for time windows
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

예: 고객 스토어 수준 기능

더 자세한 수준(고객 저장소 조합당 하나의 행)에서 기능을 집계하려면 여러 엔터티 열을 사용합니다.

source = DeltaTableSource(
    catalog_name="main",
    schema_name="retail",
    table_name="transactions",
)

Feature(
    source=source,
    entity=["user_id", "store_id"],  # Features aggregated per user-store pair
    timeseries_column="transaction_time",
    function=AggregationFunction(Sum(input="amount"), RollingWindow(window_duration=timedelta(days=7))),
)

다양한 집계 수준(예: 고객 수준 및 고객 저장소 수준)의 기능이 필요한 경우 기능 정의에서 다른 entity 값을 사용합니다. 엔터티 구성이 서로 다른 기능 간에 동일한 DeltaTableSource 것을 공유할 수 있습니다.

StreamSource

StreamSource 는 Stream을 참조합니다. Stream에는 스트리밍 원본에 대한 연결, 인증, 스키마 및 수집 구성이 포함됩니다. Kafka의 경우 기능 정의의 열 참조 앞에 읽거나 읽을 메시지의 일부를 나타내야 합니다 value.key. .

StreamSource(
    full_name: str,                       # Required: Three-part Stream name (catalog.schema.stream)
    filter_condition: Optional[str],      # Optional: SQL WHERE clause applied before aggregation
)

매개 변수:

• full_name: 스트림의 전체 세 부분으로 구성된 이름입니다(예: "my_catalog.my_schema.my_stream").
• filter_condition(선택 사항): 점 접두사 열 참조(예"value.event_type = 'purchase'": )를 사용하여 집계 전에 데이터를 스트리밍하는 데 적용되는 SQL WHERE 절입니다.

from databricks.feature_engineering.entities import StreamSource

stream_source = StreamSource(
    full_name="my_catalog.my_schema.my_stream",
    filter_condition="value.event_type = 'purchase'",
)

RequestSource

RequestSource 는 미리 구체화된 테이블에서 조회하지 않고 요청 페이로드에서 유추 시간에 제공되는 데이터에 대한 스키마를 정의합니다. 학습 중에 이러한 열은 전달된 레이블이 지정된 DataFrame에서 추출됩니다 create_training_set. 모델을 제공하는 동안 호출자는 HTTP 요청 페이로드에 포함해야 합니다.

RequestSource 은 (값을 직접 전달하기 위해) 함께 ColumnSelection 사용됩니다. 집계 함수 또는 시간 창은 지원하지 않습니다.

스키마 정의

스키마를 각각 열 이름과 FieldDefinition다음을 지정하는 개체 목록 ScalarDataType 으로 정의합니다.

from databricks.feature_engineering.entities import (
    FieldDefinition, RequestSource, ScalarDataType,
)

request_source = RequestSource(
    schema=[
        FieldDefinition(name="transaction_amount", data_type=ScalarDataType.DOUBLE),
        FieldDefinition(name="vendor_id", data_type=ScalarDataType.STRING),
        FieldDefinition(name="transaction_id", data_type=ScalarDataType.STRING),
        FieldDefinition(name="transaction_time", data_type=ScalarDataType.DATE),
    ]
)

지원되는 데이터 형식

RequestSource는 다음ScalarDataTypeINTEGERFLOATBOOLEANSTRINGDOUBLELONGTIMESTAMPDATE에서 정의된 SHORT스칼라 형식을 지원합니다. 배열, 맵 및 구조체와 같은 복합 형식은 지원되지 않습니다.

요청 데이터의 수화 방법

모델 서명

기능을 log_model 포함하는 RequestSource 학습 집합을 사용하여 RequestSource 모델을 기록하면 필요한 입력으로 MLflow 모델 서명에 열이 추가됩니다. 즉, 서비스 엔드포인트의 API 스키마는 호출자가 유추 시간에 제공해야 하는 필드를 반영합니다.

학습 및 유추 API

create_training_set()

특정 시점의 올바른 기능 계산을 사용하여 학습 데이터 세트를 만듭니다. 자세한 내용은 기능 보기를 사용하여 모델 학습을 참조하세요.

FeatureEngineeringClient.create_training_set(
    df: DataFrame,                                # DataFrame with training data
    features: Optional[List[Feature]],            # List of Feature objects
    label: Union[str, List[str], None],           # Label column name(s)
    exclude_columns: Optional[List[str]] = None,  # Optional: columns to exclude
) -> TrainingSet

log_model()

유추 중에 계보 추적 및 자동 기능 조회를 위한 기능 메타데이터를 사용하여 모델을 기록합니다. 자세한 내용은 기능 보기를 사용하여 모델 학습을 참조하세요.

FeatureEngineeringClient.log_model(
    model,                                    # Trained model object
    artifact_path: str,                       # Path to store model artifact
    flavor: ModuleType,                       # MLflow flavor module (e.g., mlflow.sklearn)
    training_set: TrainingSet,                # TrainingSet used for training
    registered_model_name: Optional[str],     # Optional: register model in Unity Catalog
)

score_batch()

자동 기능 조회를 사용하여 오프라인 일괄 처리 유추를 수행합니다. 모델과 함께 저장된 기능 메타데이터를 사용하여 특정 시점의 올바른 기능을 계산하여 학습과의 일관성을 보장합니다.

FeatureEngineeringClient.score_batch(
    model_uri: str,                           # URI of logged model (e.g., "models:/catalog.schema.model/1")
    df: DataFrame,                            # DataFrame with entity keys and timestamps
) -> DataFrame

입력 DataFrame은 학습 중에 사용되는 엔터티 및 타임스레터리 열을 포함해야 합니다. 기능은 원본 데이터에서 자동으로 계산됩니다.

fe = FeatureEngineeringClient()

# Batch scoring with automatic feature lookup
predictions = fe.score_batch(
    model_uri="models:/main.ecommerce.fraud_model/1",
    df=inference_df,
)
predictions.display()

기간

기능 보기는 세 가지 창 유형을 지원하여 시간 창 기반 집계에 대한 조회 동작을 제어합니다. 롤링, 연속 및 슬라이딩.

• 롤링 창은 이벤트 시간에서 되돌아봅니다. 기간 및 지연은 명시적으로 정의됩니다.
• 텀블링 창은 고정되고 겹치지 않는 시간 창입니다. 각 데이터 포인트는 정확히 하나의 창에 속합니다.
• 슬라이딩 윈도우는 구성 가능한 슬라이드 간격으로 겹치는 회전형 시간 창입니다.

다음 그림에서는 작동 방식을 보여 줍니다.

롤링 창

롤링 창은 up-to-date 및 실시간 집계이며 일반적으로 스트리밍 데이터에 사용됩니다. 스트리밍 파이프라인에서 롤링 창은 이벤트가 들어오거나 나가는 경우와 같이 고정 길이 창의 내용이 변경되는 경우에만 새 행을 내보낸다. 학습 파이프라인에서 롤링 창 기능을 사용하는 경우 특정 이벤트의 타임스탬프 바로 앞에 있는 고정 길이 기간 기간을 사용하여 원본 데이터에 대해 정확한 지정 시간 기능 계산이 수행됩니다. 이렇게 하면 온라인 오프라인 오차 또는 데이터 유출을 방지할 수 있습니다. [T - duration, T)에서 이벤트를 집계할 때 T 의 기능입니다.

class RollingWindow(TimeWindow):
    window_duration: datetime.timedelta
    delay: Optional[datetime.timedelta] = None

다음 표에서는 롤링 창에 대한 매개 변수를 나열합니다. 창 시작 및 종료 시간은 다음과 같은 매개 변수를 기반으로 합니다.

• 시작 시간: evaluation_time - window_duration - delay (포함)
• 종료 시간: evaluation_time - delay (전용)

from databricks.feature_engineering.entities import RollingWindow
from datetime import timedelta

# Look back 7 days from evaluation time
window = RollingWindow(window_duration=timedelta(days=7))

아래 코드를 사용하여 지연된 롤링 창을 정의합니다.

# Look back 7 days, offset by 1 minute to account for data ingestion delay
window = RollingWindow(
    window_duration=timedelta(days=7),
    delay=timedelta(minutes=1)
)

롤링 창 예제

• window_duration=timedelta(days=7): 현재 평가 시간에 끝나는 7일의 조회 창을 만듭니다. 7일의 오후 2:00 이벤트의 경우 0일 오후 2:00부터 7일 오후 2:00까지, 7일 오후 2:00을 제외한 모든 이벤트가 포함됩니다.

• window_duration=timedelta(hours=1), delay=timedelta(minutes=30): 평가 시간 30분 전에 끝나는 1시간 조회 창을 만듭니다. 오후 3시 이벤트인 경우 오후 1시 30분부터 오후 2시 30분까지의 모든 이벤트가 포함됩니다. 이는 데이터 수집 지연을 고려하는 데 유용합니다.

연속 창

고정 창을 사용하여 정의된 기능을 집계하는 경우, 집계는 미리 결정된 고정 길이의 창을 통해 계산되며, 슬라이드 간격으로 진행되어 시간을 완전히 분할하는 겹치지 않는 창을 생성합니다. 결과적으로 원본의 각 이벤트는 정확히 하나의 창에 기여합니다. 특정 시간t에서 기능은 t (배타적으로) 또는 그 이전에 끝나는 창의 데이터를 집계합니다. Windows는 Unix Epoch에서 시작합니다.

class TumblingWindow(TimeWindow):
    window_duration: datetime.timedelta

다음 표에서는 텀블링 윈도우에 대한 매개 변수를 나열합니다.

from databricks.feature_engineering.entities import TumblingWindow
from datetime import timedelta

window = TumblingWindow(
    window_duration=timedelta(days=7)
)

구르는 창 예제

• window_duration=timedelta(days=5): 사전에 설정된 고정된 길이의 시간 창이 각각 5일로 만들어집니다. 예: 창 #1은 0일에서 4일차까지, 창 #2는 5일차부터 9일차까지, 창 #3은 10일차부터 14일차까지입니다. 특히 Window #1에는 0일차의 00:00:00.00부터 5일째의 00:00:00.00 이전의 타임스탬프가 있는 모든 이벤트가 포함됩니다. 각 이벤트는 정확히 하나의 창에 속합니다.

슬라이딩 윈도우

슬라이딩 윈도우를 사용하여 정의된 기능의 경우 집계는 미리 결정된 고정 길이 창을 통해 계산되어 슬라이드 간격으로 진행되어 겹치는 창을 생성합니다. 원본의 각 이벤트는 여러 창에 대한 기능 집계에 기여할 수 있습니다. 특정 시간t에서 기능은 t (배타적으로) 또는 그 이전에 끝나는 창의 데이터를 집계합니다. Windows는 Unix Epoch에서 시작합니다.

class SlidingWindow(TimeWindow):
    window_duration: datetime.timedelta
    slide_duration: datetime.timedelta

다음 표에서는 슬라이딩 윈도우에 대한 매개 변수를 나열합니다.

from databricks.feature_engineering.entities import SlidingWindow
from datetime import timedelta

window = SlidingWindow(
    window_duration=timedelta(days=7),
    slide_duration=timedelta(days=1)
)

슬라이딩 윈도우 예제

• window_duration=timedelta(days=5), slide_duration=timedelta(days=1): 겹치는 5일 창을 만들어 매번 1일씩 진행합니다. 예: 창 #1은 0일차부터 4일차까지, 창 #2는 1일차부터 5일차까지, 창 #3은 2일차부터 6일차까지입니다. 각 창에는 시작 날짜부터 00:00:00.00 종료일(포함 안 됨) 00:00:00.00 까지의 이벤트가 포함됩니다. 창이 겹치므로 단일 이벤트가 여러 창에 속할 수 있습니다(이 예제에서는 각 이벤트가 최대 5개의 서로 다른 창에 속).

구체화 트리거

구체화 파이프라인이 실행되는 경우 제어를 트리거합니다. 트리거 유형은 기능 유형에 따라 달라집니다.

CronSchedule

집계 기능(CronSchedule)에 사용합니다AggregationFunction. 파이프라인은 Quartz cron 식으로 정의된 고정된 일정으로 실행됩니다.

from databricks.feature_engineering.entities import CronSchedule

trigger = CronSchedule(
    quartz_cron_expression="0 0 * * * ?",  # Hourly
    timezone_id="UTC",
)

TableTrigger

에서 TableTrigger 지원되는 ColumnSelection기능에 사용합니다DeltaTableSource. 파이프라인은 업스트림 델타 테이블이 새 커밋을 받을 때마다 실행됩니다.

from databricks.feature_engineering.entities import TableTrigger

trigger = TableTrigger()

StreamingMode

에서 지원되는 StreamSource기능에 사용합니다StreamingMode. 파이프라인은 연속 스트리밍 파이프라인으로 실행됩니다.

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    StreamSource, Feature, AggregationFunction, Sum,
    RollingWindow, OnlineStoreConfig, StreamingMode,
)
from datetime import timedelta

fe = FeatureEngineeringClient()

stream_source = StreamSource(full_name="my_catalog.my_schema.my_stream")

streaming_feature = fe.create_feature(
    source=stream_source,
    entity=["value.user_id"],
    timeseries_column="value.event_time",
    function=AggregationFunction(
        operator=Sum(input="value.amount"),
        time_window=RollingWindow(window_duration=timedelta(hours=1)),
    ),
    catalog_name="my_catalog",
    schema_name="my_schema",
    name="user_purchase_sum",
)

fe.materialize_features(
    features=[streaming_feature],
    online_config=OnlineStoreConfig(
        catalog_name="my_catalog",
        schema_name="my_schema",
        table_name_prefix="streaming_features_serving",
        online_store_name="feature_store_online",
    ),
    trigger=StreamingMode(),
)

트리거 선택

단일 materialize_features 호출에서 다른 트리거 유형이 필요한 기능을 구체화할 수 없습니다. 대신 별도의 호출을 실행합니다.

베타 기능을 공개 미리 보기로 마이그레이션

기능 보기 공개 미리 보기는 Unity 카탈로그의 일류 기능 엔터티를 도입하며, 권한 및 READ FEATURE 권한에 따라 CREATE FEATURE 관리되며 버전 0.16.0 이상이 필요합니다databricks-feature-engineering. 베타(버전 0.15.0 포함)에서 만든 기능은 Unity 카탈로그 함수로 저장되며 모든 공개 미리 보기 기능을 지원하지 않습니다. 장기 공개 미리 보기 지원을 받으려면 버전 0.16.0을 사용하여 베타 기능을 다시 만듭니다. 기능을 다시 구체화하는 것이 아니라 삭제하고 다시 만들어야 합니다.

기능에 대한 자세한 내용은 기능 보기를 참조하세요.

실행할 작업

• 0.16.0으로 업그레이드합니다. 퍼블릭 미리 보기 기능(일괄 처리 및 스트리밍)에 필요한 클라이언트 버전입니다.
• 기능을 다시 만듭니다. 베타 기능 보기는 모든 공개 미리 보기 기능을 지원하지 않으므로 다시 구체화되지 않고 삭제하고 다시 만들어야 합니다.
• 창이 닫히기 전에 마이그레이션합니다. 기존 베타 기능은 2026년 7월 22일 이전에 마이그레이션해야 합니다.

베타 및 공개 미리 보기 기능 식별

공개 미리 보기 기능은 Unity 카탈로그의 기능 개체(예: 카탈로그 탐색기)로 표시됩니다. 베타 기능은 YAML 정의가 있는 함수로 표시됩니다. 함수로 표시되는 모든 기능은 마이그레이션해야 하는 베타 기능입니다.

베타 기능 마이그레이션

베타 기능 마이그레이션에는 다음 세 부분으로 구성됩니다.

• 이 기능을 공개 미리 보기 기능으로 다시 만듭니다.
• 기능을 다시 구체화하여 오프라인 및 온라인 테이블이 새 기능으로 다시 작성되도록 합니다.
• 마이그레이션된 기능을 확인한 후 베타 기능 및 해당 구체화를 삭제합니다.

기능 다시 생성

베타 기능을 찾고, Feature.clone() 등록되지 않은 복사본을 만들고, register_feature 각 복사본을 공개 미리 보기 기능으로 다시 등록하는 데 사용합니다list_beta_feature_views. 복제는 기능을 다시 등록할 수 있도록 등록, 카탈로그 및 스키마를 지웁니다.

이름 충돌을 방지하려면 마이그레이션된 기능을 다른 이름으로 등록하거나 베타 기능과 다른 스키마에 등록합니다. 다음 예제에서는 이름 접미사를 사용하여 원래 스키마 _migrated 의 각 기능을 다시 등록합니다.

# Update this to the catalog whose beta Feature Views you want to migrate.
CATALOG_TO_MIGRATE = "main"

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()

# 1. Find every beta Feature View in the catalog. Returns Feature objects,
#    scanned across all schemas in the catalog.
beta_features = fe.list_beta_feature_views(catalog_name=CATALOG_TO_MIGRATE)

# Keep each beta feature paired with its migrated counterpart for the next steps.
migrations = []
for beta_feature in beta_features:
    catalog_name, schema_name, leaf_name = beta_feature.full_name.split(".")
    # 2. Clone the feature as an unregistered copy, renamed with a "_migrated" suffix.
    cloned = beta_feature.clone(new_name=f"{leaf_name}_migrated")
    # 3. Re-register the clone as a Public Preview feature.
    migrated = fe.register_feature(
        feature=cloned,
        catalog_name=catalog_name,
        schema_name=schema_name,
    )
    migrations.append((beta_feature, migrated))

마이그레이션된 기능 다시 구체화

베타 기능이 구체화된 경우 새 기능으로 오프라인 및 온라인 테이블이 다시 작성되도록 공개 미리 보기 대응을 다시 구체화합니다. 마이그레이션된 기능에 대한 오프라인 및 온라인 스토어 구성을 제공하고 베타 기능의 기존 구체화에서 트리거를 다시 구성합니다.

from databricks.feature_engineering.entities import (
    CronSchedule,
    OfflineStoreConfig,
    OnlineStoreConfig,
    TableTrigger,
)

for beta_feature, migrated in migrations:
    # Inspect the beta feature's existing materializations to see what to rebuild and
    # to reconstruct the same trigger.
    trigger = None
    needs_offline = needs_online = False
    for mf in fe.list_materialized_features(feature_name=beta_feature.full_name):
        needs_online = needs_online or bool(mf.is_online)
        needs_offline = needs_offline or not mf.is_online
        # Rebuild the trigger from the materialized feature.
        if mf.cron_schedule_trigger is not None:
            trigger = CronSchedule(
                quartz_cron_expression=mf.cron_schedule_trigger.cron_expression,
                timezone_id="UTC",  # Materialized schedules run in UTC.
            )
        elif mf.table_trigger is not None:
            trigger = TableTrigger()
        elif mf.streaming_mode is not None:
            # Streaming features use StreamingMode, which can be reused as-is.
            trigger = mf.streaming_mode
    if not (needs_offline or needs_online):
        continue  # The beta feature was never materialized.

    catalog_name, schema_name, _ = migrated.full_name.split(".")
    fe.materialize_features(
        features=[migrated],
        offline_config=OfflineStoreConfig(
            catalog_name=catalog_name,
            schema_name=schema_name,
            table_name_prefix="migrated_features",
        )
        if needs_offline
        else None,
        online_config=OnlineStoreConfig(
            catalog_name=catalog_name,
            schema_name=schema_name,
            table_name_prefix="migrated_features",
            online_store_name="my_online_store",
        )
        if needs_online
        else None,
        trigger=trigger,
    )

베타 기능 삭제

마이그레이션된 기능을 확인한 후 각 베타 기능의 구체화를 삭제한 다음 베타 기능 자체를 삭제합니다.

for beta_feature, _ in migrations:
    # Delete the beta feature's materializations first.
    mfs = list(fe.list_materialized_features(feature_name=beta_feature.full_name))
    offline_mfs = [mf for mf in mfs if not mf.is_online]
    if offline_mfs:
        # Aggregation features pair an offline and online table; deleting the offline
        # materialized feature removes its paired online table too.
        for mf in offline_mfs:
            fe.delete_materialized_feature(materialized_feature=mf)
    else:
        # Online-only features (ColumnSelection, streaming) have no offline pair; delete
        # the online materialized feature directly.
        for mf in mfs:
            fe.delete_materialized_feature(materialized_feature=mf)
    # Then delete the beta feature definition.
    fe.delete_feature(full_name=beta_feature.full_name)