Vorbereitungen
-
Sie haben zuvor eine Verbindung zu dem System erstellt, in dem die Quelldaten gespeichert sind.
-
Sie haben zuvor den Datensatz hinzugefügt, der die Quelldaten enthält.
In diesem Beispiel Streamingdaten zu Flugzeugen, d. h. Flugzeug-ID, Position und Zeitstempel.
Um dieses Szenario besser zu verstehen, finden Sie nachstehend das AVRO-Schema der im Szenario verwendeten Streamingdaten:
{
"type": "record",
"name": "aircrafts",
"fields": [
{"name": "Id", "type": "int"},
{"name": "PosTime", "type": "long"},
{"name": "Lat", "type": "double"},
{"name": "Long", "type": "double"},
{"name": "Op", "type": "string"}
]
}
Hierbei gilt: Id entspricht der Flugzeug-ID, PosTime dem Zeitstempel der Position, Lat/Long dem Breiten-/Ländengrad des Flugzeugs und Op den Fluggesellschaften.
-
Sie haben außerdem die Verbindung und den zugehörigen Datensatz erstellt, der die verarbeiteten Daten aufnehmen soll.
In diesem Beispiel eine MySQL-Tabelle.
Prozedur
-
Klicken Sie auf der Seite Pipelines auf Add pipeline (Pipeline hinzufügen). Ihre neue Pipeline wird geöffnet.
-
Legen Sie einen sinnvollen Namen für die Pipeline fest.
Example
Process streaming aircraft data (Streamingdaten zu Flugzeugen verarbeiten)
-
Klicken Sie auf ADD SOURCE (QUELLE HINZUFÜGEN), um ein Fenster zu öffnen, in dem Sie Ihre Quelldaten auswählen können, in diesem Beispiel das Flugzeug-Topic in Kafka.
-
Wählen Sie den Datensatz aus und klicken Sie auf Select (Auswählen), um ihn zur Pipeline hinzuzufügen.
Benennen Sie ihn um, falls erforderlich.
-
Klicken Sie auf
und fügen Sie einen Prozessor vom Typ Window (Fenster) zur Pipeline hinzu. Daraufhin wird das Konfigurationsfenster geöffnet.
-
Geben Sie einen sinnvollen Namen für den Prozessor an.
Example
5sec window (5-Sek.-Fenster)
-
Führen Sie im Bereich Configuration (Konfiguration) Folgendes durch:
-
Aktivieren Sie die Option Use Window session (Fenstersitzung verwenden).
-
Geben Sie den Wert 5000 als Fensterdauer ein, damit alle 5 Sekunden Daten erfasst werden.
-
Klicken Sie auf und fügen Sie einen Prozessor vom Typ Aggregate (Aggregieren) zur Pipeline hinzu. Daraufhin wird das Konfigurationsfenster geöffnet.
-
Geben Sie einen sinnvollen Namen für den Prozessor an.
Example
group by aircraft (Nach Flugzeug gruppieren)
-
Wählen Sie im Bereich Group by (Gruppieren nach) die Felder aus, die Sie für Ihre Aggregationsgruppe verwenden möchten: in diesem Beispiel .customerId (Kunden-ID).
-
Wählen Sie .Id in der Liste Field (Feld) aus, um eine Gruppierung nach der Flugzeug-ID durchzuführen.
-
Fügen Sie ein New element (Neues Element) hinzu und wählen Sie .Op (Betreiber) in der Liste Field (Feld) aus, um eine Gruppierung nach Fluggesellschaften durchzuführen.
-
Führen Sie im Bereich Operations (Operationen) Folgendes durch:
-
Wählen Sie .PosTime (.PositionUhrzeit) in der Liste Field (Feld) aus und Maximum in der Liste Operation aus.
-
Geben Sie dem generierten Feld (Output field (Ausgabefeld)) einen Namen, z. B. date (Datum).
-
Klicken Sie auf das Pluszeichen (+), um ein neues Element hinzuzufügen, wählen Sie .Lat (.Breitengrad) in der Liste Field path (Feldpfad) und List (Liste) in der Liste Operation aus.
-
Legen Sie für das generierte Feld einen Namen fest, beispielsweise lat (Breitengrad).
-
Klicken Sie auf das Pluszeichen (+), um ein neues Element hinzuzufügen, wählen Sie .Long (.Längengrad) in der Liste Field path (Feldpfad) und List (Liste) in der Liste Operation aus.
-
Legen Sie für das generierte Feld einen Namen fest, beispielsweise lon (Längengrad).
-
Klicken Sie auf Save (Speichern), um die Konfiguration zu speichern.
-
Klicken Sie auf und fügen Sie einen Prozessor vom Typ Field Selector (Feldauswahl) zur Pipeline hinzu. Daraufhin wird das Konfigurationsfenster geöffnet.
-
Geben Sie einen sinnvollen Namen für den Prozessor an.
Example
select latest position (Letzte Position auswählen)
-
Im Bereich Selectors (Selektoren) im Modus Advanced (Erweitert):
-
Geben Sie id (ID) in die Liste Output (Ausgabe) und .id (.ID) in die Liste Input (Eingabe) ein, da Sie das Feld
id (ID) auswählen und dieselbe Feldposition beibehalten möchten.
-
Klicken Sie auf das Pluszeichen (+), um ein neues Element hinzuzufügen, und geben Sie airlines (Fluggesellschaften) in der Liste Output (Ausgabe) und .Op (.Betreiber) in der Liste Input (Eingabe) ein, da Sie das Feld
Op (Betreiber) auswählen und umbenennen möchten.
-
Klicken Sie auf das Pluszeichen (+), um ein neues Element hinzuzufügen, und geben Sie (date) (Datum) in der Liste Output (Ausgabe) und .date (.Datum) in der Liste Input (Eingabe) ein, da Sie das Feld
date (Datum) auswählen und dieselbe Feldposition beibehalten möchten.
-
Klicken Sie auf das Pluszeichen (+), um ein neues Element hinzuzufügen, und geben Sie lat (Breitengrad) in der Liste Output (Ausgabe) und .lat[-1] (.Breitengrad[-1]) in die Liste Input (Eingabe) ein, da Sie das Feld
lat (Breitengrad) von der ursprünglichen Position auf eine niedrigere Ebene im Schema verschieben möchten.
-
Klicken Sie auf das Pluszeichen (+), um ein neues Element hinzuzufügen, und geben Sie lon (Längengrad) in der Liste Output (Ausgabe) und .lon[-1] (.Längengrad[-1]) in der Liste Input (Eingabe) ein, da Sie das Feld
lon (Längengrad) von der ursprünglichen Position auf eine niedrigere Ebene im Schema verschieben möchten.
In diesem Bereich können Sie die avpath-Syntax verwenden.
-
Klicken Sie auf Save (Speichern), um die Konfiguration zu speichern.
-
Klicken Sie auf und fügen Sie einen Prozessor vom Typ Python 3 zur Pipeline hinzu. Daraufhin wird das Konfigurationsfenster geöffnet.
-
Geben Sie einen sinnvollen Namen für den Prozessor an.
Example
compute geohash (Geohash verarbeiten)
-
Geben Sie im Bereich Phyton 3 code (Python 3-Code) Folgendes ein:
def encode(latitude, longitude, precision=12):
__base32 = '0123456789bcdefghjkmnpqrstuvwxyz'
__decodemap = { }
for i in range(len(__base32)):
__decodemap[__base32[i]] = i
del i
lat_interval, lon_interval = (-90.0, 90.0), (-180.0, 180.0)
geohash = []
bits = [ 16, 8, 4, 2, 1 ]
bit = 0
ch = 0
even = True
while len(geohash) < precision:
if even:
mid = (lon_interval[0] + lon_interval[1]) / 2
if longitude > mid:
ch |= bits[bit]
lon_interval = (mid, lon_interval[1])
else:
lon_interval = (lon_interval[0], mid)
else:
mid = (lat_interval[0] + lat_interval[1]) / 2
if latitude > mid:
ch |= bits[bit]
lat_interval = (mid, lat_interval[1])
else:
lat_interval = (lat_interval[0], mid)
even = not even
if bit < 4:
bit += 1
else:
geohash += __base32[ch]
bit = 0
ch = 0
return ''.join(geohash)
output = json.loads("{}")
output['id'] = input['id']
output['airlines'] = input['airlines']
output['date'] = input['date']
output['location'] = encode(input['lat'], input['lon'])
output.append(output)
Dieser Code ermöglicht Ihnen die Berechnung von Geohashes (eine Zeichenfolge, die die geografische Position ausgehend von Breiten- und Längengrad angibt).
-
Klicken Sie auf Save (Speichern), um die Konfiguration zu speichern.
-
Klicken Sie auf ADD DESTINATION (ZIEL HINZUFÜGEN) und wählen Sie den Datensatz aus, der die verarbeiteten Daten aufnehmen soll.
Benennen Sie ihn um, falls erforderlich.
-
(Optional) Sehen Sie sich die Vorschau des Prozessors vom Typ Python 3 an, um Ihre Daten zu prüfen.
-
Klicken Sie in der oberen Symbolleiste von Talend Cloud Pipeline Designer auf die Schaltfläche Run (Ausführen), um das Fenster zur Auswahl des Ausführungsprofils zu öffnen.
-
Wählen Sie Ihr Ausführungsprofil in der Liste aus (weitere Informationen finden Sie unter „Ausführungsprofile“) und klicken Sie dann auf Run (Ausführen), um die Pipeline auszuführen.
Ergebnisse
Ihre Streaming-Pipeline wird ausgeführt und bleibt aktiv, bis Sie sie explizit beenden. Die Flugzeugdaten werden geändert und die berechneten Geohash-Daten an das von Ihnen angegebene Zielsystem gesendet.
