קריאה מ-Apache Iceberg אל Dataflow

יכול להיות שתצטרכו להטמיע ולהמיר נתונים לניתוח במורד הזרם בקנה מידה גדול, שמאוחסנים בפורמטים של טבלאות פתוחות של Apache Iceberg. הטמעת נתונים מ-Apache Iceberg ב-Dataflow באמצעות מחבר קלט/פלט מנוהל מאפשרת להריץ צינורות עיבוד נתונים מבוזרים ומקבילים באצווה, כדי לקרוא ולעבד מערכי נתונים בנפח גבוה ב-Hadoop, ב-Hive, בקטלוגים של REST ובמאגר המטא-נתונים של BigQuery.

אם אתם עובדים עם טבלאות ב-BigQuery ל-Apache Iceberg, השתמשו במקום זאת במחבר BigQueryIO עם BigQuery Storage API. טבלת היעד כבר צריכה להיות קיימת, כי אין תמיכה ביצירת טבלה דינמית.

יכולות

הקלט/פלט המנוהל תומך ביכולות הבאות של Apache Iceberg:

קטלוגים	Hadoop Hive קטלוגים שמבוססים על REST ‫BigQuery metastore (אם לא משתמשים ב-Portable Runner, נדרשת גרסה 2.62.0 ומעלה של Apache Beam SDK)
יכולות קריאה	קריאה באצווה
יכולות כתיבה	כתיבה באצווה כתיבה בסטרימינג יעדים דינמיים יצירת טבלאות דינמיות

בטבלאות BigQuery ל-Apache Iceberg, משתמשים במחבר BigQueryIO עם BigQuery Storage API. הטבלה כבר צריכה להיות קיימת, כי אין תמיכה ביצירת טבלה דינמית.

יחסי תלות

מוסיפים לפרויקט את יחסי התלות הבאים:

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>

דוגמה

בדוגמה הבאה, המערכת קוראת מטבלת Apache Iceberg וכותבת את הנתונים בקובצי טקסט.

Java

כדי לבצע אימות ב-Dataflow, צריך להגדיר את Application Default Credentials. למידע נוסף, ראו הגדרת אימות לסביבת פיתוח מקומית.

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();
  }
}

המאמרים הבאים

• כתיבה ב-Apache Iceberg.
• כתיבה בסטרימינג ל-Apache Iceberg באמצעות קטלוג REST של Lakehouse.
• מידע נוסף על קלט/פלט מנוהל