Streamen von Daten

Der Eingliederungsprozess überträgt Daten von der Quelle und speichert sie in Iceberg-Tabellen. Änderungen aus den Streaming-Datenquellen werden kontinuierlich und nahezu in Echtzeit auf die Speichertabellen angewendet.

Onboarding für Daten

Daten werden innerhalb eines Pipeline-Projekts eingegliedert und Datensätze werden am S3-Speicherort gespeichert, der in den Projekteinstellungen definiert ist.

1. Klicken Sie im Projekt auf Erstellen und dann auf Daten eingliedern.

2. Fügen Sie einen Aufgabennamen und eine Beschreibung für die Eingliederung hinzu.

   Klicken Sie auf Weiter.

3. Wählen Sie die Quellverbindung aus.

   Sie können eine bestehende Streaming-Quellverbindung auswählen oder eine neue Verbindung zur Quelle erstellen.

   Weitere Informationen finden Sie unter Verbinden mit Daten-Streams.

   Klicken Sie auf Weiter und folgen Sie den Anweisungen unten für Ihre Datenquelle.

Auswählen von Daten

Apache Kafka

Die Liste zeigt die verfügbaren Kafka-Themen auf dem Cluster, der in der Quellverbindung definiert ist.

Beim Auswählen Ihrer Themen können Sie bestimmte Datensätze auswählen. Sie können auch Auswahlregeln zum Ein- oder Ausschließen von Datensatzgruppen verwenden:

• Verwenden Sie % als Wildcard, um ein Auswahlkriterium für die Datensätze zu definieren.

Wenn Themen mithilfe von Auswahlregeln ausgewählt werden, können Sie wählen, ob alle Datensätze in dieselbe Zieltabelle geladen werden sollen oder ob für jedes Quellthema eine separate Zieltabelle erstellt werden soll:

• Standardmäßig wird der Name der Iceberg-Zieltabelle vom Themennamen abgeleitet und formatiert, um den Namenskonventionen zu entsprechen (z. B. Kleinbuchstaben, Leerzeichen entfernt, Bindestriche durch Unterstriche ersetzt). Unter Name des Zieldatensatzes definieren können Sie den Namen der Zieltabelle bearbeiten.

• Wenn Auswahlregeln verwendet werden, um mehrere Themen in eine einzige Tabelle zu laden, müssen Sie den Zielnamen angeben.

• Wenn Auswahlregeln verwendet werden und die Daten in separate Tabellen (ein Datensatz pro Thema) geladen werden, sind die Standard-Zielnamen die Themennamen. In dieser Phase können Sie die Namen im Assistenten nicht bearbeiten, dies kann jedoch später in der Bereitstellungsaufgabe erfolgen.

• Wenn eine Regel so konfiguriert ist, dass Themen für das Einlesen ausgewählt werden, werden alle neuen Themen, die die Regelkriterien erfüllen, ebenfalls bereitgestellt, wenn die Option Neues Thema  > Zu Ziel hinzufügen unter „Schemaentwicklung“ in den Einstellungen der Bereitstellungsaufgabe aktiviert ist.

Wählen Sie einen oder mehrere Datensätze aus oder verwenden Sie eine Auswahlregel und klicken Sie auf Hinzufügen. Klicken Sie auf Weiter.

Amazon Kinesis

Die Liste zeigt die verfügbaren Kinesis-Streams an, die in der Quellverbindung definiert sind.

Wählen Sie einen oder mehrere Datensätze aus und klicken Sie auf Hinzufügen. Sie können die hinzugefügten Datensätze unter Ausgewählte Streams anzeigen. Klicken Sie auf Weiter.

Amazon S3

Der Verzeichnis-Browser zeigt eine Liste aller Verzeichnisse an, die sich im S3-Bucket Ihrer Quellverbindung befinden. 

• Wählen Sie die Verzeichnisse aus, die beim Bereitstellen von Daten eingeschlossen werden sollen:

  • Geben Sie für jedes Verzeichnis unter Pfad hinzufügen den Pfad und das Dateinamensmuster ein:

    • Verwenden Sie * als Wildcard, die beliebigen Zeichen entspricht.

    • Um ein Datumsmuster einzugeben, verwenden Sie <yyyy> als Platzhalter für das vierstellige Jahr, <MM> als Platzhalter für den zweistelligen Monat, <dd> als Platzhalter für den zweistelligen Tag und <HH> als Platzhalter für die zweistellige Stunde. Beispiel:

      • MyDir3/<yyyy>_<MM>_<dd>_<HH>_orders.csv

      • MyDir3/<yyyy>/<MM>/<dd>/<HH>_orders.csv

• Klicken Sie auf Vorschau, um das Dialogfeld Datenvorschau zu öffnen. Eine Liste der ein- und ausgeschlossenen Dateien wird angezeigt.

• Klicken Sie auf Validieren, um zu überprüfen, ob die Pfade und Dateinamensmuster korrekt und funktionsfähig sind.

• Geben Sie unter Zieldatensatznamen definieren einen Namen an, um das Thema der Iceberg-Zieltabelle zuzuordnen. Klicken Sie auf Weiter.

Auswählen des Inhaltstyps

Wählen Sie den Inhaltstyp der Quellereignisse aus.

• Wählen Sie den Typ der Ereignisse, die Sie erfassen, unter Typ der Datenereignisse auswählen aus.

• Weitere Informationen finden Sie unter Verbinden mit Daten-Streams.

  Der ausgewählte Inhaltstyp gilt für alle Themen, Datensätze oder Datenereignisse. Sie müssen für jeden Inhaltstyp, den Sie einlesen möchten, eine neue Aufgabe erstellen.

• Erweitern Sie Ordnungsgemäßes Laden der Elemente sicherstellen, um zu bestätigen, dass die Daten analysiert werden können. Sie sollten sicherstellen, dass die Daten in dieser Phase korrekt sind, andernfalls müssen Sie die Pipeline neu erstellen und die Daten erneut laden. Verwenden Sie Datensatz auswählen, um bestimmte Datensätze zu untersuchen und Warnungen zu prüfen, die das Laden der Daten betreffen könnten. Klicken Sie auf das Augensymbol neben beliebigen Strukturspalten, um die Daten anzuzeigen.

• Klicken Sie auf Weiter.

Festlegen von Einleseeigenschaften

Konfigurieren Sie die Einstellungen für Ihre Pipeline:

• Daten lesen aus

  • Ab dem frühesten Ereignis starten: Alle Verlaufsdaten werden eingelesen.

  • Ab jetzt starten: Neue Daten, die ab dem Start der Pipeline eingehen, werden eingelesen.

• Verschachtelung der Spalten wird aufgehoben

  • Verschachtelte Spalten beibehalten: Es werden keine Umwandlungen angewendet.

  • In separate Spalten aufteilen: Daten werden in separate Spalten aufgeteilt.

• Ladeeinstellungen für neue Datensätze

  • Nur anhängen: Dies ist im Allgemeinen die beste Option für Ereignisdaten, da diese normalerweise eine kurze Lebensdauer haben und nicht aktualisiert werden, zum Beispiel Bestellungen.

  • Änderungen anwenden: Dies eignet sich am besten für Daten, die im Lauf der Zeit aktualisiert werden, zum Beispiel Kunden. Aktualisiert vorhandene Datensätze und fügt neue Datensätze basierend auf Schlüsselfeldern ein. Sie müssen die Schlüsselfelder später beim Definieren der Aufgabe angeben.

• Partition der Zieltabelle

  Die Option für die Zieltabellenpartition gilt für alle Tabellen in der Pipeline. Sie können dies später auf Tabellenebene für eine benutzerdefinierte Partitionierung überschreiben.

  InformationshinweisDiese Option ist nur verfügbar, wenn Nur anhängen in Ladeeinstellungen ausgewählt ist.

  • Keine Partition: Tabellen werden ohne Partitionen erstellt.

  • Nach Übernahmedatum des Ereignisses partitionieren: Die Tabellen werden anhand des Datums partitioniert, an dem die Ereignisse eingelesen werden.

• Datenänderungsverarbeitung

  InformationshinweisDiese Option ist nur verfügbar, wenn Änderungen anwenden in Ladeeinstellungen ausgewählt ist.

  • Vorläufige Löschvorgänge einbeziehen: Geben Sie einen Ausdruck ein, um zu definieren, welche Datensätze zur Löschung markiert werden sollen.

  • Verlaufsdatenspeicher (Typ 2) erstellen : Dadurch werden frühere Versionen geänderter Datensätze beibehalten.

• Klicken Sie auf Weiter.

Zusammenfassung

Der Übersichtsbildschirm bietet eine visuelle Darstellung Ihrer Pipeline:

• Optional können Sie für die Streaming-Bereitstellungs- und Streaming-Umwandlungsaufgabe auf Namen und Beschreibung bearbeiten klicken, um neue Werte anzugeben.

• Wählen Sie die Option für die Aktion Nach dem Erstellen der Pipeline aus.

• Wenn Sie alle Einstellungen konfiguriert haben, klicken Sie auf Erstellen, um die Streaming-Pipeline zu erstellen.

• Wenn das Projekt angezeigt wird, können Sie jede Aufgabe vorbereiten und ausführen, um mit der Datenerfassung zu beginnen.

  • Bereiten Sie die Streaming-Bereitstellungsaufgabe vor und führen Sie sie aus.

    Weitere Informationen finden Sie unter Bereitstellen von Streaming-Daten an Qlik Open Lakehouse.

  • Bereiten Sie die Streaming-Umwandlungsaufgabe vor und führen Sie sie aus.

    Weitere Informationen finden Sie unter Speichern von Streaming-Datensätzen.

Datentypzuordnungen

Das anfängliche Quellschema basiert auf einer Datenstichprobe, die vor der PREPARE-Phase beim Erstellen Ihres Pipeline-Projekts entnommen wurde, und die Schemaentwicklung wird zur Lesezeit verarbeitet. Spiegel-Aufgaben und andere nachgelagerte Aufgaben, die STRUCT und ARRAY nicht unterstützen, verwenden einen JSON-Typ. Daten können mithilfe von SQL geparst werden.

Die folgenden Datentypzuordnungen gelten für alle unterstützten Datenquellen, variieren jedoch je nach Quelldateityp, und Folgendes sollte beachtet werden:

• Datentypen werden aus einer Stichprobe der einzugliedernden Daten abgeleitet. Wenn ein Feld in der Stichprobe beispielsweise nur Ganzzahlwerte enthält, wird es in den Streaming-Bereitstellungs- und -Umwandlungsaufgaben als INT8 erstellt. Wenn nachfolgende Daten Gleitkommawerte mit doppelter Präzision enthalten, enthalten die Bereitstellungsdateien diese Werte; wenn jedoch in der Streaming-Umwandlungsaufgabe die Einstellung Feld-Datentyp ändern auf Ignorieren festgelegt ist, bleibt die Spalte INT8 und die Gleitkommawerte werden abgeschnitten. Um unbeabsichtigte Kürzungen zu vermeiden, stellen Sie sicher, dass die Beispieldaten vor der Integration den gesamten Bereich der erwarteten Werte umfassen, oder konfigurieren Sie Feld-Datentyp ändern in frühen Phasen als Aufgabe anhalten und passen Sie die Datentypen bei Bedarf an.

• Wenn ein Feld in der Quelle zu einer Struktur hinzugefügt wird, wird es immer zum Bereitstellungsziel hinzugefügt. Für die Streaming-Umwandlung wird das Verhalten gemäß der in Einstellungen für Streaming-Umwandlungsaufgaben > Schemaentwicklung > Felder zu Struktur hinzufügen (Auf Ziel anwenden, Ignorieren, Aufgabe anhalten) gewählten Option angewendet.

• Wenn ein Feld in einem bestimmten Datensatz fehlt oder ein Array leer ist, werden sie als null behandelt.

• Wenn ein Datensatz durch ein Array abgeflacht wird und ein Datensatz eintrifft, bei dem dieses Array leer oder null ist, erstellt das System eine Zeile und das abgeflachte Feld ist null. Es wird nicht automatisch ausgeschlossen. Wenn Sie diese Zeilen ausschließen möchten, fügen Sie manuell einen Filter hinzu, zum Beispiel array_element IS NOT NULL.

• Die in der Benutzeroberfläche angezeigten Datentypen spiegeln die ausgewählte Datensatzgranularität wider. Bei abgeflachten Arrays wird der Datentyp des individuellen Elements angezeigt, nicht die Array-Struktur selbst.

• Ein neues Attribut kann nicht innerhalb einer Struktur in einem verschachtelten JSON-Feld hinzugefügt werden, sondern nur auf der Root-Ebene.

• In Stream-Umwandlungsaufgaben wird die Abflachung nur für eine einzelne Ebene eines Arrays unterstützt. Wenn die Abflachung auf ein Array mit mehreren Ebenen angewendet wird, zum Beispiel ARRAY<ARRAY<STRUCT>>, wird nur das äußere Array abgeflacht, was zu ARRAY<STRUCT> führt und nicht zu einer vollständig abgeflachten STRUCT. Zusätzlich erlaubt die aktuelle Benutzeroberfläche die Konfiguration einer Abflachung nur auf Spaltenebene. Daher wird bei Auswahl eines Arrays mit mehreren Ebenen die Abflachung implizit nur auf die erste Array-Ebene angewendet.

• Wenn Sie sich auf ein Array von Primitiven beziehen, wird der Datentyp des Elements verwendet, wenn die Granularität das Array ist. Andernfalls wird der Array-Datentyp verwendet.

  In diesem Beispiel enthält OrderDetails ein Array von CustomerID vom Datentyp INT. OrderDetails.CustomerID bedeutet INT, wenn die Granularität OrderDetails.CustomerID ist, und ARRAY<INT>, wenn die Granularität OrderDetails ist.

JSON

In JSON-Dateien bestimmt der numerische Wert in der Quelle den Zieldatentyp:

• INT8 wird für Ganzzahlwrte verwendet, die in den unterstützten Ganzzahlbereich passen und keine Bruchkomponente enthalten.

• REAL8 (DOUBLE) wird verwendet, wenn der Wert eine Bruchkomponente (Gleitkommazahl) enthält.

• STRING wird verwendet, wenn der numerische Wert den maximal unterstützten Ganzzahlbereich überschreitet.

Datentypen werden wie folgt zugeordnet:

CSV, TSV, REGEX, und SPLIT

Standardmäßig werden alle Quelldatentypen als Zeichenfolge erfasst. Verwenden Sie die Option Typen automatisch ableiten, um Quell- und Zieltypen wie folgt zuzuordnen:

Parquet

Parquet-Dateien unterstützen physische und logische Datentypen. Physische Datentypen definieren, wie Werte auf dem Datenträger gespeichert werden, wie z. B. INT32, DOUBLE oder BYTE_ARRAY. Logische Datentypen bieten eine semantische Bedeutung zusätzlich zur physischen Darstellung, zum Beispiel, indem sie identifizieren, ob ein Ganzzahlwert ein Datum darstellt. Wenn ein logischer Typ an eine Parquet-Spalte angehängt ist und in Qlik Open Lakehouse (wie unten aufgeführt) unterstützt wird, verwendet die Streaming-Bereitstellungsaufgabe den logischen Typ bei der Definition des Zielschemas anstelle des zugrunde liegenden physischen Typs. Dies stellt sicher, dass Daten korrekt interpretiert werden und beabsichtigte Semantik wie Präzision, Skalierung und zeitliche Bedeutung erhalten bleiben. Dies führt zu genaueren Schemas, wenn Daten in nachgelagerte Formate geschrieben werden.

Aus Parquet-Dateien bezogene Daten werden wie folgt zugeordnet:

Avro

Die folgenden Zuordnungen gelten für Avro-Dateien mit Schemaregistrierung.

ORC

Die folgenden Zuordnungen gelten für ORC-Dateien.

Einschränkungen und Überlegungen

• Wenn eine Struktur oder ein Array durch automatische Schema-Evolution in der Bereitstellung geändert wird, müssen nachgelagerte Ansichten, die nicht von einer Qlik Talend Cloud Streaming-Aufgabe erstellt wurden, möglicherweise aktualisiert werden, um nicht veraltet zu sein.

• Wenn eine Aufgabe Parsing-Fehler aufweist, gelangt sie nicht in einen Fehlerzustand und wird nicht als „Eingriff erforderlich“ gekennzeichnet. Da Parsing-Fehler eine ständig zunehmende Metrik sind, gibt es kein Abbruchkriterium für einen Fehlerzustand.

• Das Entfernen einer Cluster-Funktion ist nur zulässig, wenn keine Aufgaben diese Funktion verwenden.

• Aktualisierungen und Löschungen eines Datensatzes mit demselben Primärschlüssel dürfen die Partitionsgrenze nicht überschreiten, d. h. sie müssen derselben Partition zugeordnet werden.

• Wenn eine Quelle eine große Anzahl von Spalten enthält, werden in Aufgaben und im Katalog nur die 500 häufigsten Spalten angezeigt. Alle Spalten werden in den Avro-Dateien in der S3-Bereitstellung gespeichert, aber nur die 500 häufigsten Spalten werden in Iceberg-Tabellen gespeichert. Bei der Schema-Evolution wird eine neue Spalte, wenn sie hinzugefügt wird, nicht zu den Top-Spalten hinzugefügt, selbst wenn sie häufig vorkommt.