Wie erstellt man mehrere Ausgabedateien in Talend basierend auf einer Spalte von einem SQL-Eingang

Question

Ich habe mehrere Ausgabedateien basierend auf einem Wert (Spalte) von einer SQL-Eingabe in Talend Open Studio erstellen müssen.

Meine tMSSQLInput gibt etwa 50.000 Zeilen, in denen, wo eine der Säulen ist die BUILDING_NAME

Gebäude A Gebäude B Gebäude C ....

So werden alle Zeilen, die den Wert „Building A haben "sollte in einer Excel-Datei namens" buildingA.xls "sein, alle Zeilen, die" Building B "haben, sollten in einer Excel-Datei namens" buildingB.xls "sein und so weiter.

Ich habe versucht, tLoop oder tForEach zusammen mit TIterateToFlow zu verwenden, aber ich bin mir nicht sicher, ob ich weiß, wie man es implementiert.

Vielen Dank im Voraus.

Answer 1

Gabriele's answer sieht ziemlich gut zu mir.

Wenn Sie jedoch in einer Situation, in der Sie eine riesige Menge an Daten über viele Gebäude bis zu dem Punkt, wo Sie speichern Zeilen einzelner Gebäude im Speicher, aber nicht alles, dann wäre ich geneigt, einen etwas anderen Ansatz zu verwenden.

In diesem Beispiel Job Ich bin mit MySQL-Datenbank-Komponenten, nur weil ich eine lokale MySQL-Datenbank haben, passieren, aber alles über diesen Job gilt für Oracle oder MS SQL Server als auch wahr:

Gleich zu Beginn öffnen wir eine Verbindung zur Datenbank mit einer tMySqlConnection-Komponente in diesem Fall. Die verbleibenden 2 Datenbankkomponenten (tMySqlInput und tMySqlRow) verwenden dann die gemeinsam genutzten Verbindungsdetails.

Wir beginnen eine Liste der Gebäude in der Datenbank mit Grabbing die folgende Abfrage in der tMySqlInput mit:

"SELECT DISTINCT building 
FROM filesplittest" 

Das gibt jede unterschiedliche Gebäude.

Wir durchlaufen dann jedes Gebäude, was uns erlaubt, nur die Aufzeichnungen für dieses spezielle Gebäude für den Rest des Jobs im Speicher zu behalten.

Wir verwenden dann eine tMySqlRow-Komponente, um die Daten für das bestimmte Gebäude der Iteration mit einer vorbereiteten Anweisung zu ziehen. Das Beispiel Abfrage Ich verwende sieht aus wie:

"SELECT building, foo, bar 
FROM FileSplitTest 
WHERE building = ?" 

Und dann konfigurieren wir die vorbereitete Anweisung in den erweiterten Einstellungen:

Wo ich gesagt habe, dass der erste Parameter (Parameter Index = 1) ist der Gebäudewert, den wir früher abgerufen haben, und tFlowToIterate wurde hilfreich für uns in die globalMap verschoben, so dass wir sie in diesem Fall unter Verwendung von ((String)globalMap.get("row6.building")) abrufen (es ist die "Gebäude" -Spalte, die im row6-Fluss war).

Bei Verwendung einer vorbereiteten Anweisung Sie die Daten als Datensatz Objekt abrufen müssen, so dass Sie das Schema der tMySqlRow wie so eingestellt werden soll:

Und dann analysieren wir es mit die tParseRecordSet Komponente:

Mit einem Schema, dieses Beispiel passen:

Dann müssen wir über diese Datenmenge iterieren und sie an eine entsprechend benannte CSV anhängen.Um dies zu tun wir ein weitere tFlowToIterate Komponente und nehmen einen etwas lästigen Umweg über eine tFixedFlowInput Komponente verwenden, jeden Datensatz der Daten aus dem GlobalMap zu lesen, bevor es an die tFileOutputDelimited vorbei:

Und dann schließlich wir anhänge es in einer CSV nach dem Gebäude mit dem Namen:

Hinweis des anfügen Kontrollkästchen aktiviert ist, sonst jede Iteration des Auftrags wird die vorherige überschreibt. Wir benennen die Datei auch nach dem Wert in der Gebäude-Spalte.

---

Wie Gabriele erwähnt, Wenn Ihre Daten jederzeit gerne in Speicher passt auf Sie stattdessen die Arbeit vereinfachen, indem Sie einfach Ihre Daten in eine tHashOutput Komponente zu lesen und dann in der Hash-Filtern der Daten:

Wir beginnen mit dem Lesen aller Daten in einer tHashOutput-Komponente, die dann die Daten im Arbeitsspeicher speichert. Talend versteckt manchmal diese Komponenten aus irgendeinem Grund aber Sie können sie wieder aktivieren, indem sie wieder in den Projekteigenschaften Hinzufügen -> Designer -> Palette Einstellungen:

Als nächstes werden wir die Daten wieder in von der Lese Hash mit einer tHashInput-Komponente (mit der vorherigen tHashOutput-Komponente verknüpft - vergessen Sie nicht, der tHashInput-Komponente dasselbe Schema hinzuzufügen) und verwenden Sie dann eine tAggregateRow-Komponente und -Gruppe, indem Sie "building" verwenden, um die Gebäudewerte eindeutig zu definieren:

wir dann durch die unterschiedlichen Werte für „building“ iterieren die tFlowToIterate und anschließend Filtern des hash (in einem zweites Mal gelesen wird) durch den Gebäudewert, der zur Zeit iteriert wird:

Und schließlich, wir wieder sicher in eine Datei nach dem Wert in dem Gebäude Spalte mit dem Namen anhängen:

Answer 2

Eine Möglichkeit, dies zu tun, ist mit einem Prozess wie:

tMySqlInput -> tFlowToIterate -> tFixedFlowInput -> tFileOutputDelimited.

Hier in tFlowToIterate (Sie Ihren Schlüssel hinzufügen können - wie zum Beispiel Dateinamen, die den Wert aus einer Spalte im Schema von tMySqlInput erhalten werden)

In tFileOutputDelimited können Sie diese (String)glotbalMap.get("FileName") verwenden Sie einen Dateinamen Pfad, in dem diese zu bauen Der Dateiname kommt für jede Zeile von tMySqlInput.

Answer 3

Ich denke, es ist besser, es zu tun verbosely in einem Zwei-Stufen Job

• Zuerst die Liste der Datei bekommen zu
• Dann Route, die Sie gebaut werden die Zeilen auf Datei gehör

ich würde entwerfen einen Job wie diesen

tMSSSQL_Input_1------>tCacheOut_1 
     | 
     | 
    OnSubjobOk 
     | 
     | 
     v 
    tCacheIn_1------->tAggregateRow------>tFlowToIterate 
              /
              /
             (iterate) 
              /
             /
             /
     +---------------------------------+ 
     | 
     | 
     v 
    tCacheIn_1------->tFilterRow-------->tFileOutDelimited 

Lassen Sie mich erklären, was los ist

• Im ersten Subjob Sie die Tabelle in einem Speicherpuffer Dump (tCacheOut, auf Talend Exchange ist eine gute Komponente, aber out-of-the-box tHashInput/tHashOutput wird die Arbeit tun, auch) - Dies ist für die Abfrage der DB nur einmal, aber wenn keine Leistung erforderlich ist, können Sie mehrere Abfragen triggern und die Verwendung des Speicherpuffers vermeiden
• Dann lesen Sie den Dump ein erstes Mal, um Ihre Gebäude zu unterscheiden (mit tAggregateRow gegen die Gebäudesäule)
• Dann wechseln Sie zu einem iterierten Fluss und speichern den aktuellen Gebäudewert in einer globalen Variablen, nennen wir es "my_building "
• Dann lesen Sie ein zweites Mal Ihre Dump und filtern nur Zeilen des aktuellen Gebäudes. In Fakten, können Sie globalMap.get("my_building") innerhalb Ihrer Filterbedingungen
• Schließlich werden Sie diese Zeilen in einer richtigen Datei speichern, wieder mit globalMap.get("my_building"), um Ihren Dateinamen zu parametrisieren.