환경 버전 호환성

환경 버전 집합이 있는 파이프라인은 Spark Connect 통해 Python 코드를 실행합니다. 이 페이지에서는 호환되지 않는 항목, 다르게 동작하는 동작, 영향을 받는 패턴에 대해 파이프라인을 검사하는 방법 및 기존 파이프라인을 마이그레이션하는 방법을 설명합니다.

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",)].

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

호환성 검사

호환성 검사를 사용하면 환경 버전에서 다른 결과를 생성하는 코드 패턴을 파이프라인에서 찾을 수 있습니다. 검사가 옵트인(opt-in)입니다. 파이프라인에서 검사를 사용하도록 설정한 경우:

• 각 파이프라인 실행은 감지된 패턴당 파이프라인 이벤트 로그에 하나의 BehaviorChangeInSparkConnectWARN 이벤트를 내보낸다.
• 이전에 성공한 업데이트의 모든 호환성 경고를 해결할 때까지 파이프라인에서 환경 버전을 사용하도록 설정할 수 없습니다.

검사를 사용하도록 설정하지 않으면 이벤트가 내보내지지 않으며 environment_version 사용이 차단되지 않습니다. Databricks는 파이프라인에서 환경 버전을 사용하도록 설정하기 전에 검색을 사용하도록 설정하고 검색된 패턴을 확인하는 것이 좋습니다.

파이프라인에서 검사 사용

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

UI를 통해:

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

파이프라인 JSON에서:

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

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

권장 워크플로

1. 파이프라인에서 검사를 사용하도록 설정합니다.
2. 파이프라인 실행을 시작하세요.
3. 이벤트에 대한 파이프라인 이벤트 로그 를 BehaviorChangeInSparkConnectWARN 쿼리합니다. 문제 코드, 예제 패턴 및 제안된 수정 사항의 전체 목록은 호환성 이벤트 참조 를 참조하세요.
4. 파이프라인 코드를 업데이트하여 검색된 패턴을 제거하고 더 이상 이벤트가 내보내지지 않을 때까지 파이프라인을 다시 실행합니다.
5. 파이프라인에서 환경 버전 사용의 방법 중 하나를 사용하여 파이프라인에 추가 environment_version 합니다.

호환성 경고가 가양성이며 어쨌든 사용하도록 설정 environment_version 하려는 경우 파이프라인 구성에서 항목을 제거 pipelines.environmentVersion.enableCompatibilityScan 하여 검사를 무시합니다. (값을 false 설정할 수 없으므로 항목을 완전히 제거해야 합니다.)

실행 전 검사는 이전 업데이트가 없는 파이프라인 또는 환경 버전이 이미 설정된 파이프라인에서 실행되지 않습니다.

기존 파이프라인을 환경 버전으로 마이그레이션

아직 환경 버전을 사용하지 않는 기존 파이프라인을 마이그레이션하려면 이 엔드 투 엔드 워크플로를 따릅니다. Spark Connect에서 다르게 동작할 수 있는 코드 패턴을 찾고, 수정하고, 환경 버전을 안전하게 롤아웃하는 방법을 안내합니다.

1. 파이프라인에서 호환성 검사를 사용하도록 설정합니다. 호환성 검사에 설명된 대로 파이프라인에서 검사를 사용하도록 설정합니다. 이것이 감지된 패턴이 이벤트 로그에 노출되는 원인이며, 사용 시도를 보호하도록 사전 실행 검사를 가능하게 하는 원인입니다.

2. 파이프라인 실행을 트리거하고 호환성 이벤트를 검토합니다. 일반 파이프라인 업데이트를 트리거합니다. 성공적으로 완료되면 파이프라인 이벤트 로그 BehaviorChangeInSparkConnectWARN 에서 이벤트를 쿼리합니다. 각 이벤트는 검색된 패턴을 보고합니다. 문제 코드, 예제 패턴 및 제안된 수정 사항의 전체 목록은 호환성 이벤트 참조 를 참조하세요.

3. 검색된 패턴을 해결하도록 파이프라인 코드를 업데이트합니다. 검색된 각 패턴에 대해 제안된 수정 사항에 따라 파이프라인 코드를 업데이트합니다. 각 변경 후 다른 파이프라인 업데이트를 트리거하고 해당 이벤트가 더 이상 표시되지 않는지 확인합니다. 이벤트 로그가 성공적인 업데이트에 대한 호환성 이벤트를 더 이상 표시하지 않을 때까지 반복합니다.

4. 파이프라인에서 환경 버전을 사용하도록 설정합니다. 가장 최근의 성공적인 업데이트에 호환성 이벤트가 없으면 파이프라인에서 환경 버전 사용에서 설명한 대로 UI, API 또는 번들을 사용하여 파이프라인에 추가 environment_version 합니다. 다음 업데이트는 Spark Connect 및 고정된 Python 언어 버전 및 사전 설치된 라이브러리를 사용하여 실행됩니다.

   호환성 경고가 여전히 존재 environment_version하기 때문에 업데이트가 실패하는 경우 , 2단계로 돌아가서 나머지 경고를 다시 시도하기 전에 해결합니다.

5. 마이그레이션을 확인합니다. 환경 버전에 대한 첫 번째 업데이트가 완료되면 다음을 확인합니다.

   • create_update 이벤트 로그의 이벤트는 예상 값으로 설정된 것으로 표시됩니다environment_version.
   • 파이프라인은 예상 데이터를 생성하며 새 오류 이벤트가 나타나지 않습니다.
   • 동작 변경 내용에 설명된 미묘한 동작 차이점이 있는지 다운스트림 테이블을 스폿 확인합니다.

롤백

마이그레이션 후 파이프라인이 잘못 동작하는 경우 파이프라인 설정에서 제거 environment_version 합니다. 다음 업데이트는 이전 Python 런타임 구성으로 실행됩니다. 롤백 실행을 사용하여 디버그한 다음, 문제를 식별하고 해결한 후 2단계에서 마이그레이션을 반복합니다.

호환성 이벤트 참조

파이프라인에서 호환성 검사를 사용하도록 설정하면 SDP는 검색된 패턴별로 파이프라인 이벤트 로그에 하나의 BehaviorChangeInSparkConnectWARN 이벤트를 내보낸다. 검사를 사용하도록 설정하고 이전의 성공적인 업데이트에서 패턴을 감지한 경우 SDP는 패턴이 해결될 때까지 사용도 차단 environment_version 합니다.

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

이벤트 셰이프

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

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

문제 코드

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

이러한 문제는 파이프라인 코드가 기본 데이터베이스 또는 카탈로그를 변경할 때 내보내집니다. 환경 버전을 사용하면 변경 전에 생성된 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 이 없으면 카탈로그에서 marketing 확인됩니다events. 환경 버전을 df 사용하면 카탈로그에서 sales 확인됩니다events.

제안된 수정 사항: 확인이 기본 카탈로그 또는 데이터베이스에 의존하지 않도록 테이블 이름을 정규화하고 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;

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

참고하십시오

• 파이프라인에 대한 환경 버전 구성 - 기능 개요, 환경 버전을 사용하도록 설정하는 방법.
• 파이프라인 이벤트 로그 스키마 - 전체 파이프라인 이벤트 로그 스키마입니다.
• 파이프라인 이벤트 로그 - 파이프라인 이벤트 로그를 쿼리하는 방법입니다.