メイン コンテンツをスキップする 補完的コンテンツへスキップ

Apache Spark StreamingのtMysqlLookupInputプロパティ

これらのプロパティは、Spark Streamingジョブのフレームワークで実行されているtMysqlLookupInputを設定するために使われます。

Spark StreamingtMysqlLookupInputコンポーネントは、データベースファミリーに属しています。

このコンポーネントのストリーミングバージョンは、Talend Real Time Big Data PlatformおよびTalend Data Fabricで利用できます。

基本設定

[Property type] (プロパティタイプ)

[Built-in] (組み込み)[Repository] (リポジトリー)のいずれかです。

[Built-In] (組み込み): 一元的に保存されるプロパティデータはありません。

[Repository] (リポジトリー): プロパティが保存されているリポジトリーファイルを選択します。

このアイコンをクリックすると、データベース接続ウィザードが開き、コンポーネント[Basic settings] (基本設定)ビューに設定したデータベース接続パラメーターが保存されます。

データベース接続パラメーターの設定と保存の詳細は、Talend Studioユーザーガイドをご覧ください。

[Use an existing connection] (既存の接続を使用)

定義済みの接続の詳細を再利用する場合は、このチェックボックスをオンにして、[Component List] (コンポーネントリスト)ドロップダウンリストから、目的の接続コンポーネントを選択します。

[DB version] (データベースバージョン)

使用するデータベースのバージョンを選択します。

使用するデータベースがRDS Auroraの場合、Mysql 5を選択する必要があります。

[Host] (ホスト)

データベースサーバーのIPアドレス。

[Port] (ポート)

データベースサーバーのリスニングポート番号。

[Database] (データベース)

データベースの名前。

[Username] (ユーザー名)[Password] (パスワード)

データベースユーザー認証データ。

パスワードを入力するには、パスワードフィールドの横にある[...]ボタンをクリックし、ポップアップダイアログボックスにパスワードを二重引用符の間に入力し、OKをクリックして設定を保存します。

[Schema] (スキーマ)[Edit schema] (スキーマを編集)

スキーマとは行の説明のことです。処理して次のコンポーネントに渡すフィールド(カラム)数を定義します。Sparkジョブを作成する場合、フィールドの命名時は予約語のlineを避けます。

 

[Built-in] (組み込み): そのコンポーネントに対してのみスキーマを作成し、ローカルに保管します。

 

[Repository] (リポジトリー): スキーマは作成済みで、リポジトリーに保管されています。さまざまなプロジェクトやジョブデザインで再利用できます。

 

スキーマを変更するには[Edit schema] (スキーマを編集)をクリックします。現在のスキーマがリポジトリータイプの場合は、3つのオプションを利用できます。

  • [View schema] (スキーマを表示): スキーマのみを表示する場合は、このオプションを選択します。

  • [Change to built-in property] (組み込みのプロパティに変更): ローカルで変更を行うためにスキーマを組み込みに変更する場合は、このオプションを選択します。

  • [Update repository connection] (リポジトリー接続をアップデート): リポジトリーに保存されているスキーマに変更を加え、変更後にそのコンテンツをすべてのジョブにプロパゲートするかどうかを決める場合は、このオプションを選択します。変更を現在のジョブにのみ反映する場合は、変更後、[No] (いいえ)を選択し、[Repository Content] (リポジトリーコンテンツ)ウィンドウで再びこのスキーマのメタデータを選択します。

[Table Name] (テーブル名)

読み取るテーブルの名前。

[Query type] (クエリータイプ)[Query] (クエリー)

クエリーを入力します。フィールドの順序に気を付けて、スキーマ定義と一致するようにしてください。

クエリーの結果には、tMapで使用する結合キーに一致するレコードのみ含まれていなければなりません。言い換えれば、一致するレコードのみがルックアップフローに読み取られるSQLステートメントを作成するため、メインフローのスキーマをtMapで使用する必要があります。

たとえば、以下のようにします
"SELECT id, name
 FROM my_input_table
 WHERE id = `" + row1.id + "`" 

このコードでrow1tMysqlLookupInputには接続されないものの、tMapに入るメイン行となります。

この方式では、重複レコードはメモリに読み取られず、後続のコンポーネントに出力されます。

詳細設定

[Additional JDBC parameters] (追加のJDBCパラメーター)

作成するデータベース接続に接続の追加プロパティを指定します。

[Trim all the String/Char columns] (すべての文字列/文字カラムをトリミング)

すべてのStringカラム/Charカラムの先頭や末尾の空白を削除する場合は、このチェックボックスを選択します。

[Trim column] (カラムをトリミング)

定義されたカラムから先頭や末尾の空白を削除します。

情報メモ注:

[Trim all the String/Char columns] (すべての文字列/文字カラムをトリミング)の選択を解除して、このフィールドのトリミングカラムを有効にします。

[Connection pool] (接続プール)

このエリアでは、各Sparkエグゼキューターに、同時に開いたままにする接続の数を制御するための接続プールを設定するために使われます。以下の接続プールパラメーターに与えられているデフォルト値は、ほとんどのユースケースでそのまま利用できます。

  • [Max total number of connections] (接続の最大合計数): 同時に開いたままにしておくことができる接続(アイドルまたはアクティブ)の最大数を入力します。

    デフォルトの数は8です。-1を入力すると、同時に開いておける接続の数が無制限となります。

  • [Max waiting time (ms)] (最大待機時間(ミリ秒)): 接続使用の要求に対して接続プールからレスポンスが返されるまでの最大待機時間を入力します。デフォルトでは-1(無制限)となっています。

  • [Min number of idle connections] (アイドル接続の最小数): 接続プール内に維持されるアイドル接続(使用されていない接続)の最小数を入力します。

  • [Max number of idle connections] (アイドル接続の最大数): 接続プール内に維持されるアイドル接続(使用されていない接続)の最大数を入力します。

[Evict connections] (接続を無効化)

接続プール内の接続を破棄する条件を定義する場合は、このチェックボックスを選択します。オンにすると、以下のフィールドが表示されます。

  • [Time between two eviction runs] (2つの削除実行の間隔): コンポーネントが接続のステータスを確認し、アイドル状態の接続を破棄するまでの間隔(ミリ秒)を入力します。

  • [Min idle time for a connection to be eligible to eviction] (接続が削除可能になるまでの最小アイドル時間): アイドル接続が破棄されるまでの間隔(ミリ秒)を入力します。

  • [Soft min idle time for a connection to be eligible to eviction] (接続が削除可能になるまでのソフト最小アイドル時間): このパラメーターの機能は[Min idle time for a connection to be eligible to eviction] (接続が削除可能になるまでの最小アイドル時間)と同じですが、[Min number of idle connections] (アイドル接続の最小数)フィールドで定義したアイドル接続の最小数が維持されます。

使用方法

使用ルール

このコンポーネントは、開始コンポーネントとして使用され、出力リンクを必要とします。

このコンポーネントは、MySQLに接続するために同じジョブ内にあるtMysqlConfigurationコンポーネントを使用する必要があります。[Use an existing connection](既存の接続を使用)チェックボックスをオンにし、使用するtMysqlConfigurationコンポーネントを選択する必要があります。

このコンポーネントは、所属するSpark Streamingのコンポーネントのパレットと共に、Spark Streamingジョブを作成している場合にだけ表示されます。

特に明記していない限り、このドキュメントのシナリオでは、標準ジョブ、つまり従来の Talend データ統合ジョブだけを扱います。

[Spark Connection] (Spark接続)

[Run] (実行)ビューの[Spark configuration] (Spark設定)タブで、ジョブ全体でのSparkクラスターへの接続を定義します。また、ジョブでは、依存jarファイルを実行することを想定しているため、Sparkがこれらのjarファイルにアクセスできるように、これらのファイルの転送先にするファイルシステム内のディレクトリーを指定する必要があります。
  • Yarnモード(YarnクライアントまたはYarnクラスター):
    • Google Dataprocを使用している場合、[Spark configuration] (Spark設定)タブの[Google Storage staging bucket] (Google Storageステージングバケット)フィールドにバケットを指定します。

    • HDInsightを使用している場合、[Spark configuration] (Spark設定)タブの[Windows Azure Storage configuration] (Windows Azure Storage設定)エリアでジョブのデプロイメントに使用するブロブを指定します。

    • Altusを使用する場合は、[Spark configuration] (Spark設定)タブでジョブのデプロイにS3バケットまたはAzure Data Lake Storageを指定します。
    • Quboleを使用する場合は、ジョブにtS3Configurationを追加し、QuboleでS3システム内に実際のビジネスデータを書き込みます。tS3Configurationを使用しないと、このビジネスデータはQubole HDFSシステムに書き込まれ、クラスターをシャットダウンすると破棄されます。
    • オンプレミスのディストリビューションを使用する場合は、クラスターで使われているファイルシステムに対応する設定コンポーネントを使用します。一般的に、このシステムはHDFSになるため、tHDFSConfigurationを使用します。

  • [Standalone mode] (スタンドアロンモード): クラスターで使われているファイルシステム(tHDFSConfiguration Apache Spark BatchtS3Configuration Apache Spark Batchなど)に対応する設定コンポーネントを使用します。

    ジョブ内に設定コンポーネントがない状態でDatabricksを使用している場合、ビジネスデータはDBFS (Databricks Filesystem)に直接書き込まれます。

この接続は、ジョブごとに有効になります。

このページは役に立ちましたか?

このページまたはコンテンツにタイポ、ステップの省略、技術的エラーなどの問題が見つかった場合はお知らせください。