Amazon S3 Tables Catalog for Apache Iceberg를 사용하여 Amazon S3 Tables에 액세스

Amazon S3 Tables Catalog for Apache Iceberg 클라이언트 카탈로그를 사용하여 Apache Spark 같은 오픈 소스 쿼리 엔진에서 S3 Tables에 액세스할 수 있습니다. Amazon S3 Tables Catalog for Apache Iceberg는 AWS Labs에서 호스팅하는 오픈 소스 라이브러리입니다. 쿼리 엔진의 Apache Iceberg 작업(예: 테이블 검색, 메타데이터 업데이트, 테이블 추가 또는 제거)을 S3 Tables API 작업으로 변환하여 작동합니다.

Amazon S3 Tables Catalog for Apache Iceberg는 s3-tables-catalog-for-iceberg.jar이라는 Maven JAR로 배포됩니다. AWS Labs GitHub 리포지토리에서 클라이언트 카탈로그 JAR을 빌드하거나 Maven에서 다운로드할 수 있습니다. 테이블에 연결할 경우 Apache Iceberg에 대한 Spark 세션을 초기화할 때 클라이언트 카탈로그 JAR이 종속성으로 사용됩니다.

Apache Spark와 함께 Amazon S3 Tables Catalog for Apache Iceberg 사용

Spark 세션을 초기화할 때 Amazon S3 Tables Catalog for Apache Iceberg 클라이언트 카탈로그를 사용하여 오픈 소스 애플리케이션의 테이블에 연결할 수 있습니다. 세션 구성에서 Iceberg 및 Amazon S3 종속성을 지정하고 테이블 버킷을 메타데이터 웨어하우스로 사용하는 사용자 지정 카탈로그를 생성합니다.

Amazon S3 Tables Catalog for Apache Iceberg를 사용하여 Spark 세션을 초기화하려면

• 다음 명령을 사용하여 Spark를 초기화합니다. 명령을 사용하려면 Amazon S3 Tables Catalog for Apache Iceberg 버전 번호를 AWS Labs GitHub 리포지토리의 최신 버전으로 바꾸고 테이블 버킷 ARN을 실제 테이블 버킷 ARN으로 바꿉니다.

  spark-shell \
  --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1,software.amazon.s3tables:s3-tables-catalog-for-iceberg-runtime:0.1.4 \
  --conf spark.sql.catalog.s3tablesbucket=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.s3tablesbucket.catalog-impl=software.amazon.s3tables.iceberg.S3TablesCatalog \
  --conf spark.sql.catalog.s3tablesbucket.warehouse=arn:aws:s3tables:us-east-1:111122223333:bucket/amzn-s3-demo-table-bucket \
  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

Spark SQL을 사용하여 S3 테이블 쿼리

Spark를 사용하면 S3 테이블에서 DQL, DML 및 DDL 작업을 실행할 수 있습니다. 테이블을 쿼리할 때 이 패턴에 따르는 세션 카탈로그 이름을 포함하여 정규화된 테이블 이름을 사용합니다.

CatalogName.NamespaceName.TableName

다음 예시 쿼리는 S3 테이블과 상호 작용할 수 있는 몇 가지 방법을 보여줍니다. 쿼리 엔진에서 이러한 예시 쿼리를 사용하려면 사용자 입력 자리 표시자 값을 실제 값으로 바꾸세요.

Spark를 사용하여 테이블을 쿼리하는 방법

• 네임스페이스 생성

  spark.sql(" CREATE NAMESPACE IF NOT EXISTS s3tablesbucket.my_namespace")

• 테이블 생성

  spark.sql(" CREATE TABLE IF NOT EXISTS s3tablesbucket.my_namespace.`my_table` 
  ( id INT, name STRING, value INT ) USING iceberg ")

• 테이블 쿼리

  spark.sql(" SELECT * FROM s3tablesbucket.my_namespace.`my_table` ").show()

• 테이블에 데이터 삽입

  spark.sql(
  """
      INSERT INTO s3tablesbucket.my_namespace.my_table 
      VALUES 
          (1, 'ABC', 100), 
          (2, 'XYZ', 200)
  """)

• 테이블에 기존 데이터 파일 로드

  1. Spark로 데이터를 읽습니다.

     val data_file_location = "Path such as S3 URI to data file"
     val data_file = spark.read.parquet(data_file_location)

  2. Iceberg 테이블에 데이터를 기록합니다.

     data_file.writeTo("s3tablesbucket.my_namespace.my_table").using("Iceberg").tableProperty ("format-version", "2").createOrReplace()