Verwenden Sie einen Delta Lake-Cluster mit Flink

PDFRSS

Mit Amazon-EMR-Version 6.11 und höher können Sie Delta Lake mit Ihrem Flink-Cluster verwenden. In den folgenden Beispielen wird der verwendet AWS CLI , um mit Delta Lake auf einem Amazon EMR Flink-Cluster zu arbeiten.

Anmerkung

Amazon EMR unterstützt die DataStream Flink-API, wenn Sie Delta Lake mit einem Flink-Cluster verwenden.

Einen Delta-Lake-Cluster erstellen

1. Erstellen Sie eine Datei, delta_configurations.json, mit folgendem Inhalt:

   [{"Classification":"delta-defaults",  
       "Properties":{"delta.enabled":"true"}}]

2. Erstellen Sie einen Cluster mit der folgenden Konfiguration. Ersetzen Sie example Amazon S3 bucket path und subnet ID durch Ihre eigenen Werte.

   aws emr create-cluster 
   --release-label emr-6.11.0   
   --applications Name=Flink  
   --configurations file://delta_configurations.json   
   --region us-east-1  --name My_Spark_Delta_Cluster  
   --log-uri  s3://amzn-s3-demo-bucket/  
   --instance-type m5.xlarge  
   --instance-count 3   
   --service-role EMR_DefaultRole_V2  
   --ec2-attributes  InstanceProfile=EMR_EC2_DefaultRole,SubnetId=subnet-1234567890abcdef0

Eine Flink-Yarn-Sitzung installieren

Um eine Flink-Yarn-Sitzung zu initialisieren, führen Sie den folgenden Befehl aus:

flink-yarn-session -d

Einen Flink-Auftrag mit Delta Lake erstellen

Die folgenden Beispiele zeigen, wie Sie mit SBT oder Maven für die Erstellung Ihres Flink-Auftrags mit Delta Lake verwenden.

sbt

sbt ist ein Build-Tool für Scala, das Sie mit wenig bis gar keiner Konfiguration verwenden können, wenn Sie kleine Projekte haben.

libraryDependencies ++= Seq(
  "io.delta" %% "delta-flink" % deltaConnectorsVersion % "provided",
  "io.delta" %% "delta-standalone" % deltaConnectorsVersion % "provided",
  "org.apache.flink" %% "flink-clients" % flinkVersion % "provided",
  "org.apache.flink" %% "flink-parquet" % flinkVersion % "provided",
  "org.apache.hadoop" % "hadoop-client" % hadoopVersion % "provided",
  "org.apache.flink" % "flink-table-common" % flinkVersion % "provided",
  "org.apache.flink" %% "flink-table-runtime" % flinkVersion % "provided")

Maven

Maven ist ein Open-Source-Tool zur Build-Automatisierung von der Apache Software Foundation. Mit Maven können Sie einen Flink-Auftrag mit Delta Lake auf Amazon EMR erstellen, veröffentlichen und bereitstellen.

<project>
<properties>
    <scala.main.version>2.12</scala.main.version>
    <delta-connectors-version>0.6.0</delta-connectors-version>
    <flink-version>1.16.1</flink-version>
    <hadoop-version>3.1.0</hadoop-version>
</properties>

<dependencies>
    <dependency>
        <groupId>io.delta</groupId>
        <artifactId>delta-flink</artifactId>
        <version>$delta-connectors-version</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>io.delta</groupId>
        <artifactId>delta-standalone_$scala-main-version</artifactId>
        <version>$delta-connectors-version</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>$flink-version</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-parquet</artifactId>
        <version>$flink-version</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>$hadoop-version</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-common</artifactId>
        <version>$flink-version</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-runtime</artifactId>
        <version>$flink-version</version>
        <scope>provided</scope>
    </dependency>
</dependencies>

Mit der Flink-Datastream-API in eine Delta-Tabelle schreiben

Verwenden Sie das folgende Beispiel, um eine DeltaSink zu erstellen, mit der Sie in die Tabelle schreiben können deltaTablePath:

public static DataStream<RowData> createDeltaSink(
        DataStream<RowData> stream,
        String deltaTablePath,
        RowType rowType) {
    Configuration configuration = new Configuration();
    DeltaSink<RowData> deltaSink = DeltaSink
            .forRowData(
                    new org.apache.flink.core.fs.Path(deltaTablePath),
                    configuration,
                    rowType)
            .build();
    stream.sinkTo(deltaSink);
    return stream;
}

Mit der Flink-Datastream-API aus einer Delta-Tabelle lesen

Verwenden Sie das folgende Beispiel, um eine Grenze zu erstellen DeltaSource , die aus der Tabelle mit einem gelesen werden soll deltaTablePath:

public static DataStream<RowData> createBoundedDeltaSourceAllColumns(
        StreamExecutionEnvironment env,
        String deltaTablePath) {
    Configuration configuration = new Configuration();
    DeltaSource<RowData> deltaSource = DeltaSource
            .forBoundedRowData(
                    new org.apache.flink.core.fs.Path(deltaTablePath),
                    configuration)
            .build();

    return env.fromSource(deltaSource, WatermarkStrategy.noWatermarks(), "delta-source");
}

Sink-Erstellung mit Multi-Cluster-Unterstützung für Delta Lake Standalone

Verwenden Sie das folgende Beispiel, um eine Tabelle DeltaSink zum Schreiben in eine Tabelle mit A deltaTablePath - und Multi-Cluster-Unterstützung zu erstellen:

public DataStream<RowData> createDeltaSink(
        DataStream<RowData> stream,
        String deltaTablePath) {
    Configuration configuration = new Configuration();
    configuration.set("spark.delta.logStore.s3.impl", "io.delta.storage.S3DynamoDBLogStore");
    configuration.set("spark.io.delta.storage.S3DynamoDBLogStore.ddb.tableName", "delta_log");
    configuration.set("spark.io.delta.storage.S3DynamoDBLogStore.ddb.region", "us-east-1");
        
    DeltaSink<RowData> deltaSink = DeltaSink
        .forRowData(
            new Path(deltaTablePath),
            configuration,
            rowType)
        .build();
    stream.sinkTo(deltaSink);
    return stream;
}

Den Flink-Auftrag ausführen

Den folgenden Befehl verwenden, um Ihren Job auszuführen:

flink run FlinkJob.jar