점진적 변환 참조

점진적 데코레이터

incremental() 데코레이터는 변환의 계산 함수를 점진적 계산을 가능하게 하는 로직으로 감싸는 데 사용할 수 있습니다:

Copied!
1
2
3
4
5
6
7
8
9
10
11
from transforms.api import transform, incremental, Input, Output

@incremental()
@transform(
    students=Input('/examples/students_hair_eye_color'), # 학생들의 머리카락 및 눈 색 정보를 입력으로 받음
    processed=Output('/examples/hair_eye_color_processed') # 처리된 정보를 출력으로 반환
)
def filter_hair_color(students, processed):
    # type: (IncrementalTransformInput, IncrementalTransformOutput) -> None
    students_df = students.dataframe('added') # 새로 추가된 학생들의 데이터프레임을 가져옴
    processed.write_dataframe(students_df.filter(students_df.hair == 'Brown')) # 머리카락 색이 갈색인 학생들만 필터링하여 쓴다

incremental() 데코레이터는 transform(), transform_df(), 또는 transform_pandas() 데코레이터를 사용하는 기존 변환을 래핑하는 데 사용될 수 있습니다. 변환에 대한 계산 함수는 점진적으로 및 비점진적으로 실행될 수 있어야 합니다. incremental() 데코레이터는 두 가지 주요 작업을 수행합니다 :

• 변환이 이전 빌드에 대한 정보를 조회할 수 있게 합니다. 이 정보를 사용하여 incremental() 데코레이터는 변환이 점진적으로 실행될 수 있는지 여부를 아래에서 설명하는 요구 사항에 따라 결정합니다.
• 입력, 출력, 그리고 컨텍스트 오브젝트를 추가 기능을 제공하는 점진적 서브 클래스로 변환합니다. 특히, TransformInput은 IncrementalTransformInput이 되고, TransformOutput은 IncrementalTransformOutput이 되며, TransformContext는 IncrementalTransformContext가 됩니다. 이 점진적 오브젝트들은 데코레이터에 의해 래핑된 변환으로 전달됩니다.

점진적 데코레이터는 다섯 가지 인자를 받습니다:

Copied!
1
2
3
4
5
6
7
8
# transforms.api.incremental 함수는 다음과 같은 파라미터들을 가집니다.

transforms.api.incremental(
        require_incremental=False,  # require_incremental 파라미터는 증분 업데이트가 필요한지에 대한 여부를 지정합니다. 기본값은 False입니다.
        semantic_version=1,  # semantic_version 파라미터는 함수의 세맨틱 버전을 설정합니다. 기본값은 1입니다.
        snapshot_inputs=None,  # snapshot_inputs 파라미터는 스냅샷 입력을 설정합니다. 기본값은 None입니다.
        allow_retention=False,  # allow_retention 파라미터는 데이터 보존을 허용할지 여부를 설정합니다. 기본값은 False입니다.
        strict_append=False)  # strict_append 파라미터는 엄격한 추가를 설정합니다. 기본값은 False입니다.

require_incremental 인수를 True로 설정하면 변환을 점진적으로 실행할 수 없는 경우 실패합니다. require_incremental=True가 지정된 경우에도 스냅샷으로 변환을 실행할 수 있는 두 가지 경우가 있습니다:

1. 결과물 중 하나가 이전에 생성된 적이 없습니다.
2. 의미론적 버전이 변경되어 스냅샷이 명시적으로 요청되었습니다.

변환이 점진적으로 실행되지 않는 원인을 디버그하려면, 드라이버 로그에서 경고 transforms.api._incremental: Not running incrementally를 찾아보세요.

:func:~transforms.api.incremental 데코레이터의 semantic_version 인수를 사용하면 다음 변환 실행을 비점진적으로 강제할 수 있습니다.

• 현재 실행의 의미론적 버전이 이전 실행의 의미론적 버전과 다른 경우, 변환이 비점진적으로 실행됩니다.
• 지정하지 않으면 의미론적 버전은 1로 설정됩니다.
• 이전 실행의 의미론적 버전이 없는 경우(예: 기존 변환을 점진적 변환으로 변환하는 경우), 값 1이 가정됩니다. 이렇게 하면 새로운 스냅샷을 요구하지 않고 변환이 점진적으로 실행되기 시작할 수 있습니다.
• 후속 변환 실행을 비점진적으로 강제하려면, @incremental() 데코레이터의 semantic_version 인수를 증가시킬 수 있습니다.
  • 의미론적 버전을 증가시킬 때는 정수만 사용해야 한다는 점에 주의하세요.

snapshot_inputs 인수를 사용하면 일부 입력을 스냅샷 입력으로 정의할 수 있습니다. 이는 비스냅샷 입력과 달리 업데이트와 삭제 수정을 지원합니다. 더 자세한 정보는 스냅샷 입력을 참조하세요.

allow_retention 인수를 true로 설정하면 Foundry Retention에 의해 입력 및 출력 데이터셋의 파일 삭제를 허용하면서 변환의 점진성을 유지할 수 있습니다.

strict_append 파라미터를 true로 설정하면, 기본 Foundry 트랜잭션 유형이 APPEND로 설정되고, 점진적 쓰기에 APPEND 트랜잭션이 사용됩니다. 쓰기 작업은 Parquet 요약 메타데이터나 Hadoop SUCCESS 파일과 같은 보조 파일조차 덮어쓰지 않을 수 있다는 점에 주의하세요. 모든 Foundry 형식의 점진적 쓰기는 이 모드를 지원해야 합니다.

중요한 정보

위에서 언급한 바와 같이, incremental() 데코레이터로 감싸진 변환의 계산 함수는 점진적으로 또는 비점진적으로 실행할 수 있어야 합니다. 기본 읽기 및 쓰기 모드(이 페이지의 나머지 부분에서 더 자세히 설명됨)는 이런 이중 로직 요구 사항을 지원할 수 있지만, 계산 컨텍스트의 is_incremental 속성을 기반으로 분기하는 것이 여전히 필요할 수 있습니다.

또 다른 중요한 점은 incremental() 데코레이터를 transform_df() 또는 transform_pandas()와 함께 사용하면 기본 읽기 및 쓰기 모드에만 액세스할 수 있다는 것입니다. 이는 추가된 출력 행이 추가된 입력 행의 함수만인 변환에 충분합니다(추가 예제 참조). 그러나 변환이 더 복잡한 로직(예: 조인, 집계, 또는 고유)을 수행하고 입력 읽기 모드나 출력 쓰기 모드를 설정해야 하는 경우에는 incremental() 데코레이터를 transform()와 함께 사용해야 합니다. transform()와 함께 점진적 데코레이터를 사용하면 읽기 및 쓰기 모드를 설정할 수 있습니다.

입력 및 결과물의 점진적 모드

IncrementalTransformInput

transforms.api.IncrementalTransformInput 객체는 선택적 읽기 모드를 취할 수 있는 dataframe() 메서드를 확장합니다.

점진적 데코레이터를 사용하여 변환을 정의하는 경우, 읽기 모드는 변환을 점진적으로 실행하느냐 아니냐에 따라 다르게 동작합니다:

기본 읽기 모드는 added입니다.

입력이 점진적 방식으로 처리되는 것이 바람직하지 않은 경우가 있음에도 불구하고 변환이 incremental()로 표시된 경우가 있습니다. 이러한 유형의 입력에 대한 읽기 모드 동작이 어떻게 다른지와 관련한 더 많은 정보를 얻으려면 스냅샷 입력을 참조하세요.

기본 출력 읽기 모드는 current이고, 사용 가능한 출력 읽기 모드는 added, current, 그리고 previous입니다. 출력 읽기 모드에 대한 더 많은 정보는 아래 섹션을 참조하세요.

점진적 변환의 성격상, 마지막 SNAPSHOT 트랜잭션에서 입력 데이터셋의 모든 과거 트랜잭션을 로드하여 입력 뷰를 구축합니다. 점진적 변환에서 점차적인 느림증이 나타나기 시작하면, 점진적 입력 데이터셋에 SNAPSHOT 빌드를 실행하는 것을 권장합니다.

IncrementalTransformOutput

transforms.api.IncrementalTransformOutput 객체는 결과물 데이터셋에 대한 읽기 및 쓰기 모드에 대한 액세스를 제공합니다. modify의 기본 쓰기 모드는 점진적 및 비점진적 빌드와 호환되는 로직을 작성하는 핵심입니다. 두 가지 쓰기 모드가 있습니다:

• modify: 이 모드는 빌드 중에 작성된 데이터로 기존 출력을 수정합니다. 예를 들어, 출력이 modify 모드인 경우 write_dataframe()를 호출하면 작성된 DataFrame이 기존 출력에 추가됩니다.
• replace: 이 모드는 빌드 중에 작성된 데이터로 출력을 완전히 대체합니다.

변환이 점진적으로 실행된다는 것은 출력의 기본 쓰기 모드가 modify로 설정된다는 것을 의미합니다. 마찬가지로 변환이 비점진적으로 실행된다는 것은 출력의 기본 쓰기 모드가 replace로 설정된다는 것을 의미합니다.

입력 DataFrame의 기본 읽기 모드가 added라는 것을 기억하세요. added의 기본 입력 읽기 모드와 modify의 기본 출력 쓰기 모드 때문에, 점진적 및 비점진적 빌드와 호환되는 로직을 작성하는 것이 훨씬 쉬워집니다:

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@incremental()
@transform(
    students=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def incremental_filter(students, processed):
    # type: (IncrementalTransformInput, IncrementalTransformOutput) -> None
    # 이전에 보지 못한 행만 읽습니다.
    new_students_df = students.dataframe()  # 이것은 students.dataframe('added')와 동일합니다.

    # 비증분 시, 모든 행을 읽고 출력을 대체합니다.
    # 증분 시, 새로운 행만 읽고 출력에 추가합니다.
    processed.write_dataframe(
        new_students_df.filter(new_students_df.hair == 'Brown')  # 갈색 머리만 필터링합니다.
    )

점진적 계산에 대한 더 복잡한 유즈케이스에서는 올바른 쓰기 모드를 계산하고 수동으로 설정해야 할 수 있습니다. 이 작업은 점진적 결과물에 대한 set_mode() 메소드를 사용하여 수행할 수 있습니다.

쓰기 모드 외에도, transforms.api.IncrementalTransformOutput는 출력 데이터셋에서 DataFrame을 읽을 수 있게 합니다. 이 작업은 선택적인 읽기 모드를 입력으로 받는 dataframe() 메소드를 사용하여 수행할 수 있습니다. 기본 읽기 모드는 current로 설정되어 있고, 다른 사용 가능한 출력 읽기 모드는 added와 previous입니다. 읽기 모드는 데이터셋의 쓰기 모드가 어떻게 설정되어 있는지에 따라 다르게 동작합니다.

이전 실행에서 데이터 읽기, 유효한 조합:

이전 출력에서 데이터를 읽으려면 변환 작업이 점진적 모드에서 실행되어야 합니다(ctx.is_incremental is True), 그렇지 않으면 데이터프레임은 비어 있습니다.

이전 데이터프레임을 얻기 위해 current를 사용할 때는 schema를 제공할 필요가 없습니다. 이는 current가 이미 빌드되어야 하는 출력의 스키마를 사용하기 때문입니다. 그러나 current 모드는 previous보다 더 취약합니다. current 모드는 다음 경우에 실패합니다:

• 변환 작업이 비점진적으로 실행되고 출력에 대해 dataframe을 호출하기 전에 write_mode을 modify로 오버라이드하지 않는 경우
• 변환이 이전에 계산된 적이 없어서 스키마를 알 수 없기 때문에 빈 DataFrame을 구성할 수 없는 경우

이전 데이터프레임을 읽을 때 제공하는 스키마는 마지막 출력의 실제 스키마와 비교됩니다. 열 유형, 열의 null 가능성 또는 열의 순서가 일치하지 않으면 예외가 발생합니다. 열의 순서가 동일하게 유지되도록 다음 구조를 사용하세요:

Copied!
1
2
3
previous = out.dataframe('previous', schema)  # schema는 pyspark.sql.types.StructType 입니다.

out.write_dataframe(df.select(schema.fieldNames()))  # schema의 필드 이름을 선택하여 데이터프레임을 작성합니다.

자세한 내용은 병합 및 교체 예제를 참조하십시오.

현재 실행에서 작성된 데이터 읽기, 유효한 조합:

의도를 더 명확하게 하기 위해 added를 선호합니다. 현재 변환에서 작성된 데이터를 읽는데 유용한 시나리오는 데이터에 대한 검사를 실행하고 검사가 통과되지 않으면 빌드를 실패시키는 것입니다. 이렇게 하면 데이터를 다시 계산하거나 Spark에 캐시할 필요 없이 검사를 실행할 수 있습니다.

IncrementalTransformContext

TransformContext 객체와 비교하여, IncrementalTransformContext 객체는 추가로 is_incremental 속성을 제공합니다. 이 속성은 변환을 점진적으로 실행할 경우 True로 설정됩니다. 즉:

• 기본 출력 쓰기 모드가 modify로 설정되고,
• 입력의 기본 읽기 모드가 added로 설정됩니다.

점진적 모드 요약

점진적 데코레이터는 읽기 모드 "previous"를 지정함으로써 이전 입력 및 출력에 액세스할 수 있게 해주어 현재 빌드를 기록적 맥락에 기반할 수 있습니다. 변환을 스냅샷 모드에서 실행하면 이전 데이터 프레임이 비어 있게 되어 처음 실행이거나 로직 또는 데이터가 크게 변경되어 재계산이 필요한 경우입니다.

그러나 가장 일반적인 경우는 입력에 대해 "added" 모드를 사용하고 출력에 대해 "modify" 모드를 사용하는 것입니다. 이러한 모드는 기본적으로 사용됩니다. 입력 데이터셋에서 새로 추가된 행을 검색한 다음 처리하여 출력 데이터셋에 추가할 수 있게 해줍니다.

출력에 행을 추가하는 대신 출력 데이터셋에 이미 존재하는 일부 행을 수정하려면, 일반적인 시나리오의 예제에서 설명한 것처럼 "replace" 모드를 사용합니다.

점진적 계산 요구 사항

갈색 머리를 가진 학생만 포함하도록 학생을 필터링하는 점진적 변환을 분석해 보겠습니다:

Copied!
1
2
3
4
5
6
7
8
@incremental()  # 이 데코레이터는 이 함수가 중단되었을 때 부터 다시 시작하도록 만듭니다.
@transform(
    students=Input('/examples/students_hair_eye_color'),  # 이 경로의 파일을 입력으로 사용합니다.
    processed=Output('/examples/hair_eye_color_processed')  # 이 경로에 출력을 저장합니다.
)
def filter_hair_color(students, processed):  # 눈 색깔을 필터링하는 함수입니다.
    students_df = students.dataframe()  # 입력을 데이터프레임으로 변환합니다.
    processed.write_dataframe(students_df.filter(students_df.hair == 'Brown'))  # 머리 색이 갈색인 학생들만 선택하여 결과를 출력합니다.

가정해봅시다. /examples/students_hair_eye_color 입력 데이터셋이 새로운 학생들의 집합으로 완전히 교체되었습니다. 여기에서 새로운 학생 집합을 이전 출력 결과에 추가하면 부정확한 출력 데이터셋이 생성됩니다. 이러한 상황은 incremental() 데코레이터가 변환을 점진적으로 실행하지 않기로 결정하는 예입니다. 변환을 점진적으로 실행하려면 다음 요구 사항이 충족되어야 합니다:

• 마지막 실행 이후 변환의 모든 non_snapshot 입력 데이터셋이 파일만 추가된 경우 또는 Foundry Retention에서 삭제되는 경우입니다.
• 입력값이 수정된 경우나 파일이 삭제된 경우, 입력값을 snapshot 입력값으로 표시합니다.
• 변환의 입력값 목록을 수정해도 빌드의 점진성이 손상되지 않습니다.
• 다중 출력 변환의 경우, 모든 출력이 마지막으로 동일한 변환을 통해 빌드되었습니다.

변환에 incremental() 데코레이터가 있지만 위의 요구 사항 중 하나라도 충족되지 않으면 변환은 자동으로 비점진적으로 실행됩니다. 이는 기본 출력 쓰기 모드가 modify 대신 replace로 설정되고 입력값이 비점진적으로 표시된다는 것을 의미합니다. 또한 변환에서 출력을 읽으면 이전 기록에 접근할 수 없으므로 빈 데이터 프레임이 반환됩니다. 마찬가지로 입력값도 비점진적으로 제공됩니다. 만약 require_incremental=True를 설정하면, 변환은 비점진적으로 실행되지 않고 실패합니다.

Append-only 입력 변경

변환은 모든 점진적 입력값이 마지막 실행 이후에 파일만 추가된 경우(APPEND 또는 UPDATE 트랜잭션을 통해) 점진적으로 실행할 수 있습니다.

반대로, 변환은 점진적 입력값 중 하나라도 다음과 같은 경우 점진적으로 실행할 수 없습니다.

• 완전히 다시 작성된 경우, 예를 들어 SNAPSHOT 트랜잭션이 있는 경우
• UPDATE 또는 DELETE 트랜잭션을 통해 수정되거나 삭제된 파일이 있는 경우

예를 들어, students_hair_eye_color의 학생 목록이 완전히 변경되면 이전에 필터링된 학생의 출력은 유효하지 않고 교체되어야 합니다.

Foundry Retention에서 삭제되는 입력값

상류 데이터셋이 무한정 커지고 오래된 행을 삭제하려면(Foundry Retention을 사용하여) 하류 계산의 점진성에 영향을 주지 않고, 해당 데이터셋에 의존하는 점진적 변환을 명시적으로 설정하여 보유된 입력값을 허용해야 합니다. 이는 transforms.api.incremental 데코레이터의 allow_retention 인수를 사용하여 수행할 수 있습니다.

• 이 필드가 True로 설정된 경우, Foundry Retention에서 생성된 모든 삭제 사항은 입력값이 점진성을 유지하는지 평가할 때 무시됩니다. 이는 Retention에서 발생하는 removed 입력값이 점진성을 저해하지 않으며, 유일하게 non-added 입력값이 보유된 행을 가진 입력값인 경우에도 변환은 계속해서 점진적으로 실행된다는 것을 의미합니다.
• 필드가 False(기본값)인 경우, 입력 데이터셋의 모든 removed 유형 변경 사항은 변환을 스냅샷으로 실행하게 합니다.

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 'incremental' 데코레이터는 데이터 변환 작업이 중단되었을 때 중간 결과를 유지하는 것을 허용합니다.
@incremental(allow_retention=True)
# 'transform' 데코레이터는 입력 및 출력 데이터를 정의하고 해당 데이터를 처리하는 함수를 정의합니다.
@transform(
    # 'students'는 입력 데이터로서, 학생들의 머리카락 색과 눈 색에 관한 데이터를 담고 있습니다.
    students=Input('/examples/students_hair_eye_color'),
    # 'processed'는 출력 데이터로서, 처리 후의 데이터를 담게 됩니다.
    processed=Output('/examples/hair_eye_color_processed')
)
# 'filter_hair_color' 함수는 학생들의 눈 색을 필터링하는 함수입니다.
def filter_hair_color(students, processed):
    # 입력 데이터를 데이터프레임 형태로 로드합니다.
    students_df = students.dataframe()
    # 머리카락 색이 갈색인 학생들만 필터링하여 'processed'에 쓴다.
    processed.write_dataframe(students_df.filter(students_df.hair == 'Brown'))

위 예시에서, 변환기가 데이터셋 /examples/students_hair_eye_color에 있는 일련의 변경 사항에 이어 실행되고, 이 변경 사항에는 Foundry Retention을 사용하여 만든 추가 변경 사항과 제거 변경 사항만 포함되어 있다면, 변환기는 점진적으로 실행됩니다. 다른 방식으로 만든 제거 변경 사항이나 수정 변경 사항이 있으면 스냅샷이 트리거됩니다.

스냅샷 입력값

입력값이 완전히 다시 작성되어도 변환의 점진성을 무효화하지 않는 시나리오가 있습니다. 예를 들어, 전화번호 국가 코드를 국가에 매핑하는 간단한 참조 데이터셋이 있고 이 데이터셋이 주기적으로 다시 작성된다고 가정해 보세요. 이 데이터셋의 변경 사항은 반드시 이전 계산의 결과를 무효화하지 않으므로, 변환을 점진적으로 실행해야 합니다.

기본적으로 위에서 설명한 것처럼, 변환은 마지막으로 실행된 이후에 입력값 중 하나가 완전히 다시 작성된 경우 점진적으로 실행할 수 없습니다. 스냅샷 입력값은 이 검사에서 제외되며, 실행 사이에 시작 트랜잭션이 다를 수 있습니다.

스냅샷 입력값은 incremental() 데코레이터의 snapshot_inputs 인수를 사용하여 설정할 수 있습니다.

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@incremental(snapshot_inputs=['country_codes'])
@transform(
    phone_numbers=Input('/examples/phone_numbers'),
    country_codes=Input('/examples/country_codes'),
    output=Output('/examples/phone_numbers_to_country')
)
def map_phone_number_to_country(phone_numbers, country_codes, output):
    # type: (TransformInput, TransformInput, TransformOutput) -> None

    # 이는 이전 실행 이후로 확인되지 않은 모든 전화번호가 될 것입니다.
    phone_numbers = phone_numbers.dataframe()
    # 이는 이전에 무엇이 확인되었는지에 관계없이 모든 국가 코드가 될 것입니다.
    country_codes = country_codes.dataframe()

    cond = [phone_numbers.country_code == country_codes.code]
    # 출력을 작성하면서 전화번호와 국가 코드를 왼쪽 외부 조인합니다.
    output.write_dataframe(phone_numbers.join(country_codes, on=cond, how='left_outer'))

스냅샷 입력값의 동작은 변환(transform)이 점진적으로 또는 비점진적으로 실행될 때 동일합니다. 따라서, added와 current 읽기 모드는 항상 전체 데이터셋을 반환합니다. 다른 모든 읽기 모드는 빈 데이터셋을 반환합니다.

입력값에 대한 변경

기존 입력값 목록을 수정할 수 있습니다. 다음의 경우에 점진적인 속성이 유지됩니다:

• 새로운 입력값 또는 새로운 스냅샷 입력값이 추가되거나,
• 기존 입력값 또는 기존 스냅샷 입력값이 제거된 경우. 주의할 점은 점진적 변환은 최소한 하나의 입력값을 가져야 한다는 것입니다.

또한, 스냅샷이 아닌 입력 데이터셋의 시작 트랜잭션은 이전 실행에 사용된 것과 일관성이 있어야 하는 요구사항이 있습니다.

동일한 변환에 의해 마지막으로 빌드된 출력값

다중 출력 점진적 변환의 경우, 각 출력값의 마지막으로 확정된 트랜잭션은 동일한 변환에서 생성되어야 합니다.

점진적 계산에 대한 요구사항 요약

변환은 점진적으로 실행될 수 있으며, 그 변환이 점진적 입력값에 파일이 추가되었거나, 파일이 삭제된 경우에만 가능합니다. 이때 파일 삭제는 Foundry Retention을 사용하여 allow_retention=True로 설정하여 삭제해야 합니다. 스냅샷 입력값은 이 검사에서 제외됩니다.