Leggere da Apache Iceberg a Dataflow

Potresti dover importare e trasformare i dati per l'analisi downstream su larga scala archiviati nei formati di tabella aperti Apache Iceberg. L'importazione dei dati Apache Iceberg in Dataflow utilizzando il connettore I/O gestito consente di eseguire pipeline batch parallele distribuite per leggere ed elaborare set di dati ad alto volume in Hadoop, Hive, cataloghi REST e metastore BigQuery.

Se lavori con tabelle BigQuery per Apache Iceberg, utilizza invece il connettore BigQueryIO con l'API BigQuery Storage. La tabella di destinazione deve già esistere, poiché la creazione di tabelle dinamiche non è supportata.

Funzionalità

Managed I/O supporta le seguenti funzionalità per Apache Iceberg:

Cataloghi
  • Hadoop
  • Hive
  • Cataloghi basati su REST
  • Metastore BigQuery (richiede l'SDK Apache Beam 2.62.0 o versioni successive se non utilizzi il runner portatile)
Funzionalità di lettura Lettura batch
Funzionalità di scrittura

Per le tabelle BigQuery per Apache Iceberg, utilizza il connettore BigQueryIO con l'API BigQuery Storage. La tabella deve già esistere; la creazione dinamica delle tabelle non è supportata.

Dipendenze

Aggiungi le seguenti dipendenze al tuo progetto:

Java 

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-sdks-java-managed</artifactId>
  <version>${beam.version}</version>
</dependency>

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-sdks-java-io-iceberg</artifactId>
  <version>${beam.version}</version>
</dependency>

Esempio

L'esempio seguente legge da una tabella Apache Iceberg e scrive i dati in file di testo.

Java

Per eseguire l'autenticazione in Dataflow, configura le Credenziali predefinite dell'applicazione. Per saperne di più, consulta Configura l'autenticazione per un ambiente di sviluppo locale. 

import com.google.common.collect.ImmutableMap;
import java.util.Map;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.managed.Managed;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.PCollectionRowTuple;
import org.apache.beam.sdk.values.TypeDescriptors;

public class ApacheIcebergRead {

  static final String CATALOG_TYPE = "hadoop";

  public interface Options extends PipelineOptions {
    @Description("The URI of the Apache Iceberg warehouse location")
    String getWarehouseLocation();

    void setWarehouseLocation(String value);

    @Description("Path to write the output file")
    String getOutputPath();

    void setOutputPath(String value);

    @Description("The name of the Apache Iceberg catalog")
    String getCatalogName();

    void setCatalogName(String value);

    @Description("The name of the table to write to")
    String getTableName();

    void setTableName(String value);
  }

  public static void main(String[] args) {

    // Parse the pipeline options passed into the application. Example:
    //   --runner=DirectRunner --warehouseLocation=$LOCATION --catalogName=$CATALOG \
    //   --tableName= $TABLE_NAME --outputPath=$OUTPUT_FILE
    // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
    Pipeline pipeline = Pipeline.create(options);

    // Configure the Iceberg source I/O
    Map catalogConfig = ImmutableMap.<String, Object>builder()
        .put("warehouse", options.getWarehouseLocation())
        .put("type", CATALOG_TYPE)
        .build();

    ImmutableMap<String, Object> config = ImmutableMap.<String, Object>builder()
        .put("table", options.getTableName())
        .put("catalog_name", options.getCatalogName())
        .put("catalog_properties", catalogConfig)
        .build();

    // Build the pipeline.
    pipeline.apply(Managed.read(Managed.ICEBERG).withConfig(config))
        .getSinglePCollection()
        // Format each record as a string with the format 'id:name'.
        .apply(MapElements
            .into(TypeDescriptors.strings())
            .via((row -> {
              return String.format("%d:%s",
                  row.getInt64("id"),
                  row.getString("name"));
            })))
        // Write to a text file.
        .apply(
            TextIO.write()
                .to(options.getOutputPath())
                .withNumShards(1)
                .withSuffix(".txt"));

    pipeline.run().waitUntilFinish();
  }
}

Passaggi successivi