CSVファイルの解析の例でTalend Open Studioを使用する

さまざまなソースからデータを収集し、統一または便利に変換することは、かなり一般的なタスクです。 もちろん、ほとんどの場合、独自のソリューションで対応できますが、柔軟で簡単に拡張できるようにするには、多くの時間を費やす必要があります。 この場合、既製のソリューションを使用するのが賢明です。 Talend Open Studio（TOS）はそのようなソリューションの1つです。



HabréでTOSを操作することに関する記事が不足していることに、私は多少驚きました。 おそらく、これには私には理解できない理由があります。 そうであっても、私はこのギャップを埋めようとします。



おそらく、この記事を書いているときに、いくつかの問題について詳しく説明しすぎていたため、ネタバレの下にいくつかの指示を隠しました。



そのため、TOSはオープンソースのデータ統合ソリューションです。 データをTOSに変換するプロセスをカスタマイズするためのメインツールは、データとそれらの間の関係を変換する個々のノードを追加および構成できる特別なビジュアルエディターです。



TOSの興味深い機能と大きな利点は、私の意見では、TOSがコンポーネントと通信をJavaコードに変換するという事実です。 実際、データ変換グラフに基づいてコードを生成する機能を備えたJavaのライブラリを取得します。 さらに、パッケージをコンパイルして、Javaがあり、TOSがない可能性のある任意のマシンで実行できます。

コード生成により、さらにプラスが得られます。独自のコードを記述することでTOSの機能を拡張できます（このための特別なツールもあります）。



Talendの独立した全体的なデータ変換はジョブと呼ばれます。 タスクはサブタスクで構成され、サブタスクはコンポーネントと関係で構成されます。 コンポーネントはデータを直接変換するか、入力/出力を行います。 接続にはいくつかの種類があります。 コンポーネント間でデータを交換する主な手段は、「フロー」タイプの接続（フロー）です。 ストリームは、データベースのテーブルに非常に似ています。 ストリームには、スキーマ（フィールドの名前、タイプ、属性）とデータ（フィールド値）があります。 データ自体とフロー図の両方は、処理中に変更できます。 TOSのストリームは互いに同期していません。 それらは互いに独立して動作します。



次に、例を使用して、データ処理プロセスの構成方法を示します。



次の形式のCSVファイルがあるとします。



id,event_name,event_datetime,tag

1,"Hello, world!",2017-01-10T18:00:00Z,

2,"Event2",2017-01-10T19:00:00Z,tag1=q

3,Event3,2017-01-10T20:00:00Z,

4,"Hello, world!",2017-01-10T21:00:00Z,tag2=a

5,Event2,2017-01-10T22:00:00Z,

...







そして、異なるイベント（イベントフィールド）のデータを分離します。



データの操作を開始する前に、タスクを作成する必要があります。 作成プロセスは簡単なので、説明しません。



したがって、最初に行う必要があるのは、CSVを読み取って解析することです。 最初に、入力CSVファイルのメタデータレコードを作成します-これにより、さらに作業が簡単になります（メタデータ->ファイル区切り）。 ファイル区切りの作成は多かれ少なかれ直感的であるため、詳細な説明はネタバレの下に隠されています。



言及する価値がある唯一のことは、フォームフィールドの値を置換するときの引用符の配置です。 これは、区切りファイルの作成だけでなく、フォーム内の他のほとんどのフィールドにも適用されます。 実際には、さまざまなフィールドの値のほとんどがJavaコードで「現状のまま」で置き換えられます。 特定のタイプのJava式でなければなりません。 文字列定数を指定する場合、引用符で囲む必要があります。 これにより、柔軟性が向上します。 値が必要な場合はいつでも、パラメーターまたは式の値を置き換えることができます。





次に、解析に関与するコンポーネントを追加します。 必要なコンポーネントはtFileInputDelimitedと呼ばれます。







次に、パーサーを構成します。 設定のtFileInputDelimitedコンポーネントで、「プロパティタイプ」を「リポジトリ」に設定し、以前に作成したテンプレートを選択します。



これで、パーサーは必要な形式のファイルを解析するように構成され、作業を開始すると、ソースCSVファイルの内容がログに表示されます。 問題は、パーサーがテンプレートに関連付けられている場合、テンプレート内のファイルに対してハードチューニングされることです。 同じ形式の別のファイルを指定することはできません。これは、どのファイルを処理するかを事前に知らない場合、あまり便利ではない場合があります。



この状況から抜け出すには2つの方法があります。 最初の方法は、常にテンプレートのファイルを目的のファイルに置き換えることです。 2つ目は、パーサーコンポーネントとテンプレートを結び付けることです。 この場合、解析設定は保存できますが、入力ファイルを設定することは可能です。 最初の方法の欠点は明らかで、2番目の方法の欠点には、テンプレートとパーサー間の同期の欠如が含まれます。 テンプレートを変更する場合、パーサー設定を手動で同期する必要があります。 2番目の方法で、テンプレートからパーサーを切り離します。 これを行うには、「プロパティタイプ」フィールドに「ビルドイン」値を返します。 設定は保存されましたが、変更する機会がありました。



入力ファイルの名前を式context.INPUT_CSVに変更します。 名前は引用符（文字列定数）で囲まれ、引用符なしの式であることに注意してください。 これはコンテキストパラメータです。 また、コンテキストタブでこのパラメーターを作成する必要があります。 デバッグ用のデフォルト値を設定できます。 コンテキストパラメーターは、コマンドラインパラメーターとして設定できます（--context_param INPUT_CSV = pathなど）。 これは、コンパイル済みJavaパッケージの実行に適用されます。



次。 データをイベント名で区切る必要があります。



これには、tMapコンポーネントといくつかのtFilterRowが必要です。 自分自身を2つのtFilterRowに制限しましょう。 2つの異なるイベントのみをハイライトします。 図に示すようにそれらを接続します。







tMapとtFilterRowを接続する場合、接続の名前を入力する必要があります。 名前は一意である必要があります。 次に、tMapコンポーネントを構成する必要があります。 これを行うには、tMapアイコンをダブルクリックするか、コンポーネントプロパティパネルからエディターを呼び出して、Map Editorメニューに入ります。



この場合、ストリームを「コピー」するだけでよいため、すべての入力データフィールド（左側）を各出力ストリーム（右側）にドラッグアンドドロップするだけです。







中央のセクションでは、内部変数を指定できます（行番号やパラメーター置換などの新しい値を生成するために使用できます）。 エディターの各セルで、式を作成できます。 実際、私たちが行ったのは値の置換です。 スクリーンショットのrow1は、入力ストリームの名前です。 これで、マッパーは入力を2つのストリームに分割します。



tFilterRowフィルターの構成は特別なものではありません。





フィルターの後にtLogRowペアを追加して実行し、データが共有されていることを確認します。 唯一の方法は、tLogRowのModeを「Basic」モードとは異なるものに設定することです。そうしないと、データが混合されます。



tLogRowの代わりに、CSVファイルに書き込むためのtFileOutputDelimitedや、テーブルに書き込むためのデータベースコンポーネントなど、他のデータ出力コンポーネントを追加できます。



データベースの操作には多くのコンポーネントがあります。 それらの多くには、データベースへのアクセスを設定するための設定フィールドがあります。 ただし、異なるコンポーネントからデータベースに多くアクセスする場合は、次のスキームを使用するのが最適です。







Connectionコンポーネントは、データベースアクセスパラメータを設定し、接続を確立します。 Closeコンポーネントはデータベースから切断されます。 図で唯一のCommitコンポーネントがある中央のブロックでは、新しい接続を確立せずにデータベースを使用できます。 これを行うには、コンポーネント設定で「既存の接続を使用する」オプションを選択し、必要な接続コンポーネントを選択します。



別のTOSメカニズムもここで使用されます-サブジョブ。 サブタスクを作成すると、タスクの一部を完了してから他の部分を開始できます。 この例では、接続が確立されるまでCommitコンポーネントは開始されません。 OnSubjobOk接続はサブタスク間で作成されます（トリガーアイテムは、コンポーネントコンテキストメニューで利用できます。このコンテキスト内にこの接続があります）。 エラー処理のためのObSubjobErrorなど、他の関係があります。



CSVファイルの例に戻りましょう。



タグフィールドは、データベースへの書き込みにはあまり適していません-tag2 = a。 確かに、キーと値のペアをデータベース内の異なるフィールドに分割する必要があります。 これはさまざまな方法で実行できますが、tJavaFlexコンポーネントを使用して実行します。 tJavaFlexは、その動作をJavaで記述できるコンポーネントです。 設定には3つのセクションがあります。最初のセクションはデータ処理の開始前に実行され（初期化）、2番目のセクションはデータ処理に関与し、3番目のセクションはすべてのデータが処理された後に実行されます。 また、他のコンポーネントと同様に、回路エディターがあります。 データスキームからタグフィールドを削除し、いくつかの新しいタグフィールドtag_nameとtag_value（String型）を追加します。







次に、コンポーネントの中央のセクションで、

 row4.tag_name = ""; row4.tag_value = ""; if(row2.tag.contains("=")) { String[] parts = row2.tag.split("="); row4.tag_name = parts[0]; row4.tag_value = parts[1]; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

コードは簡単で、おそらく説明する価値があるのは、row4.tag_valueの形式の構築だけです。 tag_valueは、作成したフィールドの名前です。 他のフィールドにも同じ方法でアクセスできます。 row4は発信ストリーム（着信row2）の名前です。 変更することができます。







したがって、タグは2つのフィールドに分割されます。 ただし、tJavaFlex設定で[データの自動配信]チェックボックスがオンになっていることを確認する必要があります。チェックボックスをオンにしないと、他のすべてのデータが消えます。 実際、追加の変換を追加しました。 他のフィールドは名前が同じで、自動的にコピーされます。



次に、もう少し複雑で具体的な2つのことについて説明します。



まだデータをデータベースに入れたいとします。 したがって、フィールドを持つイベントラベルがあります：イベント名、イベント識別子、イベント日付、タグテーブルのエントリへのリンク。 タグテーブルには、キーと値の2つのフィールドがあります。 キーと値のペアが存在しない場合にのみタグテーブルに追加します。 また、タグとイベントテーブルの間にリンクを追加します。

すなわち 私たちが欲しい：

• タグテーブルにそのようなキーと値のペアがあるかどうかを確認し、ない場合は追加します
• タグに一致するデータベース内のレコードのIDを取得します
• タグテーブルのエントリのIDを含むイベントデータをイベントテーブルに書き込む

Postgresにない場合にのみエントリを追加するには、フォームのデザインを使用できます

挿入

存在しない場所

これは、tPostgresqlRowコンポーネントを使用して実行できます。 このコンポーネントを使用すると、任意のSQLクエリを実行できます。 ただし、リクエストでは実際のデータを置き換える必要があります。 これは、たとえば次のように行うことができます

 String.format(" INSERT INTO tag(tag_name, tag_value) SELECT \'%s\', \'%s\' WHERE NOT EXISTS (SELECT * FROM tag WHERE tag_name = \'%s\' AND tag_value = \'%s\');", input_row.tag_name, input_row.tag_value, input_row.tag_name, input_row.tag_value)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

はい、パラメーターは同じ2回リストされます（おそらくJavaの知識が乏しいかもしれません）。 Javaコードの最後にセミコロンは必要ないことに注意してください。



その後、簡単なクエリを実行して、テーブル内のレコードのIDを取得する必要があります。

しかし、Postgresの場合は、より簡単な方法でRETURNING IDを使用できます。 ただし、このメカニズムは、データが追加された場合にのみ値を返します。 ただし、サブクエリを使用すると、この制限を回避できます。 その後、リクエストは次のように変換されます。

 String.format(" WITH T1 AS ( SELECT * FROM tag WHERE tag_name = \'%s\' AND tag_value = \'%s\' ), T2 AS ( INSERT INTO tag(tag_name, tag_value) SELECT \'%s\', \'%s\' WHERE NOT EXISTS (SELECT * FROM T1) RETURNING tag_id ) SELECT tag_id FROM T1 UNION ALL SELECT tag_id FROM T2;", input_row.tag_name, input_row.tag_value, input_row.tag_name, input_row.tag_value)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

別の考慮事項は、閉じた構造を作成する必要がある場合に値します。 TOSは、周期的でない場合でも、閉じた構造を作成することを許可しません。 実際には、データストリームはそれ自体でライブであり、同期されておらず、異なる数のレコードを運ぶことができます。 確かに、閉ループを形成することなくほぼいつでも実行できます。 これを行うには、スレッドを共有してすべてを1つにまとめる必要はありません。 しかし、本当にしたい場合は、閉じた構造の作成に関する制限をバイパスできます。 tHashInputおよびtHashOutputコンポーネントが必要になります。





これらのコンポーネントを使用すると、ストリームをメモリに保存してからアクセスできます。 もちろん、一時ファイルを使用することもできますが、ハッシュはおそらくより高速です（十分なRAMがある場合）。



保存するtHashOutputコンポーネントと入力ストリームを追加します。 コンポーネントは、独立して動作するように、または別のtHashOutputコンポーネントにデータを追加するように構成できます。 この場合、コンポーネントはsqlのunionのように機能します。 データは1つの共通ストリームに記録されるため、マージを実行する新しいサブタスクを作成する必要があります。 OnSubjobOkリンクを忘れずに追加してください。



個々に動作するスレッドごとに、tHashInputコンポーネントを作成する必要があります。 このコンポーネントには欠点があります-データを取得するtHashOutputコンポーネントを指定した後でも、スキームは自動的にロードされません。 次に、すべてのtHashInputをtMapを使用して結合する必要があります。 Mainとしてマークされるのは1つのストリームのみで、残りの着信ストリームはそれを介して同期され、発信、残りの着信ストリームはLookupになります。 さらに、スレッド間の接続を設定する必要があります。設定しないと、クロスジョインが取得されます。







マッパーの左側に多くのストリームがあるという事実にもかかわらず、1つの入力ストリームがあり、マッピングに任意のフローの任意のフィールドを使用すると仮定できます。



最後まで読んでくれたすべての人に感謝します。