데이터셋 열에 Cipher 연산 적용하기

Cipher는 Pipeline Builder, Contour와 Python 변환에서 데이터셋 열을 암호화, 복호화하거나 해시할 수 있습니다. Pipeline Builder와 Code Repositories의 Preview에서는 실제 연산 결과 대신 임시 표시값이 보입니다. 실제 암호화 결과는 빌드할 때 생성하므로 Preview의 표시값을 암호문으로 사용하면 안 됩니다.

Pipeline Builder

Data Manager License 또는 Admin License의 접근 권한과 수행할 연산의 허용이 필요합니다.

Cipher encrypt 변환에서 Expression으로 암호화할 열을 선택하고 암호화가 허용된 라이선스와 출력 열 이름을 지정합니다.

Cipher decrypt는 이미 Cipher로 암호화한 열을 Expression으로 선택합니다. 복호화 라이선스는 그 값을 암호화한 채널과 같은 채널에 속해야 합니다. 출력 열 이름도 지정합니다.

Cipher hash는 해시할 열과 해시 채널의 암호화 및 해시 권한이 있는 라이선스를 선택하고 출력 열 이름을 정합니다.

라이선스에서 파이프라인 생성

암호화 권한이 있는 Data Manager 또는 Admin License를 열고 오른쪽 위 Create Pipeline을 선택할 수도 있습니다. 입력 데이터셋, 저장할 폴더와 암호화할 열을 선택하면 해당 열을 암호화하는 파이프라인을 생성합니다.

이 생성 흐름에서는 String 열만 선택할 수 있습니다. 다른 형식은 먼저 String으로 변환합니다.

Contour

Data Manager 또는 Admin License가 필요합니다. Contour 도구 모음의 검색 모드에서 Cipher board를 추가합니다. Table board나 table panel에서 결과를 확인할 수 있습니다.

Cipher board를 사용하는 분석 경로는 데이터셋으로 저장할 수 없습니다.

열 선택 순서는 연산에 영향을 주지 않습니다. 선택한 열의 값은 바뀌지만 열 이름은 유지됩니다.

Python 저장소 준비

conda_recipe/meta.yml의 requirements.run에 bellaso-python-lib를 추가하거나 저장소의 Libraries 패널에서 추가합니다. 코드 변환에는 암호 키 접근을 허용하는 Admin License가 필요합니다. 수집 시점에 가리려면 외부 연결 변환과 이 라이브러리를 함께 사용할 수 있습니다.

버전 0.75.0부터 기존 메서드는 새로운 구현으로 대체되며 사용 중단 경고를 표시합니다. 이전 메서드는 아직 사용할 수 있지만 이후 릴리스에서 제거될 예정이므로 다음 대응에 맞춰 이전합니다.

다음 예제의 /Training/Cipher/... 경로와 열 이름은 가상입니다. 실제 데이터셋 및 연산에 맞는 라이선스 경로로 바꿉니다. 라이선스 입력에는 파일시스템 경로 대신 RID도 사용할 수 있습니다. 각 예제는 별도의 변환입니다.

열 암호화

EncrypterInput에 암호화가 허용된 라이선스를 연결합니다.

from transforms.api import Input, Output, transform
from pyspark.sql import functions as F
from bellaso_python_lib.encryption.encrypter_input import EncrypterInput


@transform(
    encrypter=EncrypterInput("/Training/Cipher/licenses/encrypt-admin"),
    output=Output("/Training/Cipher/customer_encrypted"),
    input_dataset=Input("/Training/Cipher/customer_raw"),
)
def encrypt_customer_names(input_dataset, output, encrypter):
    cipher = encrypter.dataframe()
    result = input_dataset.dataframe().withColumn(
        "customer_name", cipher.encrypt_column(F.col("customer_name"))
    )
    output.write_dataframe(result)

열 복호화

DecrypterInput에 해당 암호문의 채널에 속한 복호화 라이선스를 연결합니다. 결과는 평문이므로 출력 데이터셋의 접근과 반출 권한을 관리해야 합니다.

from transforms.api import Input, Output, transform
from pyspark.sql import functions as F
from bellaso_python_lib.decryption.decrypter_input import DecrypterInput


@transform(
    decrypter=DecrypterInput("/Training/Cipher/licenses/decrypt-admin"),
    output=Output("/Training/Cipher/customer_decrypted"),
    input_dataset=Input("/Training/Cipher/customer_encrypted"),
)
def decrypt_customer_names(input_dataset, output, decrypter):
    cipher = decrypter.dataframe()
    result = input_dataset.dataframe().withColumn(
        "customer_name", cipher.decrypt_column(F.col("customer_name"))
    )
    output.write_dataframe(result)

열 해시

HasherInput에는 해시 채널의 허용된 라이선스를 사용합니다.

from transforms.api import Input, Output, transform
from pyspark.sql import functions as F
from bellaso_python_lib.encryption.hasher_input import HasherInput


@transform(
    hasher=HasherInput("/Training/Cipher/licenses/hash-admin"),
    output=Output("/Training/Cipher/customer_hashed"),
    input_dataset=Input("/Training/Cipher/customer_raw"),
)
def hash_customer_names(input_dataset, output, hasher):
    cipher = hasher.dataframe()
    result = input_dataset.dataframe().withColumn(
        "customer_name", cipher.hash_column(F.col("customer_name"))
    )
    output.write_dataframe(result)

증분 변환

증분 변환에서 사용하는 모든 encrypter, decrypter와 hasher 입력은 snapshot_inputs에 지정해야 합니다. 다음 예제는 encrypter를 스냅샷 입력으로 지정하고 암호화된 열을 추가합니다.

from transforms.api import Input, Output, incremental, transform
from pyspark.sql import functions as F
from bellaso_python_lib.encryption.encrypter_input import EncrypterInput


@incremental(snapshot_inputs=["encrypter"])
@transform(
    encrypter=EncrypterInput("/Training/Cipher/licenses/encrypt-admin"),
    output=Output("/Training/Cipher/customer_incremental_encrypted"),
    input_dataset=Input("/Training/Cipher/customer_events"),
)
def encrypt_incremental_names(input_dataset, output, encrypter):
    cipher = encrypter.dataframe()
    result = input_dataset.dataframe().withColumn(
        "encrypted_name", cipher.encrypt_column(F.col("customer_name"))
    )
    output.write_dataframe(result)

이 예제는 원래 customer_name 열을 유지하고 새 열을 추가합니다. 평문을 공유 출력에 남기면 안 되는 업무에서는 출력 열 구성을 별도로 정해야 합니다.

Python 문자열 단위 연산

문자열 단위 메서드는 Spark Column 표현식이 아니라 실제 Python 문자열을 받습니다.

pyspark.sql.functions.transform()의 람다 인수는 실제 문자열이 아니라 Column입니다. Python 문자열 메서드에 그 인수를 직접 전달하지 않습니다.

문자열 배열 처리

다음 예제는 string_array를 원소별 행으로 펼쳐 encrypt_column()을 적용한 뒤, 원래 위치 순서로 encrypted_array를 만듭니다. NULL 배열, 빈 배열과 NULL 원소를 보존하며, 같은 입력 행이 여러 번 있어도 각각 유지합니다. 암호화할 값이 없는 배열에는 암호 연산을 호출하지 않습니다.

from transforms.api import Input, Output, transform
from pyspark.sql import functions as F
from bellaso_python_lib.encryption.encrypter_input import EncrypterInput


@transform(
    encrypter=EncrypterInput("/Training/Cipher/licenses/encrypt-admin"),
    output=Output("/Training/Cipher/customer_arrays_encrypted"),
    input_dataset=Input("/Training/Cipher/customer_arrays"),
)
def encrypt_name_arrays(input_dataset, output, encrypter):
    source = input_dataset.dataframe()
    cipher = encrypter.dataframe()
    output_names = list(source.columns)
    if "encrypted_array" not in output_names:
        output_names.append("encrypted_array")

    original = F.struct(*[
        F.col("`" + name.replace("`", "``") + "`").alias(name)
        for name in source.columns
    ])
    base = source.select(original.alias("_original"))
    values = F.col("_original").getField("string_array")
    has_values = F.coalesce(
        F.exists(values, lambda value: value.isNotNull()), F.lit(False)
    )

    expanded = (
        base.filter(has_values)
        .withColumn("_row_id", F.monotonically_increasing_id())
        .select("_row_id", "_original", F.posexplode(values).alias("_position", "_value"))
        .filter(F.col("_value").isNotNull())
    )
    encrypted = expanded.withColumn("_encrypted", cipher.encrypt_column(F.col("_value")))
    grouped = encrypted.groupBy("_row_id").agg(
        F.first("_original").alias("_original"),
        F.collect_list(F.struct(
            F.col("_position").alias("position"),
            F.col("_encrypted").alias("value"),
        )).alias("_parts"),
    )
    null_parts = F.filter(
        F.transform(values, lambda value, position: F.when(
            value.isNull(),
            F.struct(position.alias("position"), F.lit(None).cast("string").alias("value")),
        )),
        lambda part: part.isNotNull(),
    )
    encrypted_array = F.transform(
        F.sort_array(F.concat(F.col("_parts"), null_parts)),
        lambda part: part.getField("value"),
    )

    def select_output(frame, result_array):
        return frame.select(*[
            result_array.alias(name) if name == "encrypted_array"
            else F.col("_original").getField(name).alias(name)
            for name in output_names
        ])

    with_values = select_output(grouped, encrypted_array)
    without_values = select_output(base.filter(~has_values), values)
    output.write_dataframe(with_values.unionByName(without_values))

Cipher 객체는 드라이버에서 열 표현식을 구성할 때 사용합니다. 내부 행 ID는 원소를 다시 모으는 분기 안에서만 사용하며, 출력하거나 별도 계산과 조인하는 영구 키로 재사용하지 않습니다.

배열 재조립에는 행 그룹화와 셔플 비용이 발생합니다. 집계 과정에서 일부 입력 열이 nullable로 추론될 수 있으므로, 고정된 출력 스키마를 사용하는 경우에는 결과 스키마도 확인합니다. 암호 연산이나 Spark 실행의 오류를 평문이나 성공값으로 바꾸지 않습니다.

이 예제도 입력 배열을 유지합니다. 최소화가 필요한 출력에서는 원래 배열을 제외하도록 출력 스키마를 정합니다. 이미지 영역에 대한 obfuscate_image() 사용은 Visual Obfuscation에서 다룹니다.