[JAVA] Entwicklung von Flink mit der DataStream-API

In diesem Artikel werden die Grundlagen der verteilten Stream-Verarbeitung erläutert und die Entwicklung von Flink und DataStream ** API ** als Beispiel erläutert.

• Dieser Blog ist eine Übersetzung aus der englischen Version. Sie können das Original von [hier] überprüfen (https://www.alibabacloud.com/blog/basic-apache-flink-tutorial-datastream-api-programming_595685). Einige maschinelle Übersetzungen werden verwendet. Wir würden uns freuen, wenn Sie auf Übersetzungsfehler hinweisen könnten. * *

Grundkonzept der Stream-Verarbeitung

Die Definition der Stream-Verarbeitung kann unterschiedlich sein. Konzeptionell sind Stream-Verarbeitung und Batch-Verarbeitung zwei Seiten derselben Medaille. Ihre Beziehung hängt davon ab, ob die ArrayList- und Java-Elemente direkt als begrenztes Dataset betrachtet werden und auf die Indizes oder Iteratoren zugreifen.

Abbildung 1. Die linke Seite ist ein Münzsortierer

Der Münzklassifizierer kann als Stromverarbeitungssystem beschrieben werden. Im Voraus werden alle zur Klassifizierung von Münzen verwendeten Komponenten in Reihe geschaltet. Münzen gelangen kontinuierlich in das System und werden zur zukünftigen Verwendung in eine andere Warteschlange ausgegeben. Gleiches gilt für das Foto rechts.

Stream-Verarbeitungssysteme weisen viele Eigenschaften auf. Stream-Verarbeitungssysteme verwenden typischerweise ein datengesteuertes Verarbeitungsschema, um die Verarbeitung einer unendlichen Anzahl von Datensätzen zu unterstützen. Stellen Sie den Bediener im Voraus ein und verarbeiten Sie die Daten. Um eine komplexe Rechenlogik darzustellen, verwenden verteilte Stream-Verarbeitungs-Engines, einschließlich Flink, normalerweise DAG-Diagramme, um die gesamte Rechenlogik darzustellen.

Jeder Punkt in der DAG repräsentiert den Operator, der die grundlegende logische Einheit darstellt. Organisieren Sie Ihre Rechenlogik in gerichteten Diagrammen, sodass Daten von speziellen Quellknoten von den Kanten in das System fließen. Daten werden zwischen Betreibern über verschiedene Datenübertragungsmethoden wie Netzwerkübertragung und lokale Übertragung übertragen und verarbeitet. Schließlich werden die Ergebnisse der Daten über andere spezialisierte Synchronisationsknoten an ein externes System oder eine externe Datenbank gesendet.

Abbildung 2. Diagramm der DAG-Rechenlogik und physisches Laufzeitmodell.

Jeder Operator im logischen Diagramm verfügt über mehrere gleichzeitige Threads im physischen Diagramm. Für eine verteilte Stream-Verarbeitungs-Engine ist das tatsächliche physikalische Laufzeitmodell komplizierter, da jeder Operator mehrere Instanzen haben kann. Wie in 2 gezeigt, hat der Quelloperator A zwei Instanzen, und der Zwischenoperator C hat auch zwei Instanzen.

Im logischen Modell sind A und B vorgelagerte Knoten von C, und im entsprechenden physikalischen Modell kann ein Datenaustausch zwischen allen Instanzen von C, A, B stattfinden.

Bei der Verteilung von Operatorinstanzen auf verschiedene Prozesse werden Daten über das Netzwerk übertragen. Das Übertragen von Daten zwischen mehreren Instanzen im selben Prozess muss normalerweise nicht über das Netzwerk erfolgen.

Tabelle 1 Mit Apache Storm erstelltes DAG-Berechnungsdiagramm. Die API-Definition von Apache Storm ist "betriebsorientiert" und daher auf niedriger Ebene.

TopologyBuilder builder = new TopologyBuilder();

builder.setSpout("spout", new RandomSentenceSpout(), 5);
builder.setBolt("split", new SplitSentence(), 8).shuffleGrouping("spout");
builder.setBolt("count", new WordCount(), 12).fieldsGrouping("split", new Fields("word"));

Tabelle 2 Mit Apache Flink erstelltes DAG-Berechnungsdiagramm. Die API-Definitionen von Apache Flink sind eher "datenorientiert" und befinden sich daher auf einer höheren Ebene.

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<String> text = env.readTextFile ("input");
DataStream<Tuple2<String, Integer>> counts = text.flatMap(new Tokenizer()).keyBy(0).sum(1);
counts.writeAsText("output");

Da der DAG-Graph die Rechenlogik der Stream-Verarbeitung darstellt, basieren die meisten APIs auf der Konstruktion dieses Rechenlogik-Graphen. Tabelle 1 zeigt ein Beispiel für Apache Storms WordCount, das vor einigen Jahren populär war.

Apache Storm fügt dem Diagramm Spout- und Bolt-Operatoren hinzu, um anzugeben, wie die Operatoren eine Verbindung herstellen. Senden Sie das gesamte Diagramm nach dem Erstellen zur Ausführung in einem Remote- oder lokalen Cluster.

Im Gegensatz dazu erstellt die Apache Flink-API auch Rechenlogikdiagramme, aber die API-Definitionen von Flink sind eher auf Datenverarbeitungslogik ausgerichtet. Flink abstrahiert den Datenstrom in eine unendliche Menge, definiert eine Gruppe von Operationen für diese Menge und erstellt automatisch das entsprechende DAG-Diagramm auf der unteren Ebene.

Infolgedessen befindet sich die Flink-API auf einer höheren Ebene. Viele Forscher bevorzugen möglicherweise die hohe Flexibilität von Storm in ihren Experimenten, da dies die Sicherung der erwarteten Diagrammstruktur erleichtert. Die gesamte Branche bevorzugt jedoch erweiterte APIs wie die Flink-API, da diese einfacher zu verwenden sind.

Übersicht über die Flink DataStream-API

Wir werden detailliert erklären, wie die Flink DataStream-API basierend auf dem bisherigen Grundkonzept der Stream-Verarbeitung verwendet wird. Beginnen wir mit einem einfachen Beispiel. Tabelle 3 ist ein Beispiel für das Streaming von WordCount. Es enthält nur 5 Codezeilen, bietet jedoch die Grundstruktur für die Entwicklung von Programmen, die auf der Flink DataStream-API basieren.

Tabelle 3 WordCount-Beispiele basierend auf der Flink DataStream-API

// 1. Set the runtime environment
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 2. Configure the data source to read data
DataStream<String> text = env.readTextFile ("input");
// 3. Perform a series of transformations
DataStream<Tuple2<String, Integer>> counts = text.flatMap(new Tokenizer()).keyBy(0).sum(1);
// 4. Configure the Sink to write data out
counts.writeAsText("output");
// 5. Submit for execution
env.execute("Streaming WordCount");

Um Streaming WordCount zu implementieren, rufen Sie zuerst das StreamExecutionEnvironment-Objekt ab. Dies ist das Kontextobjekt, das das Diagramm erstellt. Fügen Sie einen Operator hinzu, der auf diesem Objekt basiert. Erstellen Sie für Stream-Verarbeitungsebenen eine Datenquelle, um auf die Daten zuzugreifen. In diesem Beispiel wird eine integrierte Datenquelle zum Lesen von Dateien verwendet, die sich im Umgebungsobjekt befinden.

Rufen Sie dann das DataStream-Objekt ab, bei dem es sich um ein unendliches Dataset handelt. Führen Sie eine Reihe von Vorgängen für diesen Datensatz aus. Im WordCount-Beispiel wird beispielsweise jeder Datensatz (dh eine Zeile in der Datei) zuerst in Wörter getrennt und durch die FlatMap-Operation implementiert.

Durch Aufrufen von FlatMap wird der Operator zum zugrunde liegenden DAG-Diagramm hinzugefügt. Um einen Wortstrom zu erhalten, gruppieren Sie dann die Wörter im Strom (KeyBy) und berechnen Sie die Daten für jedes Wort kumulativ (Summe (1)). Die berechneten Wortdaten bilden einen neuen Stream und werden in die Ausgabedatei geschrieben.

Rufen Sie abschließend die Methode env # execute auf, um die Ausführung des Programms zu starten. Stellen Sie sicher, dass keine der zuvor aufgerufenen Methoden Daten verarbeitet, und erstellen Sie ein DAG-Diagramm, das Ihre Rechenlogik darstellt.

Erstellen Sie das gesamte Diagramm und rufen Sie zuerst explizit die Execute-Methode auf. Das Framework stellt dem Cluster Rechengraphen zur Verfügung, um auf Daten zuzugreifen und Logik auszuführen.

Streaming WordCount-basierte Beispiele zeigen, dass das Kompilieren eines Stream-Prozessors basierend auf der Flink DataStream-API im Allgemeinen drei Schritte erfordert: Zugreifen auf, Verarbeiten und Schreiben von Daten.

Rufen Sie abschließend die Execute-Methode auf.

Abbildung 3. Übersicht über den Betrieb des Flink-Datenstroms.

Wie Sie dem vorherigen Beispiel entnehmen können, ist der Kern der Flink DataStream-API ein DataStream-Objekt, das Streaming-Daten darstellt. Der gesamte Rechenlogikgraph basiert auf dem Aufrufen verschiedener Operationen für das DataStream-Objekt, um ein neues DataStream-Objekt zu erstellen.

Im Allgemeinen gibt es in DataStream vier Arten von Vorgängen. Der erste Typ ist eine einzelne Datensatzoperation, die unerwünschte Datensätze filtert (Filteroperation) und jeden Datensatz konvertiert (Kartenoperation). Der zweite Typ ist eine Mehrfachaufzeichnungsoperation. Um beispielsweise das Gesamtauftragsvolumen innerhalb einer Stunde zu zählen, fügen Sie alle Auftragsdatensätze innerhalb einer Stunde hinzu. Um diese Art von Operation zu unterstützen, müssen Sie die erforderlichen Datensätze über ein Fenster zur Verarbeitung kombinieren.

Der dritte Typ besteht darin, mehrere Streams zu bearbeiten und in einen einzigen Stream zu konvertieren. Sie können beispielsweise mehrere Streams mit Vorgängen wie Union, Join und Connect zusammenführen. Diese Operationen verwenden unterschiedliche Logik, um die Streams zusammenzuführen, erzeugen jedoch schließlich einen neuen einheitlichen Stream, der einige stromübergreifende Operationen ermöglicht.

Der vierte Typ ist eine "Split-Operation", die von DataStream unterstützt wird und im Gegensatz zur Merge-Operation steht. Diese Vorgänge teilen den Stream gemäß den Regeln in mehrere Streams auf, und jeder geteilte Stream ist eine Teilmenge des vorherigen Streams.

Abbildung 4. Verschiedene Arten von DataStream-Untertypen. Unterschiedliche Untertypen unterstützen unterschiedliche Operationssätze.

Um verschiedene Stream-Vorgänge zu unterstützen, führt Flink Sätze verschiedener Stream-Typen ein, um den Typ des Zwischen-Stream-Datasets anzugeben. Abbildung 4 zeigt den vollständigen Typ der Transformationsbeziehung.

Bei einzelnen Datensatzoperationen wie Map ist das Ergebnis vom Typ DataStream. Die Split-Operation erzeugt einen SplitStream. Verwenden Sie basierend auf SplitStream die Select-Methode, um die gewünschten Datensätze zu filtern und den Basis-Stream abzurufen.

In ähnlicher Weise erhält der Connect-Vorgang nach dem Aufrufen von StreamA.connect (StreamB) einen dedizierten ConnectedStream. Die von ConnectedStream unterstützten Vorgänge unterscheiden sich von den von Common DataStream unterstützten Vorgängen.

Dies ist das Ergebnis des Zusammenführens von zwei verschiedenen Streams, sodass Sie unterschiedliche Verarbeitungslogiken für die Datensätze in den beiden Streams angeben können. Die verarbeiteten Ergebnisse bilden einen neuen DataStream-Stream. Verarbeiten Sie verschiedene Datensätze mit demselben Operator, um Statusinformationen während der Verarbeitung auszutauschen. Einige Join-Vorgänge der oberen Ebene müssen über Connect-Vorgänge der unteren Ebene implementiert werden.

Sie können den Stream auch durch Zeit oder Zahl teilen, indem Sie das Fenster bedienen. Wählen Sie eine bestimmte Split-Logik. Wenn alle Datensätze in der Gruppe eintreffen, rufen Sie alle Datensätze ab und führen Sie Traverse- und Summenoperationen durch. Daher erhält die Verarbeitung jeder Gruppe einen Satz von Ausgabedaten, und alle Ausgabedaten bilden einen neuen Basisstrom.

Verwenden Sie für einen gemeinsamen DataStream die Operation allWindow, die einen einheitlichen Fensterungsprozess für den gesamten Stream darstellt. Daher ist es nicht möglich, simultane Berechnungen mit mehreren Operatorinstanzen durchzuführen. Um dieses Problem zu beheben, gruppieren Sie zuerst die Datensätze nach Schlüssel mithilfe der KeyBy-Methode. Danach werden für die Datensätze, die verschiedenen Schlüsseln entsprechen, separate Fensterprozesse parallel ausgeführt.

Die KeyBy-Operation ist eine der wichtigsten und am häufigsten verwendeten Operationen. Es wird nachstehend ausführlich erläutert.

Abbildung 5. Vergleich der Fensterbedienung von Basic Stream und KeyedStream

Fensteroperationen in KeyedStream ermöglichen die gleichzeitige Verarbeitung mit mehreren Instanzen. Abbildung 5 zeigt einen Vergleich aller Windows-Operationen für ein grundlegendes DataStream-Objekt und von Windows-Operationen für ein KeyedStream-Objekt. Verwenden Sie die KeyBy-Operation, um Daten zu gruppieren, um Daten gleichzeitig in mehreren Instanzen zu verarbeiten.

Sowohl KeyBy- als auch Window-Vorgänge gruppieren Daten, aber KeyBy-Vorgänge teilen den Stream horizontal und Window-Vorgänge teilen den Stream vertikal auf.

Nach dem Aufteilen der Daten mit KeyBy kann jede nachfolgende Operatorinstanz die Daten verarbeiten, die einem bestimmten Schlüsselsatz entsprechen. Außerdem ermöglicht Flink den Bedienern, einen bestimmten Status beizubehalten. Der Status der Operatoren im KeyedStream wird verteilt gespeichert.

KeyBy ist eine deterministische Datenzuweisungsmethode (im nächsten Abschnitt werden andere Zuweisungsmethoden vorgestellt). Wenn ein fehlgeschlagener Job neu gestartet wird und sich die Parallelität ändert, weist Flink die Schlüsselgruppe neu zu, um sicherzustellen, dass die Gruppe, die einen bestimmten Schlüssel verarbeitet, den Status dieses Schlüssels enthalten muss und konsistent ist. Stellen Sie Sex sicher.

Beachten Sie schließlich, dass die KeyBy-Operation nur funktioniert, wenn die Anzahl der Schlüssel die Anzahl der gleichzeitigen Instanzen des Operators überschreitet. Alle Daten, die demselben Schlüssel entsprechen, werden an dieselbe Instanz gesendet. Wenn also die Anzahl der Schlüssel geringer als die Anzahl der Instanzen ist, können einige Instanzen die Daten nicht empfangen und die Rechenleistung wird nicht vollständig genutzt. Werden.

Andere Probleme

Flink unterstützt andere physische Gruppierungsmethoden als KeyBy beim Datenaustausch zwischen Operatoren. Wie in Abbildung 1 dargestellt, umfassen die physischen Gruppierungsmethoden in Flink Data Stream:

--Global: Der Upstream-Operator sendet alle Datensätze an die erste Instanz des Downstream-Operators.

• Broadcast: Der Upstream-Operator sendet jeden Datensatz an alle Instanzen des Downstream-Operators. --Forward: Der Upstream-Operator sendet den Datensatz an alle Instanzen des Downstream-Operators. Jede Upstream-Operator-Instanz sendet einen Datensatz an die entsprechende Instanz des Downstream-Operators. Diese Methode gilt nur, wenn die Anzahl der Instanzen des Upstream-Operators mit der Anzahl der Instanzen des Downstream-Operators übereinstimmt.

--Shuffle: Der Upstream-Operator wählt zufällig den Downstream-Operator für jeden Datensatz aus.

• Ausgleich: Upstream-Betreiber senden Daten auf Round-Robin-Basis. --Rescale: Wenn die Anzahl der Instanzen der Upstream- und Downstream-Operatoren 'n'und'm' und 'n' <'m' ist, ist jede Upstream-Instanz auf Round-Robin-Basis Ceil (m / n). Oder senden Sie Daten an eine Instanz nach dem Stockwerk (m / n). Wenn n '>' m 'ist, sendet die Upstream-Instanz von Floor (n / m) oder Ceil (n / m) Daten auf Round-Robin-Basis an die Downstream-Instanz. -PartitionCustomer: Wenn die integrierte Zuordnungsmethode nicht Ihren Anforderungen entspricht, können Sie die Gruppierungsmethode anpassen.

Abbildung 6. Andere physikalische Gruppierungsmethoden als KeyBy

Ein weiteres wichtiges Konzept in der Flink DataStream-API ist neben den Gruppierungsmethoden der Systemtyp.

Wie in Abbildung 7 dargestellt, verfügt das Flink DataStream-Objekt über eine starke Systemtypeinstellung. Sie müssen den Elementtyp für jedes DataStream-Objekt angeben. Der zugrunde liegende Serialisierungsmechanismus von Flink stützt sich auf diese Informationen, um die Serialisierung zu optimieren. Insbesondere verwendet die unterste Ebene von Flink ein TypeInformation-Objekt, um den Typ zu beschreiben. Das TypeInformation-Objekt definiert eine Zeichenfolge typbezogener Informationen, die vom Serialisierungsframework verwendet werden.

Abbildung 7. Flink DataStream API-Typsystem

Flink verfügt über einige häufig verwendete integrierte Basistypen. Für diese stellt Flink auch seine Typinformationen zur Verfügung und kann direkt ohne zusätzliche Deklarationen verwendet werden. Flink kann einen Typinferenzmechanismus verwenden, um den entsprechenden Typ zu identifizieren. Es gibt jedoch Ausnahmen.

Beispielsweise unterstützt die Flink DataStream-API sowohl Java als auch Scala. Viele Scala-APIs übergeben Typinformationen über implizite Parameter. Wenn Sie also die Scala-API über Java aufrufen müssen, müssen Sie Typinformationen über implizite Parameter übergeben. Ein weiteres Beispiel ist das Java-basierte Löschen generischer Typen. Wenn der Stream-Typ ein generischer Typ ist, ist es möglicherweise nicht erforderlich, den Informationstyp nach dem Löschen abzuleiten. In diesem Fall muss auch die Art der Informationen explizit angegeben werden.

In Flink verwenden Java-APIs normalerweise den Tupel-Typ, wenn mehrere Felder verknüpft werden, während Scala-APIs häufig die Zeilen- und Fallklassentypen verwenden. Im Vergleich zum Zeilentyp unterliegt der Taple-Typ zwei Einschränkungen: Die Anzahl der Felder darf 25 nicht überschreiten, und der NULL-Wert kann nicht in allen Feldern verwendet werden.

Schließlich können Sie mit Flink neue Typen, TypeInformation, anpassen und mit Kryo serialisieren. Dies kann jedoch zu Migrationsproblemen führen. Wir empfehlen daher, benutzerdefinierte Typen zu vermeiden.

Beispiel

Schauen wir uns ein etwas komplizierteres Beispiel an. Angenommen, Sie haben eine Datenquelle in Ihrem System, die Ihre Bestellungen überwacht. Es verwendet Tuple2, um den Typ und das Volumen des bestellten Produkts zu drucken, wenn Sie eine neue Bestellung aufgeben. Anschließend wird das Transaktionsvolumen aller Arten von Artikeln in Echtzeit gezählt.

Tabelle 4 Ein Beispiel für Echtzeit-Auftragsstatistiken.

public class GroupedProcessingTimeWindowSample {
    private static class DataSource extends RichParallelSourceFunction<Tuple2<String, Integer>> {
        private volatile boolean isRunning = true;

        @Override
        public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
            Random random = new Random();
            while (isRunning) {
                Thread.sleep((getRuntimeContext().getIndexOfThisSubtask() + 1) * 1000 * 5);
                String key = "Einstufung" + (char) ('A' + random.nextInt(3));
                int value = random.nextInt(10) + 1;

                System.out.println(String.format("Emits\t(%s, %d)", key, value));
                ctx.collect(new Tuple2<>(key, value));
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStream<Tuple2<String, Integer>> ds = env.addSource(new DataSource());
        KeyedStream<Tuple2<String, Integer>, Tuple> keyedStream = ds.keyBy(0);

        keyedStream.sum(1).keyBy(new KeySelector<Tuple2<String, Integer>, Object>() {
            @Override
            public Object getKey(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                return "";
            }
        }).fold(new HashMap<String, Integer>(), new FoldFunction<Tuple2<String, Integer>, HashMap<String, Integer>>() {
            @Override
            public HashMap<String, Integer> fold(HashMap<String, Integer> accumulator, Tuple2<String, Integer> value) throws Exception {
                accumulator.put(value.f0, value.f1);
                return accumulator;
            }
        }).addSink(new SinkFunction<HashMap<String, Integer>>() {
            @Override
            public void invoke(HashMap<String, Integer> value, Context context) throws Exception {
                  //Individuelles Produktkommutierungsvolumen
                  System.out.println(value);
                  //Rohstoffbetrag
                  System.out.println(value.values().stream().mapToInt(v -> v).sum());
            }
        });

        env.execute();
    }
}

Tabelle 4 zeigt die Implementierung dieses Beispiels. Hier implementieren wir eine Scheindatenquelle, die von RichParallelSourceFunction erbt. RichParallelSourceFunction ist eine SourceFunction-API mit mehreren Instanzen.

Implementieren Sie zwei Methoden, die Run-Methode und die Cancel-Methode. Flink ruft die Run-Methode zur Laufzeit direkt an die Quelle auf. Daten müssen kontinuierlich ausgegeben werden, um den anfänglichen Stream zu bilden. Generieren Sie bei der Implementierung der Run-Methode nach dem Zufallsprinzip Datensätze für Artikeltyp und Transaktionsvolumen und senden Sie sie mit der Methode "ctx # collect". Wenn Sie die Quellaufgabe abbrechen müssen, die von der flüchtigen Variablen für Flink verwendet wird, um ihren Ausführungsstatus zu markieren und zu steuern, verwenden Sie die Cancel-Methode.

Beginnen Sie dann mit der Erstellung des Diagramms mit der Main-Methode. Erstellen Sie zunächst ein StreamExecutionEnviroment-Objekt. Die Methode getExecutionEnvironment, die zum Erstellen des Objekts aufgerufen wird, bestimmt automatisch die Umgebung, sodass das entsprechende Objekt erstellt wird. Wenn Sie beispielsweise mit der rechten Maustaste in die IDE klicken und die Methode ausführen, wird ein LocalStreamExecutionEnvironment-Objekt erstellt.

Bei der Ausführung in einer realen Umgebung wird ein RemoteStreamExecutionEnvironment-Objekt erstellt. Erstellen Sie eine Quelle, um den anfänglichen Stream basierend auf dem Umgebungsobjekt abzurufen. Um dann den Transaktionsbetrag für jeden Artikeltyp zu zählen, wird KeyBy verwendet, um den Eingabestream durch das erste Feld (Artikeltyp) von Tuple und das zweite Feld (Transaktionsbetrag) des Datensatzes zu gruppieren, die jedem Schlüssel entsprechen. ) Ist zusammengefasst.

In der untersten Ebene verwendet der Summenoperator die State-Methode, um den Gesamtwert der Transaktionsvolumina für jeden Schlüssel (Elementtyp) zu speichern. Wenn ein neuer Datensatz eintrifft, aktualisiert der Summenoperator das Gesamtvolumen und verwaltet den .NET-Datensatz.

Wenn Sie nur Datenträger zählen möchten, endet das Programm hier. Fügen Sie den Sink-Operator unmittelbar nach dem Summen-Operator hinzu, um ein kontinuierlich aktualisiertes Transaktionsvolumen für jeden Artikeltyp zu drucken. Um jedoch die Anzahl der Transaktionen aller Art zu zählen, geben Sie alle Datensätze desselben Rechenknotens aus.

Ich verwende KeyBy, um denselben Schlüssel für alle Datensätze zurückzugeben, sie zu gruppieren und alle Datensätze an dieselbe Instanz zu senden.

Verwenden Sie dann die Fold-Methode, um das Volumen für jeden Elementtyp im Operator zu verwalten. Beachten Sie, dass die Fold-Methode als veraltet markiert ist, heute jedoch nicht durch andere Vorgänge in der DataStream-API ersetzt werden kann. Daher erhält diese Methode einen Anfangswert.

Wenn dann jeder Datensatz im nachfolgenden Stream eintrifft, ruft der Bediener die übergebene FoldFunction auf, um den Anfangswert zu aktualisieren, und sendet den aktualisierten Wert.

Verwenden Sie HashMap, um das aktuelle Transaktionsvolumen für jeden Artikeltyp zu verfolgen. Aktualisieren Sie die HashMap, wenn neue eintreffen. Auf diese Weise wird HashMap des neuesten Artikeltyps und Transaktionsvolumens über Sink empfangen, und das gesamte Transaktionsvolumen und Transaktionsvolumen jedes Artikels wird basierend auf diesem Wert ausgegeben.

Dieses Beispiel zeigt, wie die DataStream-API verwendet wird. Sie können effizienter schreiben. Die oberen Tabellen und SQLs unterstützen auch einen Rückzugsmechanismus, der diese Situation besser handhabt.

Abbildung 8 Schematische Darstellung der API.

Lassen Sie uns abschließend einen Blick auf die Prinzipien der DataStream-API werfen. Wenn Sie den DataStream # -Kartenalgorithmus aufrufen, erstellt Flink ein Transformationsobjekt auf der untersten Ebene. Dieses Objekt repräsentiert einen Knoten im Computational Logic Graph. Es zeichnet eine benutzerdefinierte Funktion (UDF), MapFunction, auf.

Erstellen Sie mehr DataStream-Objekte mit mehr Methoden. Jedes Objekt verfügt über ein Transformationsobjekt, das auf der Grundlage von Rechenabhängigkeiten eine Diagrammstruktur bildet.

Dies ist ein Berechnungsdiagramm. Flink transformiert dann die Diagrammstruktur weiter, um schließlich das Jobdiagramm zu generieren, das zum Senden des Jobs erforderlich ist.

Überblick

In diesem Artikel wird die Flink DataStream-API vorgestellt, eine untergeordnete API für Flink. In der tatsächlichen Entwicklung müssen Sie einige Konzepte selbst verwenden, die auf APIs wie Status und Zeit basieren, was problematisch ist. In den folgenden Kursen werden wir auch übergeordnete Tabellen und SQL-APIs vorstellen. In Zukunft werden Table und SQL möglicherweise zum Mainstream der Flink-API.

APIs auf niedrigerer Ebene sorgen jedoch für eine stärkere Ausdruckskraft. Die DataStream-API ist möglicherweise für feinkörnige Vorgänge erforderlich.

• Alibaba Cloud ist der Cloud-Infrastrukturbetreiber Nr. 1 (2019 Gartner) im asiatisch-pazifischen Raum mit zwei Rechenzentren in Japan und mehr als 60 Verfügbarkeitszonen weltweit. Klicken Sie hier, um weitere Informationen zu Alibaba Cloud zu erhalten. Offizielle Seite von Alibaba Cloud Japan *