환경 버전 호환성

Important

Lakeflow 파이프라인의 환경 버전은 공개 미리보기 상태입니다.

환경 버전 집합이 있는 파이프라인은 Spark Connect 통해 Python 코드를 실행합니다. 이 페이지에서는 호환되지 않는 점, 다르게 동작하는 점, 그리고 Databricks가 영향을 받은 패턴을 파이프라인에서 어떻게 스캔하는지 다룹니다.

Limitations

환경 버전은 아직 모든 파이프라인 기능과 호환되지 않습니다. 파이프라인의 Python 코드에서 다음 중 어느 것을 수행하면 환경 버전 집합이 있는 파이프라인 실행이 실패합니다.

• 파이프라인 데코레이터로 데코레이팅된 함수 내에서 Spark 세션 상태를 변경합니다. 예를 들어 spark.conf.set(...), spark.sql("USE CATALOG ...")및 createOrReplaceTempView.
• Spark Connect에서 사용할 수 없는 PySpark API(예SparkContext: , RDDSQLContext및 모든 Py4J API)를 사용합니다. Spark Connect에서 지원되는 내용을 참조하세요.

파이프라인에서 환경 버전을 사용하도록 설정하면 실패하는 경우 환경 버전을 사용하지 않도록 설정하면 파이프라인이 이전 상태로 반환됩니다.

동작 변경

Spark Connect는 클래식 PySpark 런타임과 적은 수의 동작 차이가 있습니다. 전체 참조는 Spark Connect 및 클래식 Spark 를 참조하세요. 호환성 스캔은 이러한 패턴을 미리 감지하고 마이그레이션을 차단하여 해당 패턴이 해결될 때까지 차단하여 생산 데이터에 영향을 미치기 전에 찾아내고 수정할 수 있습니다.

파이프라인에서 동작이 다를 수 있는 가장 일반적인 상황은 다음과 같습니다.

• 인터리브된 DataFrame 생성 및 세션 변형
• 변경 가능한 Python 상태를 참조하는 UDF

인터리브된 DataFrame 생성 및 세션 변형

파이프라인이 DataFrame을 생성한 다음 Spark 세션 상태를 변경하거나(예: 기본 카탈로그 또는 스키마를 변경하거나, 구성을 설정하거나, 임시 보기를 대체하거나, UDF를 다시 등록하는 경우) DataFrame을 사용합니다.

• 환경 버전이 없으면 DataFrame은 변형 전 세션 상태를 사용합니다.
• 환경 버전에서 DataFrame은 변형 후 세션 상태를 사용합니다.

다음은 그 예입니다.

from pyspark import pipelines as dp

spark.createDataFrame([(1, "Original Row")], ["id", "data"]) \
  .createOrReplaceTempView("my_view")

df = spark.sql("SELECT * FROM my_view")

spark.createDataFrame([(2, "Replaced Row")], ["id", "data"]) \
  .createOrReplaceTempView("my_view")

@dp.materialized_view
def mytable():
  return df

환경 버전 mytable[(1, "Original Row")]이 없으면 . 환경 버전에는 .가 mytable 포함됩니다 [(2, "Replaced Row")].

변경 가능한 Python 상태를 참조하는 UDF

UDF가 UDF가 정의된 후 값이 변경되는 Python 전역 변수를 참조하는 경우:

• 환경 버전이 없으면 UDF는 변수의 최신 값을 사용합니다.
• 환경 버전에서 UDF는 UDF가 정의되었을 때 값을 사용합니다.

다음은 그 예입니다.

from pyspark import pipelines as dp
from pyspark.sql.functions import col, udf

suffix = "a"

@udf
def my_udf(s):
  return s + suffix

suffix = "b"

@dp.materialized_view
def my_mv():
  return spark.createDataFrame([("alex",)], ["name"]).select(my_udf(col("name")))

환경 버전 my_mv[("alex_b",)]이 없으면 . 환경 버전에는 .가 my_mv 포함됩니다 [("alex_a",)].

파이프라인이 두 패턴 중 하나를 사용하는 경우 환경 버전을 사용하도록 설정하기 전에 감사합니다.

호환성 검사

호환성 스캔은 환경 버전에 따라 다른 결과를 낼 수 있는 코드 패턴을 파이프라인에서 찾아내어, 파이프라인이 자동으로 마이그레이션되기 전에 이를 수정할 수 있습니다. 파이프라인에서 검사를 사용하도록 설정한 경우:

• 각 업데이트는 감지된 패턴마다 파이프라인 이벤트 로그에 하나의 BehaviorChangeInSparkConnectWARN 이벤트를 생성합니다.
• 파이프라인은 환경 버전으로 이전되지 않으며, 이전 성공적인 업데이트에서 발생한 모든 호환성 경고가 해결될 때까지 직접 환경 버전을 활성화할 수 없습니다.

이 체크는 이전 업데이트가 없거나 이미 환경 버전이 설정된 파이프라인에는 적용되지 않습니다.

파이프라인에서 검사 사용

파이프라인 구성을 추가하여 pipelines.environmentVersion.enableCompatibilityScan 호환성 검사를 사용하도록 설정할 수 있습니다. 파이프라인 편집기 UI를 통해 또는 파이프라인 구성 JSON에 항목을 추가하여 구성을 추가할 수 있습니다.

UI를 통해:

1. 파이프라인 편집기에서 설정을 클릭합니다.
2. 파이프라인 설정에서 구성 섹션을 찾습니다.
3. 을 클릭합니다.구성을 추가합니다.
4. 키와 pipelines.environmentVersion.enableCompatibilityScan 값으로 입력 true 합니다.
5. 파이프라인 설정을 저장합니다.

파이프라인 JSON에서:

블록에 다음 항목을 configuration 추가합니다.

"configuration": {
  "pipelines.environmentVersion.enableCompatibilityScan": "true"
}

호환성 경고를 검토하고 해결하세요

파이프라인에서 환경 버전을 차단하는 패턴을 찾고 제거하려면:

1. 파이프라인을 사전 실행 모드로 실행한 후, 파이프라인 이벤트 로그에서 이벤트를 조회 하세요 BehaviorChangeInSparkConnectWARN . 각 이벤트는 검색된 패턴을 보고합니다. 문제 코드, 예제 패턴 및 제안된 수정 사항의 전체 목록은 호환성 이벤트 참조 를 참조하세요.
2. 제안된 수정 후에 탐지된 패턴을 제거하도록 파이프라인 코드를 업데이트한 후 파이프라인을 다시 실행하세요.
3. 이 과정을 반복하여 성공적인 업데이트가 더 이상 호환성 이벤트를 발생시키지 않을 때까지 반복합니다. 파이프라인은 자동으로 마이그레이션할 수 있고, 직접 환경 버전을 활성화할 수도 있습니다.

환경 버전을 활성화할 때는 Databricks가 파이프라인을 자동으로 마이그레이션하는지 아니면 사용자가 직접 설정 environment_version 하는지에 대한 동일한 안전 검사를 수행합니다. 해결되지 않은 호환성 경고가 있는 파이프라인은 경고가 해결될 때까지 환경 버전으로 전환되지 않습니다. 마이그레이션이 안전하게 완료되지 않거나 어떤 이유로든 실패하면 데이터를 쓰기 전에 중단되고 파이프라인은 이전 런타임에서 계속 실행됩니다.

이러한 이유로 업데이트가 중단될 경우, 파이프라인 이벤트 로그와 업데이트 오류 메시지는 원인과 해결 단계를 설명합니다. 그 단계를 따라 파이프라인을 다시 실행하여 마이그레이션을 완료하세요. 호환성 경고가 오탐이라고 생각되면 플래그된 패턴을 해결하거나 Azure Databricks 지원팀에 문의하세요.

호환성 이벤트 참조

호환성 스캔이 파이프라인에서 실행될 때, 감지된 패턴마다 파이프라인 이벤트 로그에 하나의 BehaviorChangeInSparkConnectWARN 이벤트가 생성됩니다. 이전 성공적인 업데이트에서 패턴이 감지되면, 패턴이 해결될 때까지 파이프라인은 환경 버전으로 이전되지 않습니다.

각 이벤트는 검색된 내용을 식별하는 단일 문제 코드를 보고합니다. 코드를 조회하려면 문제 코드 테이블에서 찾을 수 있습니다. 각 행은 예제 패턴과 제안된 수정 사항이 포함된 범주 섹션에 연결됩니다.

이벤트 셰이프

BehaviorChangeInSparkConnect 이벤트는 표준 파이프라인 이벤트 로그 스키마를 따릅니다.

• event_type은 behavior_change_in_spark_connect입니다.
• level은 WARN입니다.
• details에는 단일 behavior_change_in_spark_connect 필드가 issue 있는 개체가 포함됩니다. 문제 값은 아래에 나열된 코드 중 하나입니다.
• message 는 감지된 패턴에 대한 사람이 읽을 수 있는 설명입니다.

문제 코드

카테고리	문제 코드	Description
데이터베이스 및 카탈로그 변형	USE_CATALOG_OUTSIDE_QUERY_FUNCTION_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR	DataFrame을 만든 후 기본 카탈로그가 변경되었습니다. 기존 DataFrame은 새 기본 카탈로그를 사용하여 테이블을 확인할 수 있습니다.
데이터베이스 및 카탈로그 변형	USE_CATALOG_OUTSIDE_QUERY_FUNCTION_COULD_CHANGE_BEHAVIOR	USE CATALOG 는 파이프라인 데코레이터에 의해 데코레이팅된 함수 외부에서 호출되었습니다. 후속 작업에 대해 기본 카탈로그가 예기치 않게 변경 될 수 있습니다.
데이터베이스 및 카탈로그 변형	USE_DATABASE_OUTSIDE_QUERY_FUNCTION_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR	DataFrame을 만든 후 기본 데이터베이스가 변경되었습니다. 기존 DataFrame은 새 기본 데이터베이스를 사용하여 테이블을 확인할 수 있습니다.
데이터베이스 및 카탈로그 변형	USE_DATABASE_OUTSIDE_QUERY_FUNCTION_COULD_CHANGE_BEHAVIOR	USE DATABASE 는 파이프라인 데코레이터에 의해 데코레이팅된 함수 외부에서 호출되었습니다. 후속 작업에 대해 기본 데이터베이스가 예기치 않게 변경 될 수 있습니다.
흐름 함수 내에서 즉시 실행	CHECKPOINT_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED	흐름 함수는 검사점 명령을 호출합니다.
흐름 함수 내에서 즉시 실행	CREATE_DATAFRAME_VIEW_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED	흐름 함수는 데이터 프레임 보기(createOrReplaceTempView 또는 이와 유사)를 즉시 만듭니다.
흐름 함수 내에서 즉시 실행	CREATE_RESOURCE_PROFILE_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED	흐름 함수는 리소스 프로필을 만듭니다.
흐름 함수 내에서 즉시 실행	GET_RESOURCES_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED	흐름 함수가 호출하거나 관련 리소스 API를 호출 spark.resources 합니다.
흐름 함수 내에서 즉시 실행	MERGE_INTO_TABLE_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED	흐름 함수는 대상 테이블에서 즉시 MERGE INTO 수행합니다.
흐름 함수 내에서 즉시 실행	ML_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED	흐름 함수는 즉시 Spark ML 작업을 수행합니다.
흐름 함수 내에서 즉시 실행	REGISTER_DATA_SOURCE_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED	흐름 함수는 Python 데이터 원본을 등록합니다.
흐름 함수 내에서 즉시 실행	STREAMING_QUERY_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED	흐름 함수는 활성 스트리밍 쿼리 핸들에서 작동합니다.
흐름 함수 내에서 즉시 실행	STREAMING_QUERY_LISTENER_BUS_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED	흐름 함수는 스트리밍 쿼리 수신기를 등록하거나 제거합니다.
흐름 함수 내에서 즉시 실행	STREAMING_QUERY_MANAGER_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED	흐름 함수는 스트리밍 쿼리를 관리하기 위해 호출 spark.streams 합니다.
흐름 함수 내에서 즉시 실행	WRITE_OPERATION_V2_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED	흐름 함수는 즉시 DataFrameWriterV2 작업을 수행합니다.
흐름 함수 내에서 즉시 실행	WRITE_OPERATION_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED	흐름 함수는 즉시 DataFrame.write 작업을 수행합니다.
흐름 함수 내에서 즉시 실행	WRITE_STREAM_OPERATION_START_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED	흐름 함수는 스트리밍 쿼리(writeStream.start())를 시작합니다.
Spark 구성 변경	CHANGE_CONF_INSIDE_QUERY_FUNCTION_NOT_SUPPORTED	spark.conf.set() 또는 spark.conf.unset() 파이프라인 데코레이터에 의해 데코레이팅된 함수 내에서 호출되었습니다. 환경 버전에서는 지원되지 않습니다.
Spark 구성 변경	SET_CONF_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR	spark.conf.set() 는 DataFrame을 만든 후 파이프라인 데코레이터에 의해 데코레이팅된 함수 외부에서 호출되었습니다. 구성 변경은 실행 시 기존 DataFrame에 영향을 줄 수 있습니다.
Spark 구성 변경	UNSET_CONF_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR	spark.conf.unset() 는 DataFrame을 만든 후 파이프라인 데코레이터에 의해 데코레이팅된 함수 외부에서 호출되었습니다. 구성 변경은 실행 시 기존 DataFrame에 영향을 줄 수 있습니다.
임시 보기 대체	REPLACE_GLOBAL_TEMP_VIEW_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR	전역 임시 보기는 참조하는 DataFrame을 만든 후에 대체되었습니다. 교체는 기존 DataFrame에 반영될 수 있습니다.
임시 보기 대체	REPLACE_TEMP_VIEW_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR	임시 보기는 참조하는 DataFrame을 만든 후에 대체되었습니다. 교체는 기존 DataFrame에 반영될 수 있습니다.
UDF 및 UDTF 돌연변이	OVERWRITE_SESSION_UDF_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR	UDF를 참조하는 DataFrame을 만든 후 동일한 이름으로 UDF가 다시 등록되었습니다. 기존 DataFrame은 새 UDF 정의를 사용할 수 있습니다.
UDF 및 UDTF 돌연변이	OVERWRITE_SESSION_UDTF_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR	UDTF를 참조하는 DataFrame을 만든 후 동일한 이름으로 UDTF가 다시 등록되었습니다. 기존 DataFrame은 새 UDTF 정의를 사용할 수 있습니다.
UDF 및 UDTF 돌연변이	UDF_REFERENCES_GLOBAL_VARIABLE_COULD_CHANGE_BEHAVIOR	UDF는 전역 변경 가능한 Python 변수를 참조합니다. 환경 버전을 사용하면 UDF는 호출 시간이 아니라 UDF가 정의될 때 변수 값을 사용합니다.
UDF 및 UDTF 돌연변이	UDTF_REFERENCES_GLOBAL_VARIABLE_COULD_CHANGE_BEHAVIOR	UDTF는 전역 변경 가능한 Python 변수를 참조합니다. 환경 버전에서 UDTF는 호출 시간이 아니라 UDTF가 정의될 때 변수 값을 사용합니다.

데이터베이스 및 카탈로그 변형

이러한 문제는 파이프라인 코드가 기본 데이터베이스 또는 카탈로그를 변경할 때 내보내집니다. 환경 버전을 사용하면 변경 전에 생성된 DataFrame이 새 데이터베이스 또는 카탈로그를 사용하여 테이블을 확인할 수 있습니다.

이벤트를 트리거하는 예제 패턴:

from pyspark import pipelines as dp

spark.sql("USE CATALOG marketing")
df = spark.read.table("events")

spark.sql("USE CATALOG sales")  # changes the default catalog after df was created

@dp.materialized_view
def events_summary():
  return df.groupBy("region").count()

환경 버전 df 이 없으면 카탈로그에서 events 확인됩니다marketing. 환경 버전을 df 사용하면 카탈로그에서 events 확인됩니다sales.

제안된 수정 사항: 확인이 기본 카탈로그 또는 데이터베이스에 의존하지 않도록 테이블 이름을 정규화하고 DataFrame을 만들고 사용하는 사이에 기본 카탈로그 또는 데이터베이스를 변경하지 않도록 합니다.

from pyspark import pipelines as dp

df = spark.read.table("marketing.default.events")

@dp.materialized_view
def events_summary():
  return df.groupBy("region").count()

Spark 구성 변경

이러한 문제는 파이프라인 코드가 환경 버전에서 DataFrame 동작을 변경할 수 있는 방식으로 Spark 구성을 변경할 때 내보내집니다.

이벤트를 트리거하는 예제 패턴:

from pyspark import pipelines as dp

df = spark.read.table("events")

spark.conf.set("spark.sql.ansi.enabled", "true")  # changes session conf after df was created

@dp.materialized_view
def events_strict():
  return df.selectExpr("CAST(price AS INT) AS price")

환경 버전이 없으면 캐스트는 DataFrame 생성 시 conf 값을 사용합니다. 환경 버전에서는 캐스트가 잘못된 입력을 사용하고 spark.sql.ansi.enabled=true 실패할 수 있습니다.

제안된 수정 사항: DataFrame을 만들기 전에 파이프라인 파일의 맨 위에 필요한 모든 Spark 구성을 설정합니다. 쿼리별 구성의 경우 파이프라인 사양에서 파이프라인의 configuration 설정을 사용합니다.

임시 보기 대체

이러한 문제는 파이프라인 코드가 데이터 프레임을 참조하는 임시 뷰를 만든 후 대체될 때 내보내집니다. 환경 버전을 사용하면 기존 데이터 프레임에 새 보기 내용이 반영될 수 있습니다.

이벤트를 트리거하는 예제 패턴:

from pyspark import pipelines as dp

spark.createDataFrame([(1, "Original Row")], ["id", "data"]) \
  .createOrReplaceTempView("my_view")

df = spark.sql("SELECT * FROM my_view")

spark.createDataFrame([(2, "Replaced Row")], ["id", "data"]) \
  .createOrReplaceTempView("my_view")

@dp.materialized_view
def mytable():
  return df

환경 버전 mytable[(1, "Original Row")]이 없으면 . 환경 버전에는 .가 mytable 포함됩니다 [(2, "Replaced Row")].

제안된 수정 사항: 각 임시 보기를 한 번 만들고 대체하지 않습니다. 관련 데이터가 있는 여러 보기가 필요한 경우 각각 고유한 이름을 지정합니다.

UDF 및 UDTF 돌연변이

이러한 문제는 파이프라인 코드가 환경 버전에서 동작을 변경하는 방식으로 UDF 또는 UDTF를 변경할 때 내보내집니다.

이벤트를 트리거하는 예제 패턴:

from pyspark import pipelines as dp
from pyspark.sql.functions import col, udf

suffix = "a"

@udf
def my_udf(s):
  return s + suffix

suffix = "b"

@dp.materialized_view
def my_mv():
  return spark.createDataFrame([("alex",)], ["name"]).select(my_udf(col("name")))

환경 버전 my_mv[("alex_b",)]이 없으면 . 환경 버전에는 .가 my_mv 포함됩니다 [("alex_a",)].

수거 수정: 값을 Python 전역에서 캡처하는 대신 인수로 UDF에 전달하거나 UDF를 정의하기 전에 전역을 설정하고 나중에 변경하지 않습니다.

from pyspark import pipelines as dp
from pyspark.sql.functions import col, lit, udf

@udf
def append_suffix(s, suffix):
  return s + suffix

@dp.materialized_view
def my_mv():
  return spark.createDataFrame([("alex",)], ["name"]).select(append_suffix(col("name"), lit("b")))

흐름 함수 내에서 즉시 실행

이러한 문제는 파이프라인 코드가 파이프라인 데코레이터(@table@materialized_view등)로 데코레이팅된 함수 내에서 즉시 Spark 명령을 수행할 때 내보내집니다. 흐름 함수는 DataFrame을 정의하고 반환해야 합니다. 데이터를 작성하거나, 스트리밍 쿼리를 관리하거나, 리소스를 등록하거나, ML 작업을 실행하는 즉시 실행 명령은 환경 버전이 설정된 흐름 함수 내에서 허용되지 않습니다.

제안된 수정 사항: 즉시 실행 작업을 흐름 함수 외부로 이동하고 대신 흐름 함수에서 DataFrame을 반환합니다. 테이블에 쓰거나 스트리밍 쿼리를 시작하는 것과 같은 부작용은 파이프라인 정의 외부에 속합니다. 파이프라인 엔진은 흐름 함수에서 반환된 DataFrame의 구체화를 처리합니다.

이벤트 로그에서 호환성 이벤트 찾기

다음 쿼리는 파이프라인에 대한 모든 호환성 이벤트를 반환하며 가장 최근 순서로 정렬됩니다.

SELECT
  timestamp,
  message,
  details:behavior_change_in_spark_connect:issue AS issue
FROM event_log(<pipeline-id>)
WHERE event_type = 'behavior_change_in_spark_connect'
  AND level = 'WARN'
ORDER BY timestamp DESC;

최근 업데이트에서 발급 코드별로 이벤트를 계산하려면 다음을 수행합니다.

SELECT
  details:behavior_change_in_spark_connect:issue AS issue,
  COUNT(*) AS occurrences
FROM event_log(<pipeline-id>)
WHERE event_type = 'behavior_change_in_spark_connect'
  AND level = 'WARN'
GROUP BY 1
ORDER BY occurrences DESC;

이벤트 로그를 쿼리하는 방법은 이벤트 로그 쿼리를 참조하세요.

추가 리소스

• 파이프라인용 환경 버전 구성 — 기능 개요, 자동 마이그레이션, 그리고 환경 버전을 직접 활성화하는 방법.
• 파이프라인 이벤트 로그 스키마 - 전체 파이프라인 이벤트 로그 스키마입니다.
• 파이프라인 이벤트 로그 - 파이프라인 이벤트 로그를 쿼리하는 방법입니다.