CDE 에서 돌아가는 PySpark Job 에서 사내 KMS 로 암호화된 DB 컬럼을 복호화해야 한다. 추가 패키지 설치가 불가능하고 pandas 도 쓸 수 없다. 원래 받은 샘플은 jpype + requests 기반이었으나 PySpark 환경에서는 그대로 쓸 수 없다.
| 항목 | 문제 |
|---|---|
jpype |
PySpark 에 없는 별도 패키지. PySpark 는 py4j 를 쓴다 |
requests |
표준 라이브러리가 아니다. urllib.request 로 대체 |
JFunction |
JPype 에 없는 API |
| 생성자 시그니처 | 샘플은 KmsClient(String) 이지만 실제는 KmsClient(InputStream, String) |
public KmsClient(java.io.InputStream certStream, java.lang.String kmsUri) throws KMSUtilException
KmsCipher getCipher(String keyGroup, String systemId, String keyId, Function<String,String> apiCaller)
// KmsCipher
String encryptStringToBase64(String data, KmsCipher.IvType ivType)
String decryptBase64ToString(String encrypted)
getCipher 가 Function<String,String> 콜백을 받으므로 py4j 콜백으로 구현해야 한다. KmsClient 인스턴스는 한 번 만들어 Job 안에서 재사용한다.
from urllib.request import Request, urlopen
from pyspark.sql import SparkSession
DECRYPT_CERT_PATH = "/app/mount/certs/kms.p12"
DECRYPT_BASE_URL = "https://kms.example.com"
DECRYPT_TENANT_CD = "COMMON"
DECRYPT_SERVICE_CD = "PT"
DECRYPT_KEY_ID = "KMS-COMMON"
spark = SparkSession.builder.getOrCreate()
spark._sc._gateway.start_callback_server() # py4j 콜백 서버 기동
jvm = spark._jvm
KmsClient = jvm.com.example.kms.securityutil.KmsClient
cert_stream = jvm.java.io.FileInputStream(DECRYPT_CERT_PATH)
kms_client = KmsClient(cert_stream, DECRYPT_BASE_URL)
class ApiCaller(object):
"""Java 의 Function<String,String> 을 파이썬으로 구현"""
def apply(self, uri):
nonce = str(kms_client.getNonce())
req = Request(str(uri), headers={"Accept": "application/json", "nonce": nonce})
with urlopen(req) as resp:
return resp.read().decode("utf-8")
class Java:
implements = ["java.util.function.Function"]
cipher = kms_client.getCipher(DECRYPT_TENANT_CD, DECRYPT_SERVICE_CD, DECRYPT_KEY_ID, ApiCaller())
class Java: implements = [...] 선언이 있어야 py4j 가 파이썬 객체를 Java 인터페이스 구현체로 넘긴다.
cipher.decryptBase64ToString() 은 로컬 연산이라 네트워크 비용이 없지만, 객체가 드라이버 JVM 에 있어 UDF 로 만들면 executor 에서 직렬화되지 않는다. 데이터 규모에 따라 둘 중 하나를 고른다.
df.collect() 후 파이썬에서 복호화하고 다시 DataFrame 으로 만든다. 구현이 단순하다.mapPartitions 안에서 초기화한다. 파티션마다 인증서를 읽고 KMS 를 호출하므로 파티션 수와 KMS 호출 한도를 함께 고려한다.rows = df.select("id", "enc_col").collect()
decoded = [(r["id"], cipher.decryptBase64ToString(r["enc_col"])) for r in rows]
result = spark.createDataFrame(decoded, ["id", "dec_col"])
Job 이 끝나면 콜백 서버가 남지 않도록 spark._sc._gateway.shutdown_callback_server() 를 호출한다. 인증서 경로와 KMS URL 은 코드에 박지 말고 Job 파라미터나 Resource 로 넘긴다.