Gebeurtenisberichten schrijven naar Azure Data Lake Storage Gen2 met Apache Flink® DataStream-API

Artikel
09/20/2024

Belangrijk

Azure HDInsight op AKS is op 31 januari 2025 buiten gebruik gesteld. Kom meer te weten via deze aankondiging.

U moet uw workloads migreren naar Microsoft Fabric- of een gelijkwaardig Azure-product om plotselinge beëindiging van uw workloads te voorkomen.

Belangrijk

Deze functie is momenteel beschikbaar als preview-versie. De aanvullende gebruiksvoorwaarden voor Microsoft Azure Previews meer juridische voorwaarden bevatten die van toepassing zijn op Azure-functies die bèta, in preview of anderszins nog niet in algemene beschikbaarheid zijn vrijgegeven. Zie Azure HDInsight in AKS preview-informatievoor meer informatie over deze specifieke preview. Voor vragen of suggesties voor functies dient u een aanvraag in op AskHDInsight- met de details en volgt u ons voor meer updates over Azure HDInsight Community-.

Apache Flink gebruikt bestandssystemen om gegevens te gebruiken en permanent op te slaan, zowel voor de resultaten van toepassingen als voor fouttolerantie en herstel. In dit artikel leert u hoe u gebeurtenisberichten schrijft naar Azure Data Lake Storage Gen2 met datastream-API.

Voorwaarden

Apache Flink-cluster in HDInsight op AKS-
Apache Kafka-cluster in HDInsight-
- U moet ervoor zorgen dat de netwerkinstellingen worden verzorgd zoals beschreven in Apache Kafka gebruiken in HDInsight. Zorg ervoor dat HDInsight op AKS- en HDInsight-clusters zich in hetzelfde virtuele netwerk bevinden.
MSI gebruiken voor toegang tot ADLS Gen2
IntelliJ voor ontwikkeling op een Virtuele Azure-machine in HDInsight in AKS Virtual Network

Apache Flink FileSystem-connector

Deze bestandssysteemconnector biedt voor zowel BATCH als STREAMING dezelfde garanties en is ontworpen om exacte eenmalige semantiek te bieden voor STREAMING-uitvoering. Zie Flink DataStream Filesystemvoor meer informatie.

Apache Kafka connector

Flink biedt een Apache Kafka-connector voor het lezen van gegevens uit en het schrijven van gegevens naar Kafka-topics met exacte eenmaal garanties. Zie Apache Kafka Connectorvoor meer informatie.

Het project bouwen voor Apache Flink

pom.xml op IntelliJ IDEA

<properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <flink.version>1.17.0</flink.version>
        <java.version>1.8</java.version>
        <scala.binary.version>2.12</scala.binary.version>
        <kafka.version>3.2.0</kafka.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-java -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-files -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-files</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka</artifactId>
            <version>${flink.version}</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.0.0</version>
                <configuration>
                    <appendAssemblyId>false</appendAssemblyId>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

-programma voor ADLS Gen2 Sink-

abfsGen2.java

Notitie

Vervang Apache Kafka op de HDInsight-cluster bootStrapServers door uw eigen brokers voor Kafka 3.2

package contoso.example;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.time.Duration;

public class KafkaSinkToGen2 {
    public static void main(String[] args) throws Exception {
        // 1. get stream execution env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
         
        Configuration flinkConfig = new Configuration(); 

         flinkConfig.setString("classloader.resolve-order", "parent-first"); 

         env.getConfig().setGlobalJobParameters(flinkConfig);  

        // 2. read kafka message as stream input, update your broker ip's
        String brokers = "<update-broker-ip>:9092,<update-broker-ip>:9092,<update-broker-ip>:9092";
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers(brokers)
                .setTopics("click_events")
                .setGroupId("my-group")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
        stream.print();

        // 3. sink to gen2, update container name and storage path
        String outputPath  = "abfs://<container-name>@<storage-path>.dfs.core.windows.net/flink/data/click_events";
        final FileSink<String> sink = FileSink
                .forRowFormat(new Path(outputPath), new SimpleStringEncoder<String>("UTF-8"))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(Duration.ofMinutes(2))
                                .withInactivityInterval(Duration.ofMinutes(3))
                                .withMaxPartSize(MemorySize.ofMebiBytes(5))
                                .build())
                .build();

        stream.sinkTo(sink);

        // 4. run stream
        env.execute("Kafka Sink To Gen2");
    }
}

Package JAR en dien deze in bij Apache Flink.

Upload het jar-bestand naar ABFS.
Geef de informatie van de taak-JAR door bij de creatie van het AppMode-cluster.

Notitie

Zorg ervoor dat u classloader.resolve-order toevoegt als 'parent-first' en hadoop.classpath.enable als true
Selecteer Taaklogboekaggregatie om taaklogboeken naar het opslagaccount te pushen.
U kunt zien dat de taak wordt uitgevoerd.

Validatie van streaminggegevens op ADLS Gen2

We zien dat de click_events naar ADLS Gen2 wordt gestreamd.

Schermopname met ADLS Gen2-uitvoer. Schermopname met de gebeurtenisuitvoer van Flink click.

U kunt een rollend beleid opgeven waarmee het bestand in uitvoering wordt meegerold op een van de volgende drie voorwaarden:

.withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(Duration.ofMinutes(5))
                                .withInactivityInterval(Duration.ofMinutes(3))
                                .withMaxPartSize(MemorySize.ofMebiBytes(5))
                                .build())

Referentie

Apache Kafka-connector
Flink DataStream Filesystem
Apache Flink Website
Apache, Apache Kafka, Kafka, Apache Flink, Flink en bijbehorende opensource-projectnamen zijn handelsmerken van de Apache Software Foundation (ASF).

Delen via