커스텀 카탈로그

커스텀 카탈로그 (Custom Catalog)

아이스버그 테이블은 hdfs 경로나 하이브 테이블에서 읽을 수 있어요. 또한 하이브 대신 커스텀 메타스토어를 사용할 수도 있어요. 이 문서에서는 커스텀 메타스토어를 만들기 위한 다섯 가지 단계인 커스텀 TableOperations, 커스텀 Catalog, 커스텀 FileIO, 커스텀 LocationProvider, 커스텀 IcebergSource를 각각의 코드 예시와 함께 알려드릴게요.

출처: 문서

본문

아이스버그 테이블은 hdfs 경로나 하이브 테이블에서 읽을 수 있어요. 또한 하이브 대신 커스텀 메타스토어를 사용할 수도 있어요. 이를 위한 단계는 다음과 같아요.

• 커스텀 TableOperations (Custom TableOperations)
• 커스텀 Catalog (Custom Catalog)
• 커스텀 FileIO (Custom FileIO)
• 커스텀 LocationProvider (Custom LocationProvider)
• 커스텀 IcebergSource (Custom IcebergSource)

참고: 암호화된 테이블을 작업하려면 커스텀 카탈로그가 여러 보안 요구사항을 충족해야 해요.

커스텀 테이블 연산 구현 (Custom table operations implementation)

BaseMetastoreTableOperations를 확장해서 메타데이터를 읽고 쓰는 방법을 제공해요.

예시:

class CustomTableOperations extends BaseMetastoreTableOperations {
  private String dbName;
  private String tableName;
  private Configuration conf;
  private FileIO fileIO;

  protected CustomTableOperations(Configuration conf, String dbName, String tableName) {
    this.conf = conf;
    this.dbName = dbName;
    this.tableName = tableName;
  }

  // The doRefresh method should provide implementation on how to get the metadata location
  @Override
  public void doRefresh() {

    // Example custom service which returns the metadata location given a dbName and tableName
    String metadataLocation = CustomService.getMetadataForTable(conf, dbName, tableName);

    // When updating from a metadata file location, call the helper method
    refreshFromMetadataLocation(metadataLocation);

  }

  // The doCommit method should provide implementation on how to update with metadata location atomically
  @Override
  public void doCommit(TableMetadata base, TableMetadata metadata) {
    String oldMetadataLocation = base.location();

    // Write new metadata using helper method
    String newMetadataLocation = writeNewMetadata(metadata, currentVersion() + 1);

    // Example custom service which updates the metadata location for the given db and table atomically
    CustomService.updateMetadataLocation(dbName, tableName, oldMetadataLocation, newMetadataLocation);

  }

  // The io method provides a FileIO which is used to read and write the table metadata files
  @Override
  public FileIO io() {
    if (fileIO == null) {
      fileIO = new HadoopFileIO(conf);
    }
    return fileIO;
  }
}

TableOperations 인스턴스는 보통 Catalog.newTableOps(TableIdentifier)를 호출해서 얻어요. 커스텀 카탈로그 구현과 로드에 대한 내용은 다음 섹션을 참고해주세요.

커스텀 카탈로그 구현 (Custom catalog implementation)

BaseMetastoreCatalog를 확장해서 기본 warehouse 위치를 제공하고 CustomTableOperations를 인스턴스화해요.

예시:

public class CustomCatalog extends BaseMetastoreCatalog {

  private Configuration configuration;

  // must have a no-arg constructor to be dynamically loaded
  // initialize(String name, Map<String, String> properties) will be called to complete initialization
  public CustomCatalog() {
  }

  public CustomCatalog(Configuration configuration) {
    this.configuration = configuration;
  }

  @Override
  protected TableOperations newTableOps(TableIdentifier tableIdentifier) {
    String dbName = tableIdentifier.namespace().level(0);
    String tableName = tableIdentifier.name();
    // instantiate the CustomTableOperations
    return new CustomTableOperations(configuration, dbName, tableName);
  }

  @Override
  protected String defaultWarehouseLocation(TableIdentifier tableIdentifier) {

    // Can choose to use any other configuration name
    String tableLocation = configuration.get("custom.iceberg.warehouse.location");

    // Can be an s3 or hdfs path
    if (tableLocation == null) {
      throw new RuntimeException("custom.iceberg.warehouse.location configuration not set!");
    }

    return String.format(
            "%s/%s.db/%s", tableLocation,
            tableIdentifier.namespace().levels()[0],
            tableIdentifier.name());
  }

  @Override
  public boolean dropTable(TableIdentifier identifier, boolean purge) {
    // Example service to delete table
    CustomService.deleteTable(identifier.namespace().level(0), identifier.name());
  }

  @Override
  public void renameTable(TableIdentifier from, TableIdentifier to) {
    Preconditions.checkArgument(from.namespace().level(0).equals(to.namespace().level(0)),
            "Cannot move table between databases");
    // Example service to rename table
    CustomService.renameTable(from.namespace().level(0), from.name(), to.name());
  }

  // implement this method to read catalog name and properties during initialization
  public void initialize(String name, Map<String, String> properties) {
  }
}

Catalog 구현은 대부분의 컴퓨트 엔진에서 동적으로 로드될 수 있어요. 스파크와 플링크의 경우 catalog-impl 카탈로그 속성을 지정해서 로드할 수 있어요. 자세한 내용은 구성(Configuration) 섹션을 읽어주세요. MapReduce의 경우 org.apache.iceberg.mr.CatalogLoader를 구현하고 Hadoop 속성 iceberg.mr.catalog.loader.class를 설정해서 로드해요. 카탈로그가 특정 환경 속성에 접근하기 위해 Hadoop 구성을 읽어야 한다면, 카탈로그가 org.apache.hadoop.conf.Configurable을 구현하게 해주세요.

커스텀 파일 입출력 구현 (Custom file IO implementation)

FileIO를 확장하고 데이터 파일을 읽고 쓰는 구현을 제공해요.

예시:

public class CustomFileIO implements FileIO {

  // must have a no-arg constructor to be dynamically loaded
  // initialize(Map<String, String> properties) will be called to complete initialization
  public CustomFileIO() {
  }

  @Override
  public InputFile newInputFile(String s) {
    // you also need to implement the InputFile interface for a custom input file
    return new CustomInputFile(s);
  }

  @Override
  public OutputFile newOutputFile(String s) {
    // you also need to implement the OutputFile interface for a custom output file
    return new CustomOutputFile(s);
  }

  @Override
  public void deleteFile(String path) {
    Path toDelete = new Path(path);
    FileSystem fs = Util.getFs(toDelete);
    try {
        fs.delete(toDelete, false /* not recursive */);
    } catch (IOException e) {
        throw new RuntimeIOException(e, "Failed to delete file: %s", path);
    }
  }

  // implement this method to read catalog properties during initialization
  public void initialize(Map<String, String> properties) {
  }
}

이미 자신만의 카탈로그를 구현하고 있다면 TableOperations.io()를 구현해서 자신만의 커스텀 FileIO를 사용할 수 있어요. 또한 커스텀 FileIO 구현은 io-impl 카탈로그 속성을 지정해서 HadoopCatalog와 HiveCatalog에서도 동적으로 로드될 수 있어요. 자세한 내용은 구성(Configuration) 섹션을 읽어주세요. FileIO가 특정 환경 속성에 접근하기 위해 Hadoop 구성을 읽어야 한다면, FileIO가 org.apache.hadoop.conf.Configurable을 구현하게 해주세요.

커스텀 위치 제공자 구현 (Custom location provider implementation)

LocationProvider를 확장하고 데이터를 쓸 파일 경로를 결정하는 구현을 제공해요.

예시:

public class CustomLocationProvider implements LocationProvider {

  private String tableLocation;

  // must have a 2-arg constructor like this, or a no-arg constructor
  public CustomLocationProvider(String tableLocation, Map<String, String> properties) {
    this.tableLocation = tableLocation;
  }

  @Override
  public String newDataLocation(String filename) {
    // can use any custom method to generate a file path given a file name
    return String.format("%s/%s/%s", tableLocation, UUID.randomUUID().toString(), filename);
  }

  @Override
  public String newDataLocation(PartitionSpec spec, StructLike partitionData, String filename) {
    // can use any custom method to generate a file path given a partition info and file name
    return newDataLocation(filename);
  }
}

이미 자신만의 카탈로그를 구현하고 있다면 TableOperations.locationProvider()를 재정의해서 자신만의 기본 LocationProvider를 사용할 수 있어요. 특정 테이블에 다른 커스텀 위치 제공자를 사용하려면 테이블 생성 시 테이블 속성 write.location-provider.impl로 구현을 지정해요.

예시:

CREATE TABLE hive.default.my_table (
  id bigint,
  data string,
  category string)
USING iceberg
OPTIONS (
  'write.location-provider.impl'='com.my.CustomLocationProvider'
)
PARTITIONED BY (category);

커스텀 IcebergSource

IcebergSource를 확장하고 CustomCatalog에서 읽는 구현을 제공해요.

예시:

public class CustomIcebergSource extends IcebergSource {

  @Override
  protected Table findTable(DataSourceOptions options, Configuration conf) {
    Optional<String> path = options.get("path");
    Preconditions.checkArgument(path.isPresent(), "Cannot open table: path is not set");

    // Read table from CustomCatalog
    CustomCatalog catalog = new CustomCatalog(conf);
    TableIdentifier tableIdentifier = TableIdentifier.parse(path.get());
    return catalog.loadTable(tableIdentifier);
  }
}

META-INF/services/org.apache.spark.sql.sources.DataSourceRegister을 완전한 클래스 이름으로 업데이트해서 CustomIcebergSource를 등록해요.

더 알아보기 (Learn more)

• 카탈로그 속성 (Catalog properties)
• FileIO
• 아이스버그 자바 API (Java API)