Amazon S3 연결과 가상 테이블

Amazon S3 커넥터는 외부 S3 버킷과 Foundry 사이에서 데이터를 수집하거나 내보냅니다. Foundry 데이터셋 자체를 S3 버킷처럼 제공하는 S3 호환 API와는 연결 대상이 다릅니다.

원천 탐색, 대량 가져오기, 지원 파일 형식의 증분 수집, 미디어 집합, 가상 테이블과 파일 내보내기를 일반 제공(GA)합니다. Data Connection의 +New Source에서는 S3를 선택합니다.

기본 연결과 네트워크

Foundry의 외부 통신 정책에는 예를 들어 bucket-name.s3.us-east-1.amazonaws.com:443을 허용합니다. Data Connection은 입력한 연결 정보에 맞는 정책을 제안합니다. 버킷 정책에서도 Foundry의 관련 IP와 버킷 접근을 허용해야 합니다. 출발 IP는 Control Panel의 Network egress에서 확인합니다.

Foundry와 버킷이 같은 AWS 리전에 있으면 S3 버킷 및 엔드포인트 정책의 추가 조건도 적용됩니다. Foundry 측 허용 목록만 구성하고 버킷 측 통제를 생략하지 않습니다. 가상 테이블에는 별도의 네트워크 정책 유형 제한도 적용됩니다.

인증과 기본 권한

액세스 키 방식은 대상 AWS 계정의 IAM 사용자에 버킷 접근 권한을 부여하고 Access Key ID와 Secret을 입력합니다. OIDC는 소스 화면에 표시되는 원천 시스템 설정 절차에 따라 구성합니다.

다음은 가상 버킷 example-bucket의 읽기용 IAM 정책 Statement 예시입니다. 실제 버킷 이름과 필요한 범위로 바꿉니다.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Action": ["s3:ListBucket"],
      "Resource": ["arn:aws:s3:::example-bucket"],
      "Effect": "Allow"
    },
    {
      "Action": ["s3:GetObject"],
      "Resource": ["arn:aws:s3:::example-bucket/*"],
      "Effect": "Allow"
    }
  ]
}

내보내기의 Statement는 다음과 같습니다.

{
  "Action": ["s3:PutObject"],
  "Resource": ["arn:aws:s3:::example-bucket/*"],
  "Effect": "Allow"
}

연결 동작 설정

Path-style URL은 https://s3.region-code.amazonaws.com/bucket-name/key-name, virtual-hosted-style은 https://bucket-name.s3.region-code.amazonaws.com/key-name 형식입니다.

에이전트의 프록시 설정

프록시는 소스 구성에 직접 지정할 수 있습니다. 소스 설정이 없으면 에이전트 시스템 속성의 JVM 인자를 사용할 수 있으며 -Dhttps.proxyHost=example.proxy.com은 형식 예시입니다.

STS와 클라우드 신원

STS 역할 사용 시 roleArn, roleSessionName은 필수입니다. roleSessionDuration은 선택 항목이며 기본값은 3,600초이고 externalId도 선택 항목입니다.

클라우드 신원은 Control Panel에서 플랫폼 인스턴스 수준으로 관리합니다. Cloud identity 인증을 선택하면 역할 ARN이 표시됩니다.

대상 AWS 계정에 IAM 역할을 만들고 버킷 접근 권한을 부여합니다. 이 역할을 S3 소스의 STS role에 지정하고, 대상 역할의 신뢰 정책에서 Foundry 클라우드 신원 역할이 역할을 맡도록 허용합니다.

S3 가상 테이블

Foundry 워커가 필요하며 에이전트 워커는 지원하지 않습니다. 필요한 양방향 연결과 허용 목록을 구성하고 Code Repositories에서 사용하려면 관련 소스의 코드 사용 설정도 수행합니다.

Agent proxy, Private link 또는 Bucket endpoint egress 정책을 사용하는 소스에서는 가상 테이블을 지원하지 않습니다. Agent proxy는 네트워크 연결 경로이며 에이전트 워커에서 작업을 실행하는 설정과는 별개입니다. 같은 리전 버킷을 연결할 때에도 일반 연결용 Bucket endpoint 정책을 가상 테이블에 그대로 적용할 수 없습니다. 정책별 지원 범위는 외부 연결 정책과 S3 네트워크 구성을 확인합니다.

버킷 이름에 .이 있으면 path-style을 활성화하고 이에 맞는 외부 통신 정책을 설정해야 합니다.

Delta

증분 읽기에는 원천 Delta 테이블의 Change Data Feed가 활성화되어 있어야 합니다. Python Transforms에서 current, added 읽기 모드와 _change_type, _commit_version, _commit_timestamp 열을 제공합니다.

Iceberg 카탈로그

같은 소스에서 등록한 모든 Iceberg 테이블은 하나의 카탈로그를 사용합니다.

Iceberg 증분 읽기는 추가 전용이며 current, added를 지원합니다. Parquet 가상 테이블의 스키마 추론에는 최대 100개 파일을 사용합니다.

파일 내보내기

소스의 내보내기를 활성화하고 새 내보내기를 생성합니다. 선택 항목 Path Prefix를 지정하면 최종 경로는 s3://<bucket-name>/<path-in-source-config>/<path-prefix>/<exported-file>이 됩니다. Canned ACL은 업로드한 객체에 붙일 AWS의 미리 정의된 ACL이며 선택 사항입니다.

Iceberg BYOB 저장소의 권한

Foundry Iceberg 테이블의 자체 버킷 저장소에는 다음 S3 권한이 필요합니다. bucket-name은 예시입니다.

{
  "Action": ["s3:DeleteObject", "s3:GetObject", "s3:ListBucket", "s3:PutObject"],
  "Resource": ["arn:aws:s3:::bucket-name", "arn:aws:s3:::bucket-name/*"],
  "Effect": "Allow"
}

암호화에 사용하는 KMS 키에는 다음 권한이 필요합니다. 리전, 계정과 키 ID를 실제 값으로 바꿉니다.

{
  "Action": ["kms:Decrypt", "kms:Encrypt", "kms:GenerateDataKey"],
  "Resource": ["arn:aws:kms:region:account-id:key/key-id"],
  "Effect": "Allow"
}

인증 정보 발급에는 sts:GetFederationToken 권한도 필요합니다.

기존 Export task

신규 내보내기는 현재 파일 내보내기 기능을 사용합니다. 이전 Export task에는 소스 유형 s3-direct와 작업 유형 export-s3-task가 필요합니다.

type: export-s3-task
directoryPath: /some/directory

경로와 값은 예시입니다. 다음 구성은 증분 처리와 파일별 재시도를 사용합니다.

type: export-s3-task
directoryPath: /exports/data
incrementalType: incremental
retriesPerFile: 2

여러 객체를 병렬로 내보내며 객체 접근 제어 값을 지정하는 구성입니다.

type: export-s3-task
directoryPath: /exports/data
useMultipart: false
threads: 4
bucketPolicy: BucketOwnerFullControl
setBucketPolicy: true

Spark 접두사는 다음처럼 제거할 수 있습니다.

type: export-s3-task
directoryPath: /exports/data
rewritePaths:
  "^spark/(.*)": "$1"

코드에서 데이터셋 파일 내보내기

다음 외부 변환은 데이터셋 하나의 파일을 S3에 업로드하고 데이터셋 RID, 시각, 파일 수와 목적지 경로를 로그 데이터셋에 기록합니다. boto3를 추가하고 S3 소스를 가져온 뒤 리전, 버킷, endpoint, 대상 경로와 리소스 RID를 실제 값으로 바꿉니다.

세션 인증 정보에는 access key, secret key와 session token이 함께 필요합니다. 예제는 현재 세션 인증 정보를 한 번 가져오므로 대략 1시간의 수명보다 오래 걸리는 내보내기에서는 갱신 가능한 인증 정보 구성을 사용해야 합니다. 실행하면 외부 버킷에 파일을 씁니다.

파일 목록은 Spark 파티션별로 처리합니다. 각 워커는 자기 Boto3 세션과 클라이언트를 만들며, 드라이버가 만든 네트워크 클라이언트를 워커로 직렬화하여 전달하지 않습니다.

# Export one dataset's files and record the completed export in a log dataset.

from pyspark.sql import types as T
from transforms.api import Input, Output, transform
from transforms.external.systems import external_systems, Source
from datetime import datetime
import boto3
import logging

log = logging.getLogger(__name__)

S3_BUCKET_REGION = "<region>"
S3_BUCKET_NAME = "<name>"
S3_ENDPOINT_URL = "<endpoint_url>"
S3_DEST_PATH = "<s3_dest_path>"

S3_SOURCE_RID = "<s3_source_rid>"

LOG_OUTPUT_PATH = "<log_output_path>"

DATASET_RID = "<dataset_rid>"

@external_systems(
    s3_source=Source(S3_SOURCE_RID)
)
@transform(
    export_log=Output(f"{LOG_OUTPUT_PATH}/export_{DATASET_RID}_log"),
    input_dataset=Input(f"{DATASET_RID}")
)
def compute_function(ctx, input_dataset, s3_source, export_log):
    # 1. Retrieve credentials for the S3 bucket
    refreshable_creds = s3_source.get_session_credentials()
    creds = refreshable_creds.get()

    # 2. Construct S3 key for the current dataset
    s3_dataset_key = f"{S3_DEST_PATH}/{DATASET_RID}/"

    # Pass credential values to workers, not a live network client.
    client_kwargs = {
        "aws_access_key_id": creds.access_key_id,
        "aws_secret_access_key": creds.secret_access_key,
        "aws_session_token": creds.session_token,
        "region_name": S3_BUCKET_REGION,
        "endpoint_url": S3_ENDPOINT_URL,
    }

    def write_partition(rows):
        s3_client = boto3.session.Session().client("s3", **client_kwargs)
        try:
            for row in rows:
                try:
                    with input_dataset.filesystem().open(row.path, "rb") as in_f:
                        s3_key = s3_dataset_key + row.path
                        s3_client.upload_fileobj(in_f, S3_BUCKET_NAME, s3_key)
                except Exception as e:
                    log.error(f"Exception when uploading {DATASET_RID} to S3: {e}")
                    raise
        finally:
            s3_client.close()

    # 4. Write dataset files to S3
    result = input_dataset.filesystem().files()
    result.foreachPartition(write_partition)

    # 5. Define output export log schema
    schema = T.StructType([
        T.StructField("dataset_rid", T.StringType(), False),
        T.StructField("export_timestamp", T.TimestampType(), False),
        T.StructField("files_uploaded", T.IntegerType(), False),
        T.StructField("s3_dest_path", T.StringType(), False)
    ])
    file_count = result.count()
    output_df = ctx.spark_session.createDataFrame([(DATASET_RID, datetime.now(), file_count, s3_dataset_key)], schema)
    export_log.write_dataframe(output_df)