Important
이 기능은 공개 미리보기 단계에 있습니다. 작업 영역 관리자는 미리 보기 페이지에서 이 기능에 대한 액세스를 제어할 수 있습니다. Azure Databricks 미리 보기 관리를 참조하세요.
액세스 제어
기능은 관리 가능한 Unity 카탈로그 개체입니다. 기능에 대한 액세스는 , CREATE FEATURE및 READ FEATURE Unity 카탈로그 권한에 의해 MANAGE제어됩니다. 자세한 내용은 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삭제하고 기능을 사용하거나materialize_features해당 기능을delete_materialized_feature구체화하려면 이 기능이 필요합니다MANAGE.
모든 기능 작업에는 부모 카탈로그 및 USE CATALOG 부모 스키마에도 필요합니다USE SCHEMA. 구체화 방법 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 all sources except RequestSource: entity columns
timeseries_column: Optional[str] = None, # Required for all sources except RequestSource: 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 all sources except RequestSource: entity columns
timeseries_column: Optional[str] = None, # Required for all sources except RequestSource: timestamp column
name: Optional[str] = None, # Optional: Feature name (auto-generated if omitted)
description: Optional[str] = None, # Optional: Feature description
) -> Feature
매개 변수:
-
source: 기능 계산(DeltaTableSource또는StreamSourceRequestSource)에 사용되는 데이터 원본입니다. -
functionAggregationFunction: 연산자(예Sum(input="amount"): ), 입력 열 및 시간 창을 함께 번들로 묶는 값입니다. 또는ColumnSelection("column_name")통과 기능의 경우. -
catalog_name: 기능의 Unity 카탈로그 이름입니다. -
schema_name: 기능의 Unity 카탈로그 스키마 이름입니다. -
entity: 집계 또는 조회 키(기본 키)를 정의하는 열 이름 목록입니다. 를 제외한RequestSource모든 소스 형식에 필요합니다. 예를 들어["user_id"]사용자별로 집계 또는 조회합니다. -
timeseries_column: 시간 창 집계 또는 최신 값 선택에 사용되는 타임스탬프 열입니다. 를 제외한RequestSource모든 소스 형식에 필요합니다. -
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일 트랜잭션 수)
지원되는 함수
집계 함수
| 기능 | Description | 예제 사용 사례 |
|---|---|---|
Sum(input="column") |
값의 합계 | 사용자별 일일 앱 사용량(분) |
Avg(input="column") |
값의 평균 | 평균 트랜잭션 금액 |
Count(input="column") |
레코드 수 | 사용자당 로그인 수 |
Min(input="column") |
최소값 | 웨어러블 디바이스에서 기록한 가장 낮은 심박수 |
Max(input="column") |
최대값 | 세션당 가장 높은 트랜잭션 양 |
StddevPop(input="column") |
모집단 표준 편차 | 모든 고객에 대한 일일 트랜잭션 금액 가변성 |
StddevSamp(input="column") |
샘플 표준 편차 | 광고 캠페인 클릭률의 가변성 |
VarPop(input="column") |
모집단 분산 | 팩터리에서 IoT 디바이스에 대한 센서 판독값 분산 |
VarSamp(input="column") |
샘플 분산 | 샘플링된 그룹에 대한 영화 등급 분산 |
ApproxCountDistinct(input="column", relativeSD=0.05) |
대략적인 고유 개수 | 구매한 항목의 고유 개수 |
ApproxPercentile(input="column", percentile=0.95, accuracy=100) |
근사치 백분위수 | p95 응답 대기 시간 |
First(input="column") |
첫 번째 값 | 첫 번째 로그인 타임스탬프 |
Last(input="column") |
마지막 값 | 가장 최근 구매 금액 |
FirstN(input="column", n=3) |
첫 번째 n 값들을 배열로 보기 |
세션에서 처음 본 세 가지 제품 |
LastN(input="column", n=3) |
배열로서의 마지막 n 값들 |
최근 세 가지 지원 사례 상태 |
FirstDistinct(input="column", n=3) |
배열로서의 첫 번째 n 구별되는 값들 |
처음 세 가지 뚜렷한 제품 카테고리를 살펴보았습니다 |
LastDistinct(input="column", n=3) |
배열로서의 마지막 n 구별되는 값들 |
가장 최근에 등장한 세 가지 뚜렷한 상인 카테고리 |
메모
First, , Last, FirstNLastNFirstDistinct기본적으로 LastDistinct null 값을 포함합니다. null을 건너뛰려면 null인 filter_condition 입력 열을 명시적으로 제외하는 열을 추가합니다.
FirstN
LastN
FirstDistinct, 그리고 LastDistinct 특징을 timeseries_column 사용하여 입력 행을 정렬하고 최대 값을 n 포함하는 배열을 반환합니다. 매개변수는 n 양의 정수여야 합니다.
FirstN
FirstDistinct 그리고 가장 오래된 값부터 최신 값까지 선택합니다.
LastN
LastDistinct 그리고 가장 늦은 순서대로 값을 선택한 후, 선택한 값을 타임스탬프 순서대로 반환합니다.
FirstDistinct
LastDistinct 그리고 그 방향으로 값을 선택할 때 중복된 값을 제거하세요.
예를 들어, 엔터티의 소스 행이 로 정렬 event_time["A", "A", "B", "C", "B", "B"]되어 있으면 다음 함수들이 반환됩니다:
| 기능 | Result |
|---|---|
FirstN(input="event_type", n=3) |
["A", "A", "B"] |
LastN(input="event_type", n=3) |
["C", "B", "B"] |
FirstDistinct(input="event_type", n=3) |
["A", "B", "C"] |
LastDistinct(input="event_type", n=3) |
["A", "C", "B"] |
FirstN
LastN, , FirstDistinct, 그리고 LastDistinct 0.17.0 이상의 버전을 요구 databricks-feature-engineering 합니다.
ColumnSelection (통과)
ColumnSelection 는 집계를 적용하지 않고 원본에서 단일 열을 선택합니다. 내부가 아닌 function매개 변수에 AggregationFunction 직접 래핑됩니다. 반환 형식은 원본 스키마에서 유추됩니다.
| 기능 | Description | 예제 사용 사례 |
|---|---|---|
ColumnSelection("col") |
열의 최신 값(집계 없음) | 가장 최근의 공급업체 범주, 요청 필드의 통과 |
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 절로 작동합니다.
메모
filter_condition는 이전에 적용WHERE된 SQL GROUP BY 절처럼 집계 전에 행을 필터링합니다. 기능 정의에서 항상 정의 entity 되는 세분성은 변경되지 않습니다.
필터는 기능 계산에 필요한 데이터의 상위 집합을 포함하는 큰 원본 테이블로 작업할 때 유용하며 이러한 테이블 위에 별도의 뷰를 만들 필요가 최소화됩니다.
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: 집계 전에 적용된 SQLWHERE절입니다. 예:"status = 'completed'". -
transformation_sql: 원본 테이블에 적용되는 SQLSELECT식입니다. 집계 전에 열, 캐스트 형식 또는 컴퓨팅 파생 열의 이름을 바꾸는 데 사용합니다. 생략하면 모든 열이 선택됩니다(*). 예:"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}같습니다.
메모
timeseries_column 기능 정의DeltaTableSource에 지정된 형식 또는 TimestampType형식이어야 DateType 합니다. 정수 형식은 작동하지만 시간 창 집계의 전체 자릿수가 손실될 수 있습니다.
예제: 열 변환에 사용 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(),
)
지원되는 transformation_sql 표현
와 에 DeltaTableSourceStreamSource대한 동일한 규칙이 transformation_sql 적용됩니다.
transformation_sql 모든 행별 표현식을 지원합니다; 각 행에 대해 독립적으로 평가되는 연산들. 이들은 행 수나 소스와의 일대일 대응을 변경하지 않습니다. 행별 표현식에는 열 이름 변경, 캐스트, 산술 연산 등이 포함됩니다.
형태나 행 수를 바꾸는 연산은 지원되지 않으며, 예를 들어 집계(예: SUM()COUNT())는 지원되지 않습니다. 대신 기능 정의에 사용합니다 AggregationFunction .
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] = None, # Optional: SQL WHERE clause applied before aggregation
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
)
매개 변수:
-
full_name: 스트림의 전체 세 부분으로 구성된 이름입니다(예:"my_catalog.my_schema.my_stream"). -
filter_condition(선택 사항): 점 접두사 열 참조(예WHERE: )를 사용하여 집계 전에 데이터를 스트리밍하는 데 적용되는 SQL"value.event_type = 'purchase'"절입니다. -
transformation_sql(선택): 집계나 열 선택 전에 적용되는 SQLSELECT표현식으로, 그리고value와 구조체에key점이 접두사가 붙은 참조를 사용합니다. 와 동일한 행별 표현DeltaTableSource식을 지원합니다. 생략하면 소스는 모든 열(*)을 사용합니다. -
dataframe_schema: 투영된 출력의 SparkStructTypeJSON 스키마입니다. 설정하면transformation_sql필수입니다.
from databricks.feature_engineering.entities import StreamSource
stream_source = StreamSource(
full_name="my_catalog.my_schema.my_stream",
filter_condition="value.event_type = 'purchase'",
)
Stream의 인제스팅 테이블에 대해 투영을 실행하여 와 구조체를 노출 keyvalue 시켜 도출합니다dataframe_schema.
transformation_sql = (
"value.amount * value.conversion_rate AS converted_amount, "
"struct(value.user_id AS user_id, value.event_time AS time) AS event"
)
ingestion_table = "my_catalog.my_schema.events_ingestion"
dataframe_schema = spark.sql(
f"SELECT {transformation_sql} FROM {ingestion_table} LIMIT 0"
).schema.json()
stream_source = StreamSource(
full_name="my_catalog.my_schema.my_stream",
transformation_sql=transformation_sql,
dataframe_schema=dataframe_schema,
)
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스칼라 형식을 지원합니다. 배열, 맵 및 구조체와 같은 복합 형식은 지원되지 않습니다.
요청 데이터의 수화 방법
| 컨텍스트 | 작동 방식 |
|---|---|
교육 (create_training_set) |
열은 레이블이 지정된 DataFrame에서 추출됩니다. 형식은 선언된 스키마에 대해 유효성을 검사합니다. 불일치는 오류를 발생합니다(암시적 캐스팅 없음). |
| 서비스 (모델 엔드포인트) | 열은 HTTP 요청에서 또는 dataframe_records HTTP 요청에서 dataframe_split 가져옵니다. JSON 값은 선언된 형식(예: JSON 번호 → DOUBLE)으로 캐스팅됩니다. |
모델 서명
기능을 log_model 포함하는 RequestSource 학습 집합을 사용하여 RequestSource 모델을 기록하면 필요한 입력으로 MLflow 모델 서명에 열이 추가됩니다. 즉, 서비스 엔드포인트의 API 스키마는 호출자가 유추 시간에 제공해야 하는 필드를 반영합니다.
학습 및 유추 API
create_training_set 및 score_batch 원본 데이터에서 요청 시 지정 시간 올바른 기능 값을 계산합니다. 델타 테이블 원본의 슬라이딩 윈도우 집계와 같이 오프라인 구체화를 지원하는 기능의 경우 먼저 오프라인 저장소로 기능을 구체화하면 두 작업의 성능이 향상됩니다. 구체화된 오프라인 기능을 사용할 수 있는 경우 작업은 원본에서 기능 값을 다시 계산하는 대신 미리 계산된 오프라인 데이터를 읽습니다. 오프라인 저장소에 기능을 구체화하려면 기능 뷰 구체화 를 참조하세요.
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()
기간
기능 보기는 세 가지 창 유형을 지원하여 시간 창 기반 집계에 대한 조회 동작을 제어합니다. 롤링, 연속 및 슬라이딩.
- 롤링 창은 이벤트 시간에서 되돌아봅니다. 기간 및 지연은 명시적으로 정의됩니다.
- 텀블링 창은 고정되고 겹치지 않는 시간 창입니다. 각 데이터 포인트는 정확히 하나의 창에 속합니다.
- 슬라이딩 윈도우는 구성 가능한 슬라이드 간격으로 겹치는 회전형 시간 창입니다.
다음 그림에서는 작동 방식을 보여 줍니다.
롤링 창
메모
RollingWindow 은 이전에 이름이 지정되었습니다 ContinuousWindow. 이전 SDK 버전에서 마이그레이션하는 경우 그에 따라 가져오기를 업데이트합니다.
롤링 창은 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(전용)
| 매개 변수 | Constraints |
|---|---|
delay(선택 사항) |
≥ 0이어야 합니다(평가 타임스탬프에서 창을 뒤로 이동). 이벤트가 생성되는 시간과 이벤트 타임스탬프 사이의 시스템 지연을 고려하여 향후 이벤트가 학습 데이터 세트로 누출되는 것을 방지하는 데 사용합니다 delay . 예를 들어 이벤트가 생성되는 시간 사이에 1분의 지연이 있고 이러한 이벤트가 결국 타임스탬프가 할당된 원본 테이블에 착륙하는 경우 지연이 발생합니다 timedelta(minutes=1). |
window_duration |
> 0이어야 합니다. |
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
다음 표에서는 텀블링 윈도우에 대한 매개 변수를 나열합니다.
| 매개 변수 | Constraints |
|---|---|
window_duration |
> 0이어야 합니다. |
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
다음 표에서는 슬라이딩 윈도우에 대한 매개 변수를 나열합니다.
| 매개 변수 | Constraints |
|---|---|
window_duration |
> 0이어야 합니다. |
slide_duration |
> 0과 <window_duration이어야 합니다. |
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
특징 또는 집계 특징()에 사용 TableTriggerColumnSelection 되며, AggregationFunction.DeltaTableSource 파이프라인은 업스트림 델타 테이블이 새 커밋을 받을 때마다 실행됩니다.
집계 기능의 경우, 파이프라인이 제한되어 모든 커밋에서 실행되지 않습니다. 파이프라인은 기능 창 길이의 절반마다 최대 한 번씩 실행되며, 5분 간격 이상은 없습니다. 예를 들어, 1시간 동안 펼쳐지는 장편은 최대 30분에 한 번, 8시간 창을 가진 장편은 최대 4시간에 한 번 방영됩니다. 5분 바닥은 창의 절반이 그보다 작을 때 적용되므로, 10분 이하의 창은 최대 5분마다 한 번씩 실행됩니다. 5분 미만의 집계 기능은 사용할 TableTrigger수 없으니, 대신 스트리밍 트리거를 사용하세요.
from databricks.feature_engineering.entities import TableTrigger
trigger = TableTrigger()
StreamingMode
에서 지원되는 StreamingMode기능에 사용합니다StreamSource. 파이프라인은 연속 스트리밍 파이프라인으로 실행됩니다.
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(),
)
트리거 선택
각 기능은 하나의 트리거를 사용합니다; 기능 유형별 옵션은 다음과 같습니다:
| 기능 유형 | Trigger | 실행 시 |
|---|---|---|
다음의 집계(AggregationFunction) DeltaTableSource |
CronSchedule |
고정된 cron 일정 |
다음의 집계(AggregationFunction) DeltaTableSource |
TableTrigger |
각 원본 테이블 커밋에서 |
ColumnSelection (DeltaTableSource에서) |
TableTrigger |
각 원본 테이블 커밋에서 |
다음의 기능 StreamSource |
StreamingMode |
연속 스트리밍 |
단일 materialize_features 호출에서 다른 트리거 유형이 필요한 기능을 구체화할 수 없습니다. 대신 별도의 호출을 실행합니다.