Usar tabelas do Apache Iceberg com o metastore do Dataproc

Nesta página, explicamos como usar tabelas do Apache Iceberg com um serviço de Metastore do Dataproc anexado a um cluster do Dataproc. Apache Iceberg é um formato de tabela aberto para grandes conjuntos de dados analíticos.

Compatibilidades

As tabelas Iceberg oferecem suporte aos recursos a seguir.

Observação: não é possível escrever uma tabela de Iceberg no Hive. Em vez disso, crie uma tabela de Iceberg, insira dados no Spark e leia-os no Hive.

Drivers	Seleção	Inserir	Criar tabela
Spark	✓	✓	✓
Hive	✓		✓
Presto	✓	✓	✓

Antes de começar

• Crie um serviço Metastore do Dataproc.
• Anexe o Metastore do Dataproc a um cluster do Dataproc.

Usar a tabela Iceberg com o Spark

O exemplo a seguir mostra que é necessário usar tabelas do Iceberg com o Spark.

As tabelas de Iceberg são compatíveis com operações de leitura e gravação. Para mais informações, consulte Apache Iceberg - Spark.

Configurações do Spark

Primeiro, inicie o shell do Spark e use um bucket do Cloud Storage para armazenar dados. Para incluir o Iceberg na instalação do Spark, adicione o arquivo JAR de tempo de execução do Iceberg Spark à pasta JARs do Spark. Para fazer o download do arquivo JAR, consulte Downloads do Apache Iceberg. O comando a seguir inicia o shell do Spark com suporte para Apache Iceberg:

$ spark-shell --conf spark.sql.warehouse.dir=gs://BUCKET_NAME/spark-warehouse --jars /path/to/iceberg-spark-runtime.jar

Usar o Hive Catalog para criar tabelas de Iceberg

1. Defina as configurações do Hive Catalog para criar tabelas de Iceberg no Spark Spark:

   import org.apache.iceberg.hive.HiveCatalog
   import org.apache.iceberg.catalog._
   import org.apache.iceberg.Schema
   import org.apache.iceberg.types.Types._
   import org.apache.iceberg.PartitionSpec
   import org.apache.iceberg.spark.SparkSchemaUtil
   import org.apache.spark.sql._
   import java.util.HashMap
   

2. Crie uma tabela para inserir e atualizar dados. Veja um exemplo.

   1. Crie uma tabela chamada example no banco de dados default:

      val catalog = new HiveCatalog();
      catalog.setConf(spark.sparkContext.hadoopConfiguration);
      catalog.initialize("hive", new HashMap[String,String]());
      
      val name = TableIdentifier.of("default","example");
      

   2. Inserir dados de amostra:

      val df1 = Seq((1,"Vincent","Computer Science"),(2,"Dan", "Economics"),(3,"Bob", "Politics"),(4,"Cindy", "UX Design")).toDF("id","name","major");
      val df1_schema = SparkSchemaUtil.convert(df1.schema);
      

   3. Especifique a estratégia de particionamento com base na coluna id:

      val partition_spec=PartitionSpec.builderFor(df1_schema).identity("id").build;
      

   4. Crie a tabela:

      val table=catalog.createTable(name,df1_schema,partition_spec);
      

   5. Adicione o gerenciador de armazenamento Iceberg e o SerDe como a propriedade da tabela:

      table.updateProperties().set("engine.hive.enabled", "true").commit();
      

   6. Grave os dados na tabela:

      df1.write.format("iceberg").mode("overwrite").save("default.example");
      

   7. Leia os dados:

      val read_df1=spark.read.format("iceberg").load("default.example");
      read_df1.show;
      

3. Altere o esquema da tabela. Veja um exemplo.

   1. Receba a tabela e adicione uma nova coluna grade:

      val table = catalog.loadTable(TableIdentifier.of("default", "example"));
      table.updateSchema.addColumn("grade", StringType.get()).commit();
      

   2. Verifique o novo esquema da tabela:

      table.schema.toString;
      

4. Insira mais dados e veja a evolução do esquema. Veja um exemplo.

   1. Adicione novos dados à tabela:

      val df2=Seq((5,"Amy","UX Design","Sophomore")).toDF("id","name","major","grade");
      df2.write.format("iceberg").mode("append").save("default.example");
      
      val df3=Seq((6,"Rachael","Economics","Freshman")).toDF("id","name","major","grade");
      df3.write.format("iceberg").mode("append").save("default.example");
      

   2. Verifique os novos dados inseridos:

      val read_df2=spark.read.format("iceberg").load("default.example");
      read_df2.show;
      

   3. Veja o histórico da tabela:

      spark.read.format("iceberg").load("default.example.history").show(truncate = false);
      

   4. Veja os snapshots:

      spark.read.format("iceberg").load("default.example.snapshots").show(truncate = false);
      

   5. Veja os arquivos de manifesto:

      spark.read.format("iceberg").load("default.example.manifests").show(truncate = false);
      

   6. Veja os arquivos de dados:

      spark.read.format("iceberg").load("default.example.files").show(truncate = false);
      

   7. Suponha que você cometeu um erro ao adicionar a linha com o valor de id=6 e queira voltar para uma versão correta da tabela:

      spark.read.format("iceberg").option("snapshot-id","2273922295095144317").load("default.example").show();
      

      Substitua snapshot-id pela versão que você quer consultar.

Usar tabelas do Hadoop para criar tabelas de Iceberg

1. Defina as configurações da tabela do Hadoop para criar tabelas de Iceberg no Spark Spark:

   import org.apache.hadoop.conf.Configuration
   import org.apache.iceberg.hadoop.HadoopTables
   import org.apache.iceberg.Table
   import org.apache.iceberg.Schema
   import org.apache.iceberg.types.Types._
   import org.apache.iceberg.PartitionSpec
   import org.apache.iceberg.spark.SparkSchemaUtil
   import org.apache.spark.sql._
   

2. Crie uma tabela para inserir e atualizar dados. Veja um exemplo.

   1. Crie uma tabela chamada example no banco de dados default:

      val conf = new Configuration();
      val tables = new HadoopTables(conf);
      

   2. Inserir dados de amostra:

      val df1 = Seq((1,"Vincent","Computer Science"),(2,"Dan", "Economics"),(3,"Bob", "Politics"),(4,"Cindy", "UX Design")).toDF("id","name","major");
      val df1_schema = SparkSchemaUtil.convert(df1.schema);
      

   3. Especifique a estratégia de particionamento com base na coluna id:

      val partition_spec=PartitionSpec.builderFor(df1_schema).identity("id").build;
      

   4. Crie a tabela:

      val table_location = "gs://<gcs-bucket-name>/hive-warehouse/<database-name>";
      val table = tables.create(df1_schema, partition_spec, table_location);
      

   5. Grave os dados na tabela:

      df1.write.format("iceberg").mode("overwrite").save(table_location);
      

   6. Leia os dados:

      val read_df1=spark.read.format("iceberg").load(table_location);
      read_df1.show;
      

3. Altere o esquema da tabela. Veja um exemplo.

   1. Receba a tabela e adicione uma nova coluna grade:

      val table = tables.load(table_location);
      table.updateSchema.addColumn("grade", StringType.get()).commit();
      

   2. Verifique o novo esquema da tabela:

      table.schema.toString;
      

4. Insira mais dados e veja a evolução do esquema. Veja um exemplo.

   1. Adicione novos dados à tabela:

      val df2=Seq((5,"Amy","UX Design","Sophomore")).toDF("id","name","major","grade");
      df2.write.format("iceberg").mode("append").save(table_location);
      
      val df3=Seq((6,"Rachael","Economics","Freshman")).toDF("id","name","major","grade");
      df3.write.format("iceberg").mode("append").save(table_location);
      

   2. Verifique os novos dados inseridos:

      val read_df2=spark.read.format("iceberg").load(table_location);
      read_df2.show;
      

   3. Veja o histórico da tabela:

      spark.read.format("iceberg").load("gs://<gcs-bucket-name>/hive-warehouse/<database-name>#history").show(truncate=false);
      

   4. Veja os snapshots:

      spark.read.format("iceberg").load("gs://<gcs-bucket-name>/hive-warehouse/<database-name>#snapshots").show(truncate=false);
      

   5. Veja os arquivos de manifesto:

      spark.read.format("iceberg").load("gs://<gcs-bucket-name>/hive-warehouse/<database-name>#manifests").show(truncate=false);
      

   6. Veja os arquivos de dados:

      spark.read.format("iceberg").load("gs://<gcs-bucket-name>/hive-warehouse/<database-name>#files").show(truncate=false);
      

   7. Volte para ver uma versão específica da tabela:

      spark.read.format("iceberg").option("snapshot-id","3943776515926014142L").format("iceberg").load(table_location).show;
      

      Substitua snapshot-id pela versão que você quer voltar e adicione "L" ao final. Por exemplo, "3943776515926014142L".

Usar uma tabela do Iceberg no Hive

O Iceberg oferece suporte a tabelas lidas usando o Hive com um StorageHandler. Observe que somente as versões Hive 2.x e 3.1.2 são compatíveis. Para mais informações, consulte Apache Iceberg - Hive. Além disso, inclua o arquivo JAR do ambiente de execução Hive do Iceberg no caminho de classe do Hive. Para fazer o download do arquivo JAR, consulte Downloads do Apache Iceberg.

Para sobrepor uma tabela do Hive sobre uma tabela de Iceberg, é necessário criar uma tabela do Iceberg com um catálogo do Hive ou uma tabela do Hadoop. Além disso, configure o Hive de acordo com isso para ler dados da tabela do Iceberg.

Ler tabela do Iceberg (catálogo do Hive) no Hive

1. Abra o cliente Hive e defina configurações para ler as tabelas de Iceberg na sessão do cliente Hive:

   add jar /path/to/iceberg-hive-runtime.jar;
   set iceberg.engine.hive.enabled=true;
   set engine.hive.enabled=true;
   set iceberg.mr.catalog=hive;
   set hive.vectorized.execution.enabled=false;
   

2. Ler esquema e dados de tabela. Veja um exemplo.

   1. Verifique o esquema da tabela e se o formato da tabela é Iceberg:

      describe formatted example;
      

   2. Leia os dados da tabela:

      select * from example;
      

Ler tabela do Iceberg (tabela do Hadoop) no Hive

1. Abra o cliente Hive e defina configurações para ler as tabelas de Iceberg na sessão do cliente Hive:

   add jar /path/to/iceberg-hive-runtime.jar;
   set engine.hive.enabled=true;
   set hive.vectorized.execution.enabled=false;
   

2. Ler esquema e dados de tabela. Veja um exemplo.

   1. Crie uma tabela externa (sobreponha uma tabela do Hive na tabela do Iceberg):

      CREATE EXTERNAL TABLE hadoop_table
      STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler'
      LOCATION 'gs://<gcs-bucket-name>/hive-warehouse/<database-name>'
      TBLPROPERTIES ('iceberg.catalog'='gs://<gcs-bucket-name>/hive-warehouse/<database-name>');
      

   2. Verifique o esquema da tabela e se o formato da tabela é Iceberg:

      describe formatted hadoop_table;
      

   3. Leia os dados da tabela:

      select * from hadoop_table;
      

Usar uma tabela de gelo no Presto

As consultas do Presto usam o conector Hive para receber locais de partição. Portanto, você precisa configurar o Presto corretamente para ler e gravar dados na tabela do Iceberg. Para mais informações, consulte Presto/Trino - Hive Connector e Presto/Trino - Iceberg Connector.

Configurações do Presto

1. Em cada nó de cluster do Dataproc, crie um arquivo chamado iceberg.properties /etc/presto/conf/catalog/iceberg.properties e configure o hive.metastore.uri da seguinte maneira:

   connector.name=iceberg
   hive.metastore.uri=thrift://<example.net:9083>
   

   Substitua example.net:9083 pelo host e pela porta corretos do serviço Hive Thrift.

2. Reinicie o serviço Presto para enviar as configurações:

   sudo systemctl restart presto.service
   

   Observação: é possível usar a ferramenta presto-admin para atualizar todos os nós no Batch.

Criar tabela de Iceberg em Presto

1. Abra o cliente Presto e use o conector "Iceberg" para receber o metastore:

   --catalog iceberg --schema default
   

2. Crie uma tabela para inserir e atualizar dados. Veja um exemplo.

   1. Crie uma tabela chamada example no banco de dados default:

      CREATE TABLE iceberg.default.example (
        id integer,
        name VARCHAR,
        major VARCHAR,
        grade VARCHAR)
      WITH (partitioning = ARRAY['major', 'grade']);
      

   2. Inserir dados de amostra:

      INSERT INTO iceberg.default.example
        VALUES (1, 'Vincent', 'Computer Science', 'Junior'), (2,'Dan', 'Economics', 'Senior'), (3,'Bob', 'Politics', 'Freshman');
      

   3. Leia os dados da tabela:

      SELECT * FROM iceberg.default.example;
      

   4. Insira mais dados para verificar os instantâneos:

      INSERT INTO example
        VALUES (4, 'Cindy', 'UX Design', 'Junior');
      
      INSERT INTO example
        VALUES (5, 'Amy', 'UX Design', 'Sophomore');
      

   5. Veja os snapshots:

      SELECT snapshot_id FROM iceberg.default."example$snapshots";
      

      Ao adicionar o comando ORDER BY committed_at DESC LIMIT 1;, é possível encontrar o ID do snapshot mais recente.

   6. Reverter para uma versão específica da tabela:

      CALL iceberg.system.rollback_to_snapshot('default', 'example', 8424394414541782448);
      

      Substitua snapshot-id pela versão que você quer consultar.

A seguir

• Gerenciar um serviço.
• Importar metadados para um serviço.