Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

Behalten Sie die Best Practices für Managed Service für Apache Flink-Anwendungen bei

Dieser Abschnitt enthält Informationen und Empfehlungen für die Entwicklung eines stabilen, leistungsstarken Managed Service für Apache Flink-Anwendungen.

Minimiere die Größe des Uber-JAR

Java/Scala Die Anwendung muss in einer uber (super/fat) -JAR gepackt sein und alle zusätzlichen erforderlichen Abhängigkeiten enthalten, die nicht bereits von der Laufzeit bereitgestellt werden. Die Größe der Uber-JAR wirkt sich jedoch auf die Start- und Neustartzeiten der Anwendung aus und kann dazu führen, dass die JAR das Limit von 512 MB überschreitet.

Um die Bereitstellungszeit zu optimieren, sollte Ihr Uber-JAR Folgendes nicht enthalten:

Bereitgestellte Abhängigkeiten

Der Managed Service für die Apache Flink-Laufzeit bietet eine Reihe von Abhängigkeiten. Diese Abhängigkeiten sollten nicht im Fat-JAR enthalten sein und müssen in der POM-Datei einen provided Gültigkeitsbereich haben oder in der maven-shade-plugin Konfiguration explizit ausgeschlossen werden. Jede dieser Abhängigkeiten, die im Fat-JAR enthalten sind, wird zur Laufzeit ignoriert, erhöht jedoch die Größe des JARs, was während der Bereitstellung zu einem zusätzlichen Aufwand führt.

Von der Laufzeit bereitgestellte Abhängigkeiten in den Laufzeitversionen 1.18, 1.19 und 1.20:

Darüber hinaus stellt die Laufzeitumgebung die Bibliothek bereit, die zum Abrufen der Laufzeiteigenschaften von Anwendungen in Managed Service for Apache Flink, verwendet wird. com.amazonaws:aws-kinesisanalytics-runtime:1.2.0

Alle von der Runtime bereitgestellten Abhängigkeiten müssen die folgenden Empfehlungen verwenden, damit sie nicht in das Uber-JAR aufgenommen werden:

Jede angegebene Abhängigkeit, die versehentlich in der Uber-JAR enthalten ist, wird zur Laufzeit ignoriert, da die übergeordnete Klasse von Apache Flink geladen wird. Weitere Informationen finden Sie unter Parent-First-Patterns in der Apache Flink-Dokumentation.

Konnektoren

Die meisten Konnektoren, mit Ausnahme des FileSystem Connectors, die nicht in der Laufzeit enthalten sind, müssen in der POM-Datei mit dem Standardbereich () enthalten sein. compile

Weitere Empfehlungen

In der Regel sollte Ihr Apache Flink Uber JAR, das Managed Service for Apache Flink zur Verfügung gestellt wird, den Mindestcode enthalten, der zum Ausführen der Anwendung erforderlich ist. Das Einbeziehen von Abhängigkeiten, die die Quellklassen, Testdatensätze oder den Bootstrapping-Status beinhalten, sollte in diesem JAR nicht enthalten sein. Wenn statische Ressourcen zur Laufzeit abgerufen werden müssen, teilen Sie dieses Problem in eine Ressource wie Amazon S3 auf. Beispiele hierfür sind State-Bootstraps oder ein Inferenzmodell.

Nehmen Sie sich etwas Zeit, um Ihren tiefen Abhängigkeitsbaum zu prüfen und Abhängigkeiten zu entfernen, die nicht zur Laufzeit gehören.

Obwohl Managed Service für Apache Flink Jar-Größen von 512 MB unterstützt, sollte dies als Ausnahme von der Regel betrachtet werden. Apache Flink unterstützt derzeit in seiner Standardkonfiguration Jar-Größen von ~104 MB, und das sollte die maximale Zielgröße eines benötigten Jars sein.

Fehlertoleranz: Prüfpunkte und Savepoints

Verwenden Sie Checkpoints und Savepoints, um Fehlertoleranz in Ihrer Managed Service for Apache Flink-Anwendung zu implementieren. Berücksichtigen Sie bei Entwicklung und Wartung Ihrer Anwendung Folgendes:

Weitere Informationen zur Implementierung der Fehlertoleranz finden Sie unter Implementieren Sie Fehlertoleranz.

Nicht unterstützte Konnektor-Versionen

Ab Apache Flink Version 1.15 oder höher verhindert Managed Service für Apache Flink automatisch, dass Anwendungen gestartet oder aktualisiert werden, wenn sie nicht unterstützte Kinesis-Connector-Versionen verwenden, die in Anwendungs-JARs gebündelt sind. Stellen Sie beim Upgrade auf Managed Service für Apache Flink Version 1.15 oder höher sicher, dass Sie den neuesten Kinesis-Connector verwenden. Dies ist jede Version, die Version 1.15.2 entspricht oder neuer ist. Alle anderen Versionen werden von Managed Service für Apache Flink nicht unterstützt, da sie Konsistenzprobleme oder Fehler bei der Funktion „Mit Savepoint beenden“ verursachen können, wodurch saubere Operationen verhindert werden. stop/update Weitere Informationen zur Connector-Kompatibilität in Amazon Managed Service für Apache Flink-Versionen finden Sie unter Apache Flink-Konnektoren.

Leistung und Parallelität

Ihre Anwendung kann so skaliert werden, dass sie jedes Durchsatzniveau erreicht, indem Sie die Parallelität Ihrer Anwendung optimieren und Leistungsprobleme vermeiden. Berücksichtigen Sie bei Entwicklung und Wartung Ihrer Anwendung Folgendes:

Weitere Informationen über die Implementierung von Skalierung finden Sie unter Implementieren Sie Anwendungsskalierung.

Parallelität pro Operator festlegen

Standardmäßig ist die Parallelität für alle Operatoren auf Anwendungsebene festgelegt. Sie können die Parallelität eines einzelnen Operators mithilfe der DataStream API überschreiben. .setParallelism(x) Sie können eine Operatorparallelität auf eine beliebige Parallelität festlegen, die gleich oder niedriger als die Anwendungsparallelität ist.

Wenn möglich, definieren Sie die Operatorparallelität als Funktion der Anwendungsparallelität. Auf diese Weise variiert die Operatorparallelität mit der Anwendungsparallelität. Wenn Sie beispielsweise automatische Skalierung verwenden, variieren alle Operatoren ihre Parallelität im gleichen Verhältnis:

int appParallelism = env.getParallelism();
...
...ops.setParalleism(appParallelism/2);

In einigen Fällen möchten Sie möglicherweise die Operatorparallelität auf eine Konstante setzen. Stellen Sie beispielsweise die Parallelität einer Kinesis Stream-Quelle auf die Anzahl der Shards ein. In diesen Fällen sollten Sie erwägen, die Operatorparallelität als Anwendungskonfigurationsparameter zu übergeben, um ihn zu ändern, ohne den Code zu ändern, z. B. um den Quellstream erneut zu teilen.

Protokollierung

Mithilfe von Protokollen können Sie die Leistung und die Fehlerbedingungen Ihrer Anwendung überwachen. CloudWatch Berücksichtigen Sie bei der Konfiguration der Protokollierung für Ihre Anwendung Folgendes:

Weitere Informationen über die Implementierung von Protokollierung finden Sie unter .

Codierung

Sie können Ihre Anwendung leistungsfähig und stabil machen, indem Sie empfohlene Programmierpraktiken anwenden. Berücksichtigen Sie beim Schreiben von Anwendungscode Folgendes:

Verwalten von Anmeldeinformationen.

Sie sollten keine langfristigen Anmeldeinformationen in Produktionsanwendungen (oder andere) integrieren. Long-term Anmeldeinformationen werden wahrscheinlich in ein Versionskontrollsystem eingecheckt und können leicht verloren gehen. Stattdessen können Sie der Anwendung Managed Service for Apache Flink eine Rolle zuordnen und dieser Rolle Berechtigungen gewähren. Die laufende Flink-Anwendung kann dann temporäre Anmeldeinformationen mit den entsprechenden Berechtigungen aus der Umgebung auswählen. Falls eine Authentifizierung für einen Dienst erforderlich ist, der nicht nativ in IAM integriert ist, z. B. für eine Datenbank, für die ein Benutzername und ein Passwort zur Authentifizierung erforderlich sind, sollten Sie erwägen, Geheimnisse im AWS Secrets Manager zu speichern.

Viele AWS native Dienste unterstützen die Authentifizierung:

Lesen aus Quellen mit wenigen shards/partitions

Beim Lesen aus Apache Kafka oder einem Kinesis Data Stream kann es zu einer Diskrepanz zwischen der Parallelität des Streams (die Anzahl der Partitionen für Kafka und die Anzahl der Shards für Kinesis) und der Parallelität der Anwendung kommen. Bei einem naiven Design kann die Parallelität einer Anwendung nicht über die Parallelität eines Streams hinaus skaliert werden: Jede Unteraufgabe eines Quell-Operators kann nur aus einem oder mehreren Daten lesen. shards/partitions Das bedeutet für einen Stream mit nur 2 Shards und eine Anwendung mit einer Parallelität von 8, dass nur zwei Unteraufgaben den Stream tatsächlich verbrauchen und 6 Unteraufgaben inaktiv bleiben. Dies kann den Durchsatz der Anwendung erheblich einschränken, insbesondere wenn die Deserialisierung teuer ist und von der Quelle durchgeführt wird (was die Standardeinstellung ist).

Um diesen Effekt zu mildern, können Sie den Stream entweder skalieren. Dies ist jedoch möglicherweise nicht immer wünschenswert oder möglich. Alternativ können Sie die Quelle so umstrukturieren, dass sie keine Serialisierung durchführt und nur die byte[] weitergibt. Anschließend können Sie die Daten umverteilen, um sie gleichmäßig auf alle Aufgaben zu verteilen, und die Daten dann dort deserialisieren. Auf diese Weise können Sie alle Unteraufgaben für die Deserialisierung nutzen, und dieser potenziell teure Vorgang ist nicht mehr an die Nummer des Streams gebunden. shards/partitions

Aktualisierungsintervall für Studio-Notebooks

Wenn Sie das Aktualisierungsintervall für die Absatzergebnisse ändern, setzen Sie es auf einen Wert, der mindestens 1000 Millisekunden beträgt.

Optimale Leistung des Studio-Notebooks

Wir haben mit der folgenden Aussage getestet und festgestellt, dass die optimale Leistung, events-per-second multipliziert number-of-keys mit, unter 25.000.000 lag. Das war bei events-per-second weniger als 150.000.

SELECT key, sum(value) FROM key-values GROUP BY key

Wie sich Strategien mit Wasserzeichen und ungenutzte Shards auf Zeitfenster auswirken

Beim Lesen von Ereignissen aus Apache Kafka und Kinesis Data Streams kann die Quelle die Ereigniszeit basierend auf den Attributen des Streams festlegen. Im Fall von Kinesis entspricht die Ereigniszeit der ungefähren Ankunftszeit von Ereignissen. Die Festlegung der Ereigniszeit an der Quelle für Ereignisse reicht jedoch nicht aus, damit eine Flink-Anwendung die Ereigniszeit verwenden kann. Die Quelle muss außerdem Wasserzeichen erzeugen, die Informationen über die Ereigniszeit von der Quelle an alle anderen Operatoren weitergeben. Die Flink-Dokumentation bietet einen guten Überblick darüber, wie dieser Prozess funktioniert.

Standardmäßig ist der Zeitstempel eines aus Kinesis gelesenen Ereignisses auf die ungefähre Ankunftszeit gesetzt, die von Kinesis bestimmt wird. Eine zusätzliche Voraussetzung dafür, dass die Ereigniszeit in der Anwendung funktioniert, ist eine Wasserzeichen-Strategie.

WatermarkStrategy<String> s = WatermarkStrategy
    .<String>forMonotonousTimestamps()
    .withIdleness(Duration.ofSeconds(...));

Die Wasserzeichenstrategie wird dann auf einen DataStream mit der Methode assignTimestampsAndWatermarks angewendet. Es gibt einige nützliche integrierte Strategien:

Aus der Flink-Dokumentation:

Jede parallel Unteraufgabe einer Quellfunktion generiert ihre Wasserzeichen normalerweise unabhängig. Diese Wasserzeichen definieren die Ereigniszeit an dieser bestimmten parallelen Quelle.

Während die Wasserzeichen durch das Streaming-Programm fließen, verschieben sie die Ereigniszeit bei den Operatoren, wo sie ankommen. Immer wenn ein Betreiber seine Ereigniszeit vorverlegt, generiert er nachgelagert ein neues Wasserzeichen für seine nachfolgenden Operatoren.

Manche Operatoren verbrauchen mehrere Eingabestreams; zum Beispiel eine Union oder Operatoren, die einer keyBy(...)- oder partition(...)-Funktion folgen. Die aktuelle Ereigniszeit eines solchen Operators ist das Minimum der Ereigniszeiten seiner Eingabestreams. Wenn seine Eingabestreams ihre Ereigniszeiten aktualisieren, tut dies auch der Operator.

Das heißt, wenn eine Quell-Unteraufgabe von einem inaktiven Shard aus konsumiert, erhalten nachgeschaltete Operatoren keine neuen Wasserzeichen von dieser Unteraufgabe, sodass die Verarbeitung für alle nachgeschalteten Operatoren, die Zeitfenster verwenden, zum Stillstand kommt. Um dies zu vermeiden, können Kunden die withIdleness-Option zur Wasserzeichenstrategie hinzufügen. Bei dieser Option schließt ein Operator bei der Berechnung der Ereigniszeit des Operators die Wasserzeichen aus inaktiven Upstream-Unteraufgaben aus. Die Unteraufgabe im Leerlauf verhindert daher nicht mehr das Vorrücken der Eventzeit in nachgeschalteten Operatoren.

Je nachdem, welchen Shard-Zuweiser Sie verwenden, werden einigen Workern möglicherweise keine Kinesis-Shards zugewiesen. In diesem Fall verhalten sich diese Worker wie im Leerlauf der Quelle, auch wenn alle Kinesis-Shards kontinuierlich Ereignisdaten liefern. Sie können dieses Risiko mindern, indem Sie es uniformShardAssigner zusammen mit dem Quell-Operator verwenden. Dadurch wird sichergestellt, dass alle Quell-Unteraufgaben über zu verarbeitende Shards verfügen, solange die Anzahl der Worker kleiner oder gleich der Anzahl der aktiven Shards ist.

Die Option „Leerlauf“ mit den integrierten Wasserzeichen-Strategien verschiebt die Eventzeit jedoch nicht, wenn keine Unteraufgabe ein Ereignis liest, d. h. es gibt keine Ereignisse im Stream. Dies wird besonders bei Testfällen sichtbar, in denen eine begrenzte Menge von Ereignissen aus dem Stream gelesen wird. Da die Eventzeit nicht nach dem Lesen des letzten Events verschoben wird, wird das letzte Fenster (das das letzte Event enthält) nicht geschlossen.

Zusammenfassung

Beispiel

Die folgende Anwendung liest aus einem Stream und erstellt Sitzungsfenster auf der Grundlage der Ereigniszeit.

Properties consumerConfig = new Properties();
consumerConfig.put(AWSConfigConstants.AWS_REGION, "eu-west-1");
consumerConfig.put(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "TRIM_HORIZON");

FlinkKinesisConsumer<String> consumer = new FlinkKinesisConsumer<>("...", new SimpleStringSchema(), consumerConfig);

WatermarkStrategy<String> s = WatermarkStrategy
    .<String>forMonotonousTimestamps()
    .withIdleness(Duration.ofSeconds(15));
    
env.addSource(consumer)
    .assignTimestampsAndWatermarks(s)
    .map(new MapFunction<String, Long>() {
        @Override
        public Long map(String s) throws Exception {
            return Long.parseLong(s);
        }
    })
    .keyBy(l -> 0l)
    .window(EventTimeSessionWindows.withGap(Time.seconds(10)))
    .process(new ProcessWindowFunction<Long, Object, Long, TimeWindow>() {
        @Override
        public void process(Long aLong, ProcessWindowFunction<Long, Object, Long, TimeWindow>.Context context, Iterable<Long>iterable, Collector<Object> collector) throws Exception {
            long count = StreamSupport.stream(iterable.spliterator(), false).count();
            long timestamp = context.currentWatermark();

            System.out.print("XXXXXXXXXXXXXX Window with " + count + " events");
            System.out.println("; Watermark: " + timestamp + ", " + Instant.ofEpochMilli(timestamp));


            for (Long l : iterable) {
                System.out.println(l);
            }
        }
    });

Im folgenden Beispiel werden 8 Ereignisse in einen Stream mit 16 Shards geschrieben (die ersten 2 und das letzte Ereignis landen zufällig im selben Shard).

$ aws kinesis put-record --stream-name hp-16 --partition-key 1 --data MQ==
$ aws kinesis put-record --stream-name hp-16 --partition-key 2 --data Mg==
$ aws kinesis put-record --stream-name hp-16 --partition-key 3 --data Mw==
$ date

{
    "ShardId": "shardId-000000000012",
    "SequenceNumber": "49627894338614655560500811028721934184977530127978070210"
}
{
    "ShardId": "shardId-000000000012",
    "SequenceNumber": "49627894338614655560500811028795678659974022576354623682"
}
{
    "ShardId": "shardId-000000000014",
    "SequenceNumber": "49627894338659257050897872275134360684221592378842022114"
}
Wed Mar 23 11:19:57 CET 2022

$ sleep 10
$ aws kinesis put-record --stream-name hp-16 --partition-key 4 --data NA==
$ aws kinesis put-record --stream-name hp-16 --partition-key 5 --data NQ==
$ date

{
    "ShardId": "shardId-000000000010",
    "SequenceNumber": "49627894338570054070103749783042116732419934393936642210"
}
{
    "ShardId": "shardId-000000000014",
    "SequenceNumber": "49627894338659257050897872275659034489934342334017700066"
}
Wed Mar 23 11:20:10 CET 2022

$ sleep 10
$ aws kinesis put-record --stream-name hp-16 --partition-key 6 --data Ng==
$ date

{
    "ShardId": "shardId-000000000001",
    "SequenceNumber": "49627894338369347363316974173886988345467035365375213586"
}
Wed Mar 23 11:20:22 CET 2022

$ sleep 10
$ aws kinesis put-record --stream-name hp-16 --partition-key 7 --data Nw==
$ date

{
    "ShardId": "shardId-000000000008",
    "SequenceNumber": "49627894338525452579706688535878947299195189349725503618"
}
Wed Mar 23 11:20:34 CET 2022

$ sleep 60
$ aws kinesis put-record --stream-name hp-16 --partition-key 8 --data OA==
$ date

{
    "ShardId": "shardId-000000000012",
    "SequenceNumber": "49627894338614655560500811029600823255837371928900796610"
}
Wed Mar 23 11:21:27 CET 2022

Diese Eingabe sollte zu 5 Sitzungsfenstern führen: Ereignis 1,2,3; Ereignis 4,5; Ereignis 6; Ereignis 7; Ereignis 8. Das Programm liefert jedoch nur die ersten 4 Fenster.

11:59:21,529 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 5 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000006,HashKeyRange: {StartingHashKey: 127605887595351923798765477786913079296,EndingHashKey: 148873535527910577765226390751398592511},SequenceNumberRange: {StartingSequenceNumber: 49627894338480851089309627289524549239292625588395704418,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,530 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 5 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000006,HashKeyRange: {StartingHashKey: 127605887595351923798765477786913079296,EndingHashKey: 148873535527910577765226390751398592511},SequenceNumberRange: {StartingSequenceNumber: 49627894338480851089309627289524549239292625588395704418,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 0
11:59:21,530 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 6 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000007,HashKeyRange: {StartingHashKey: 148873535527910577765226390751398592512,EndingHashKey: 170141183460469231731687303715884105727},SequenceNumberRange: {StartingSequenceNumber: 49627894338503151834508157912666084957565273949901684850,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,530 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 6 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000010,HashKeyRange: {StartingHashKey: 212676479325586539664609129644855132160,EndingHashKey: 233944127258145193631070042609340645375},SequenceNumberRange: {StartingSequenceNumber: 49627894338570054070103749782090692112383219034419626146,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,530 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 6 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000007,HashKeyRange: {StartingHashKey: 148873535527910577765226390751398592512,EndingHashKey: 170141183460469231731687303715884105727},SequenceNumberRange: {StartingSequenceNumber: 49627894338503151834508157912666084957565273949901684850,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 0
11:59:21,531 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 4 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000005,HashKeyRange: {StartingHashKey: 106338239662793269832304564822427566080,EndingHashKey: 127605887595351923798765477786913079295},SequenceNumberRange: {StartingSequenceNumber: 49627894338458550344111096666383013521019977226889723986,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,532 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 4 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000005,HashKeyRange: {StartingHashKey: 106338239662793269832304564822427566080,EndingHashKey: 127605887595351923798765477786913079295},SequenceNumberRange: {StartingSequenceNumber: 49627894338458550344111096666383013521019977226889723986,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 0
11:59:21,532 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 3 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000004,HashKeyRange: {StartingHashKey: 85070591730234615865843651857942052864,EndingHashKey: 106338239662793269832304564822427566079},SequenceNumberRange: {StartingSequenceNumber: 49627894338436249598912566043241477802747328865383743554,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,532 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 2 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000003,HashKeyRange: {StartingHashKey: 63802943797675961899382738893456539648,EndingHashKey: 85070591730234615865843651857942052863},SequenceNumberRange: {StartingSequenceNumber: 49627894338413948853714035420099942084474680503877763122,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,532 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 3 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000015,HashKeyRange: {StartingHashKey: 319014718988379809496913694467282698240,EndingHashKey: 340282366920938463463374607431768211455},SequenceNumberRange: {StartingSequenceNumber: 49627894338681557796096402897798370703746460841949528306,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,532 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 2 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000014,HashKeyRange: {StartingHashKey: 297747071055821155530452781502797185024,EndingHashKey: 319014718988379809496913694467282698239},SequenceNumberRange: {StartingSequenceNumber: 49627894338659257050897872274656834985473812480443547874,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,532 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 3 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000004,HashKeyRange: {StartingHashKey: 85070591730234615865843651857942052864,EndingHashKey: 106338239662793269832304564822427566079},SequenceNumberRange: {StartingSequenceNumber: 49627894338436249598912566043241477802747328865383743554,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 0
11:59:21,532 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 2 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000003,HashKeyRange: {StartingHashKey: 63802943797675961899382738893456539648,EndingHashKey: 85070591730234615865843651857942052863},SequenceNumberRange: {StartingSequenceNumber: 49627894338413948853714035420099942084474680503877763122,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 0
11:59:21,532 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 0 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000001,HashKeyRange: {StartingHashKey: 21267647932558653966460912964485513216,EndingHashKey: 42535295865117307932921825928971026431},SequenceNumberRange: {StartingSequenceNumber: 49627894338369347363316974173816870647929383780865802258,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,532 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 0 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000009,HashKeyRange: {StartingHashKey: 191408831393027885698148216680369618944,EndingHashKey: 212676479325586539664609129644855132159},SequenceNumberRange: {StartingSequenceNumber: 49627894338547753324905219158949156394110570672913645714,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,532 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 7 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000000,HashKeyRange: {StartingHashKey: 0,EndingHashKey: 21267647932558653966460912964485513215},SequenceNumberRange: {StartingSequenceNumber: 49627894338347046618118443550675334929656735419359821826,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,533 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 0 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000012,HashKeyRange: {StartingHashKey: 255211775190703847597530955573826158592,EndingHashKey: 276479423123262501563991868538311671807},SequenceNumberRange: {StartingSequenceNumber: 49627894338614655560500811028373763548928515757431587010,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,533 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 7 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000008,HashKeyRange: {StartingHashKey: 170141183460469231731687303715884105728,EndingHashKey: 191408831393027885698148216680369618943},SequenceNumberRange: {StartingSequenceNumber: 49627894338525452579706688535807620675837922311407665282,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,533 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 0 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000001,HashKeyRange: {StartingHashKey: 21267647932558653966460912964485513216,EndingHashKey: 42535295865117307932921825928971026431},SequenceNumberRange: {StartingSequenceNumber: 49627894338369347363316974173816870647929383780865802258,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 0
11:59:21,533 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 7 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000011,HashKeyRange: {StartingHashKey: 233944127258145193631070042609340645376,EndingHashKey: 255211775190703847597530955573826158591},SequenceNumberRange: {StartingSequenceNumber: 49627894338592354815302280405232227830655867395925606578,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,533 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 7 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000000,HashKeyRange: {StartingHashKey: 0,EndingHashKey: 21267647932558653966460912964485513215},SequenceNumberRange: {StartingSequenceNumber: 49627894338347046618118443550675334929656735419359821826,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 0
11:59:21,568 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 1 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000002,HashKeyRange: {StartingHashKey: 42535295865117307932921825928971026432,EndingHashKey: 63802943797675961899382738893456539647},SequenceNumberRange: {StartingSequenceNumber: 49627894338391648108515504796958406366202032142371782690,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,568 INFO  org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer [] - Subtask 1 will be seeded with initial shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000013,HashKeyRange: {StartingHashKey: 276479423123262501563991868538311671808,EndingHashKey: 297747071055821155530452781502797185023},SequenceNumberRange: {StartingSequenceNumber: 49627894338636956305699341651515299267201164118937567442,}}'}, starting state set as sequence number EARLIEST_SEQUENCE_NUM
11:59:21,568 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 1 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000002,HashKeyRange: {StartingHashKey: 42535295865117307932921825928971026432,EndingHashKey: 63802943797675961899382738893456539647},SequenceNumberRange: {StartingSequenceNumber: 49627894338391648108515504796958406366202032142371782690,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 0
11:59:23,209 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 0 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000009,HashKeyRange: {StartingHashKey: 191408831393027885698148216680369618944,EndingHashKey: 212676479325586539664609129644855132159},SequenceNumberRange: {StartingSequenceNumber: 49627894338547753324905219158949156394110570672913645714,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 1
11:59:23,244 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 6 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000010,HashKeyRange: {StartingHashKey: 212676479325586539664609129644855132160,EndingHashKey: 233944127258145193631070042609340645375},SequenceNumberRange: {StartingSequenceNumber: 49627894338570054070103749782090692112383219034419626146,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 1
event: 6; timestamp: 1648030822428, 2022-03-23T10:20:22.428Z
11:59:23,377 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 3 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000015,HashKeyRange: {StartingHashKey: 319014718988379809496913694467282698240,EndingHashKey: 340282366920938463463374607431768211455},SequenceNumberRange: {StartingSequenceNumber: 49627894338681557796096402897798370703746460841949528306,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 1
11:59:23,405 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 2 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000014,HashKeyRange: {StartingHashKey: 297747071055821155530452781502797185024,EndingHashKey: 319014718988379809496913694467282698239},SequenceNumberRange: {StartingSequenceNumber: 49627894338659257050897872274656834985473812480443547874,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 1
11:59:23,581 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 7 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000008,HashKeyRange: {StartingHashKey: 170141183460469231731687303715884105728,EndingHashKey: 191408831393027885698148216680369618943},SequenceNumberRange: {StartingSequenceNumber: 49627894338525452579706688535807620675837922311407665282,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 1
11:59:23,586 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 1 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000013,HashKeyRange: {StartingHashKey: 276479423123262501563991868538311671808,EndingHashKey: 297747071055821155530452781502797185023},SequenceNumberRange: {StartingSequenceNumber: 49627894338636956305699341651515299267201164118937567442,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 1
11:59:24,790 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 0 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000012,HashKeyRange: {StartingHashKey: 255211775190703847597530955573826158592,EndingHashKey: 276479423123262501563991868538311671807},SequenceNumberRange: {StartingSequenceNumber: 49627894338614655560500811028373763548928515757431587010,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 2
event: 4; timestamp: 1648030809282, 2022-03-23T10:20:09.282Z
event: 3; timestamp: 1648030797697, 2022-03-23T10:19:57.697Z
event: 5; timestamp: 1648030810871, 2022-03-23T10:20:10.871Z
11:59:24,907 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 7 will start consuming seeded shard StreamShardHandle{streamName='hp-16', shard='{ShardId: shardId-000000000011,HashKeyRange: {StartingHashKey: 233944127258145193631070042609340645376,EndingHashKey: 255211775190703847597530955573826158591},SequenceNumberRange: {StartingSequenceNumber: 49627894338592354815302280405232227830655867395925606578,}}'} from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 2
event: 7; timestamp: 1648030834105, 2022-03-23T10:20:34.105Z
event: 1; timestamp: 1648030794441, 2022-03-23T10:19:54.441Z
event: 2; timestamp: 1648030796122, 2022-03-23T10:19:56.122Z
event: 8; timestamp: 1648030887171, 2022-03-23T10:21:27.171Z
XXXXXXXXXXXXXX Window with 3 events; Watermark: 1648030809281, 2022-03-23T10:20:09.281Z
3
1
2
XXXXXXXXXXXXXX Window with 2 events; Watermark: 1648030834104, 2022-03-23T10:20:34.104Z
4
5
XXXXXXXXXXXXXX Window with 1 events; Watermark: 1648030834104, 2022-03-23T10:20:34.104Z
6
XXXXXXXXXXXXXX Window with 1 events; Watermark: 1648030887170, 2022-03-23T10:21:27.170Z
7

Die Ausgabe zeigt nur 4 Fenster (es fehlt das letzte Fenster, das Ereignis 8 enthält). Das liegt an der Ereigniszeit und der Wasserzeichen-Strategie. Das letzte Fenster kann nicht geschlossen werden, da die vordefinierten Wasserzeichen-Strategien dafür sorgen, dass die Zeit nie über den Zeitpunkt des letzten Ereignisses hinausgeht, das aus dem Stream gelesen wurde. Damit das Fenster geschlossen werden kann, muss die Zeit jedoch um mehr als 10 Sekunden über das letzte Ereignis hinausgehen. In diesem Fall ist das letzte Wasserzeichen der 23.03.2022T10:21:27.170Z, aber damit das Sitzungsfenster geschlossen werden kann, ist ein Wasserzeichen 10 Sekunden und 1 ms später erforderlich.

Wenn die withIdleness Option aus der Wasserzeichenstrategie entfernt wird, wird kein Sitzungsfenster jemals geschlossen, da das „globale Wasserzeichen“ des Fensterbetreibers nicht weiterkommen kann.

Wenn die Flink-Anwendung gestartet wird (oder wenn die Daten verzerrt sind), werden einige Shards möglicherweise schneller verbraucht als andere. Dies kann dazu führen, dass einige Wasserzeichen zu früh von einer Unteraufgabe ausgegeben werden (die Unteraufgabe könnte das Wasserzeichen aufgrund des Inhalts eines Shards ausgeben, ohne dass es von den anderen Shards, die sie abonniert hat, verbraucht hat). Um dem entgegenzuwirken, gibt es verschiedene Wasserzeichen-Strategien, die einen Sicherheitspuffer hinzufügen oder Ereignisse, die spät eintreffen, explizit zulassen. (forBoundedOutOfOrderness(Duration.ofSeconds(30)) (allowedLateness(Time.minutes(5))

Festlegen einer UUID für alle Operatoren

Wenn Managed Service für Apache Flink einen Flink-Auftrag für eine Anwendung mit einem Snapshot startet, kann der Flink-Auftrag aufgrund bestimmter Probleme nicht gestartet werden. Eines davon ist die Nichtübereinstimmung der Operator-ID. Flink erwartet explizite, konsistente Operator-IDs für Flink-Auftragsdiagramm-Operatoren. Wenn nicht explizit gesetzt, generiert Flink eine ID für die Operatoren. Das liegt daran, dass Flink diese Operator-IDs verwendet, um die Operatoren in einem Auftragsdiagramm eindeutig zu identifizieren, und sie verwendet, um den Zustand jedes Operators in einem Savepoint zu speichern.

Das Problem der Nichtübereinstimmung der Operator-ID tritt auf, wenn Flink keine 1:1-Zuordnung zwischen den Operator-IDs eines Auftragsdiagramms und den in einem Savepoint definierten Operator-IDs findet. Dies passiert, wenn keine expliziten konsistenten Operator-IDs gesetzt sind und Flink Operator-IDs generiert, die möglicherweise nicht bei jeder Erstellung eines Jobdiagramms konsistent sind. Die Wahrscheinlichkeit, dass Anwendungen bei Wartungsarbeiten auf dieses Problem stoßen, ist hoch. Um dies zu vermeiden, empfehlen wir Kunden, die UUID für alle Operatoren im Flink-Code festzulegen. Weitere Informationen finden Sie im Abschnitt Festlegen einer UUID für alle Operatoren unter Produktionsbereitschaft.

Zum Maven ServiceResourceTransformer Shade-Plugin hinzufügen

Flink verwendet die Service Provider Interfaces (SPI) von Java, um Komponenten wie Konnektoren und Formate zu laden. Mehrere Flink-Abhängigkeiten, die SPI verwenden, können zu Konflikten im Uber-Jar und zu unerwartetem Anwendungsverhalten führen. Wir empfehlen, dass Sie das Maven Shade-Plugin hinzufügen, das in ServiceResourceTransformer der Datei pom.xml definiert ist.

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
                <executions>
                    <execution>
                        <id>shade</id>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <transformers combine.children="append">
                                <!-- The service transformer is needed to merge META-INF/services files -->
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                                <!-- ... -->
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
        </plugin>