Sparkジョブ用にAWSでのDatabricks接続パラメーターを定義

ジョブの[Run] (実行)ビューの[Spark configuration] (Spark設定)タブでDatabricks接続設定を完了します。この設定は、ジョブごとに有効です。

このセクションの情報は、Talend Data Fabric またはビッグデータ対応のTalend製品のいずれかにサブスクライブしているユーザーだけを対象とします。また、Talend Open Studio for Big Dataユーザーは対象外です。

始める前に

1. Spark Streamingジョブを実行する場合、同じDatabricksクラスター上で実行できるジョブは一度に1つだけです。
2. Spark Batchジョブの実行時に、[Do not restart the cluster when submitting] (送信時にクラスターを再起動しない)チェックボックスをオンにした場合のみ、複数のジョブを送信して同じDatabricksクラスターで並行して実行できます。そうしないと、実行ごとにクラスターが自動的に再起動されるため、並行して起動されたジョブが互いに割り込みや実行エラーが発生します。
使用するAWSアカウントで使用するS3バケットに対する適切な読み書き権限があることを確認します。AWSシステムの管理者に問い合わせてご確認ください。

手順

基本的な設定情報を入力します:

[Use local timezone] (ローカルタイムゾーンを使用)	このチェックボックスをオンにすると、Sparkはシステムで指定されているローカルタイムゾーンを使います。注: このチェックボックスをオフにすると、SparkによってUTCタイムゾーンが使用されます。コンポーネントによっては[Use local timezone for date] (日付にローカルタイムゾーンを使用)チェックボックスもあります。コンポーネントのチェックボックスをオフにすると、Spark設定からのタイムゾーンを継承します。
[Use dataset API in migrated components] (移行したコンポーネントでデータセットAPIを使用)	このチェックボックスを選択し、コンポーネントにResilient Distribued Dataset (RDD) APIの代わりにDataset (DS) APIを使用させます: チェックボックスを選択する場合は、ジョブ内のコンポーネントはDSで実行されて、パフォーマンスが向上します。チェックボックスをオフにする場合は、ジョブ内のコンポーネントはRDDで実行されて、変更されていない状態のままです。これにより、下位互換性が保証されます。重要: ジョブにtDeltaLakeInputコンポーネントとtDeltaLakeOutputコンポーネントが含まれている場合は、このチェックボックスを選択する必要があります。注: 7.3で新しく作成されたジョブはDSを使用し、7.3以前のバージョンからインポートされたジョブはデフォルトでRDDを使用します。しかし、すべてのコンポーネントがRDDからDSに移行されないため、デフォルトでエラーを避けるには、チェックボックスをオフにすることをお勧めします。
[Use timestamp for dataset components] (データセットコンポーネントにタイムスタンプを使用)	このチェックボックスをオンにすると、日付で`java.sql.Timestamp`使われます。注: このチェックボックスをオフのままにしておくと、パターンに応じて`java.sql.Timestamp`か`java.sql.Date`が使われます。

[Cloud provider] (クラウドプロバイダー)ドロップダウンリストでAWSを選択します。
From the Run mode drop-down list, select the method you want to use to run your Job on Databricks:
- Create and run now: a new Job is created and runned immediately. With this method, you can retrieve your Job with its ID in your Databricks workspace. For more information, see Run now, from the official Databricks documentation.
- [Runs submit] (実行を送信): ジョブを作成することなく、1回限りの実行を送信できます。With this method, nothing is displayed in the user interface and no Job ID are created in your Databricks workspace. 詳細は、Databricksの公式ドキュメンテーションでRuns submitを参照してください。

Databricksの基本的な接続情報を入力します:

[Standalone] (スタンドアロン)	[Use pool] (プールを使用): Databricksプールを活用する場合はこのチェックボックスをオンにします。選択した場合は、[Spark configuration] (Spark設定)でクラスターIDではなくプールIDを指定する必要があります。また、[Use transient cluster] (一時クラスターを使用)チェックボックスをオンにする必要もあります。 [Endpoint] (エンドポイント)フィールドに、Databricks on AWSのワークスペースのURLアドレスを入力します。このURLは、https://adb-$workspaceId.$random.azuredatabricks.netといった形式になります。 [Cluster ID] (クラスターID)フィールドに、使用する DatabricksクラスターのIDを入力します。このIDは、Sparkクラスターのspark.databricks.clusterUsageTags.clusterIdプロパティの値です。このプロパティは、クラスターの[Spark UI]ビューの[Environment] (環境)タブのプロパティリストにあります。このIDはDatabricksクラスターのURLからでも容易に見つかります。このURLのcluster/の直後に置かれています。 [Use Pool] (プールを使用)オプションを選択した場合は、使用するDatabricksプールのIDを[Pool ID] (プールID)フィールドに入力する必要があります。このIDは、使用するプールのDatabricksInstancePoolIdキーの値です。このキーは、使用するプールの[Configuration] (設定)タブの[Tags] (タグ)の下で見つかります。また、プールを使用するクラスターのタグ内にもあります。このIDはDatabricksプールのURLでもすぐに見つかります。このURLのcluster/instance-pools/view/の直後に置かれています。 [Token] (トークン)フィールドの横にある[...]ボタンをクリックして、Databricksユーザーアカウントに生成された認証トークンを入力します。このトークンは、Databricksワークスペースの[User settings] (ユーザー設定)ページで生成または検索できます。詳細は、DatabricksドキュメンテーションでPersonal access tokens (パーソナルアクセストークン)をご覧ください。 [DBFS dependencies folder] (DBFS依存項目フォルダー)フィールドに、Databricksファイルシステムでのジョブ依存項目の保存にランタイムで使用するディレクトリーを入力し、このディレクトリーの末尾にスラッシュ( / )を入力します。たとえば、/jars/と入力して、依存項目をjarsという名前のフォルダーに保管します。このフォルダーが存在しない場合、ここで作成されます。このディレクトリーにはDBFSに対するジョブ依存項目のみ保存されます。ジョブでtS3Configurationコンポーネント、tDynamoDBConfigurationコンポーネント、またはSpark StreamingジョブでKinesisのコンポーネントを使用して、関連するシステムとの間でビジネスデータの読み取りと書き込みを行います。 [Poll interval when retrieving Job status (in ms)] (ジョブステータスを取得する時のポーリング間隔(ミリ秒)): StudioがジョブのステータスをSparkに要求する時間間隔(ミリ秒)を引用符なしで入力します。たとえば、このステータスは[Pending] (保留中)または[Running] (実行中)になります。デフォルト値は`300000`で30秒です。この間隔は、ジョブステータスを正しく取得できるようDatabricksによって推奨されています。 [Use transient cluster] (一時クラスターを使用): このチェックボックスをオンにすると一時的なDatabricksクラスターを活用できます。 [Advanced properties] (詳細プロパティ)テーブルで定義したカスタムプロパティは、一時クラスターによってランタイムで自動的に有効になります。 [Use policy] (使用ポリシー): このチェックボックスをオンにして、一時クラスターで使用するポリシーの名前を入力します。ポリシーを使用して、ルールのセットに基づいてクラスターを設定する機能を制限できます。クラスターポリシーの詳細は、Databricksの公式ドキュメントでManage cluster policiesをご覧ください。 [Autoscale] (自動スケーリング): このチェックボックスをオンまたはオフにして、一時クラスターで使用するワーカーの数を定義します。このチェックボックスをオンにすると、自動スケーリングが有効になります。次に、[Min workers] (最小ワーカー)にワーカーの最小数、[Max workers] (最大ワーカー)に最大ワーカー数を定義します。一時クラスターは、ワークロードに基づいてこのスコープ内でスケールアップおよびスケールダウンされます。 Databricksのドキュメンテーションによると、自動スケーリングはDatabricksランタイムバージョン3.0以降で最適に機能します。このチェックボックスをオフにすると、自動スケーリングは無効になります。次に、一時クラスターに必要なワーカーの数を定義します。この数には、Sparkドライバーノードは含まれません。 [Node type] (ノードタイプ)および[Driver node type] (ドライバーノードタイプ): ワーカーとSparkドライバーノードのノードタイプを選択します。これらのタイプは、ノードの容量とDatabricksによる価格設定を決定します。これらのノードタイプと使用するDatabricksユニットの詳細は、DatabricksドキュメンテーションでSupported Instance Types (サポートされるインスタンスタイプ)をご覧ください。 [Elastic disk] (エラスティックディスク): このチェックボックスをオンにすると、Sparkワーカーのディスクスペースが不足している時に、一時クラスターが自動的にディスクスペースを拡大できるようになります。このエラスティックディスク機能の詳細は、Databricksのドキュメンテーションでローカルストレージの自動スケーリングに関するセクションを検索してください。 [SSH public key] (SSHパブリックキー): クラスターにSSHアクセスがセットアップされている場合は、生成されたSSHキーペアのパブリックキーを入力します。このパブリックキーは、一時クラスターの各ノードに自動的に追加されます。SSHアクセスが設定されていない場合、このフィールドは無視してください。クラスターへのSSHアクセスの詳細は、公式DatabricksドキュメンテーションからSSH access to clusters(クラスターへのSSHアクセス)をご覧ください。 [Configure cluster log] (クラスターログの設定): このチェックボックスをオンにすると、Sparkログを長期間保存する場所を定義できます。このストレージシステムは、S3またはDBFSです。 [Do not restart the cluster when submitting] (送信時にクラスターを再起動しない): Studioがジョブを送信する時にStudioがクラスターを再起動しないようにする場合は、このチェックボックスをオンにします。ただし、ジョブで変更を行う場合は、このチェックボックスをオフにして、Studioがクラスターを再起動してこれらの変更が考慮されるようにします。

[Standalone] (スタンドアロン)

[Use pool] (プールを使用): Databricksプールを活用する場合はこのチェックボックスをオンにします。選択した場合は、[Spark configuration] (Spark設定)でクラスターIDではなくプールIDを指定する必要があります。また、[Use transient cluster] (一時クラスターを使用)チェックボックスをオンにする必要もあります。
[Endpoint] (エンドポイント)フィールドに、Databricks on AWSのワークスペースのURLアドレスを入力します。このURLは、https://adb-$workspaceId.$random.azuredatabricks.netといった形式になります。
[Cluster ID] (クラスターID)フィールドに、使用する DatabricksクラスターのIDを入力します。このIDは、Sparkクラスターのspark.databricks.clusterUsageTags.clusterIdプロパティの値です。このプロパティは、クラスターの[Spark UI]ビューの[Environment] (環境)タブのプロパティリストにあります。

このIDはDatabricksクラスターのURLからでも容易に見つかります。このURLのcluster/の直後に置かれています。
[Use Pool] (プールを使用)オプションを選択した場合は、使用するDatabricksプールのIDを[Pool ID] (プールID)フィールドに入力する必要があります。このIDは、使用するプールのDatabricksInstancePoolIdキーの値です。このキーは、使用するプールの[Configuration] (設定)タブの[Tags] (タグ)の下で見つかります。また、プールを使用するクラスターのタグ内にもあります。

このIDはDatabricksプールのURLでもすぐに見つかります。このURLのcluster/instance-pools/view/の直後に置かれています。
[Token] (トークン)フィールドの横にある[...]ボタンをクリックして、Databricksユーザーアカウントに生成された認証トークンを入力します。このトークンは、Databricksワークスペースの[User settings] (ユーザー設定)ページで生成または検索できます。詳細は、DatabricksドキュメンテーションでPersonal access tokens (パーソナルアクセストークン)をご覧ください。
[DBFS dependencies folder] (DBFS依存項目フォルダー)フィールドに、Databricksファイルシステムでのジョブ依存項目の保存にランタイムで使用するディレクトリーを入力し、このディレクトリーの末尾にスラッシュ( / )を入力します。たとえば、/jars/と入力して、依存項目をjarsという名前のフォルダーに保管します。このフォルダーが存在しない場合、ここで作成されます。

このディレクトリーにはDBFSに対するジョブ依存項目のみ保存されます。ジョブでtS3Configurationコンポーネント、tDynamoDBConfigurationコンポーネント、またはSpark StreamingジョブでKinesisのコンポーネントを使用して、関連するシステムとの間でビジネスデータの読み取りと書き込みを行います。
[Poll interval when retrieving Job status (in ms)] (ジョブステータスを取得する時のポーリング間隔(ミリ秒)): StudioがジョブのステータスをSparkに要求する時間間隔(ミリ秒)を引用符なしで入力します。たとえば、このステータスは[Pending] (保留中)または[Running] (実行中)になります。

デフォルト値は300000で30秒です。この間隔は、ジョブステータスを正しく取得できるようDatabricksによって推奨されています。
[Use transient cluster] (一時クラスターを使用): このチェックボックスをオンにすると一時的なDatabricksクラスターを活用できます。

[Advanced properties] (詳細プロパティ)テーブルで定義したカスタムプロパティは、一時クラスターによってランタイムで自動的に有効になります。
1. [Use policy] (使用ポリシー): このチェックボックスをオンにして、一時クラスターで使用するポリシーの名前を入力します。ポリシーを使用して、ルールのセットに基づいてクラスターを設定する機能を制限できます。クラスターポリシーの詳細は、Databricksの公式ドキュメントでManage cluster policiesをご覧ください。
2. [Autoscale] (自動スケーリング): このチェックボックスをオンまたはオフにして、一時クラスターで使用するワーカーの数を定義します。
  1. このチェックボックスをオンにすると、自動スケーリングが有効になります。次に、[Min workers] (最小ワーカー)にワーカーの最小数、[Max workers] (最大ワーカー)に最大ワーカー数を定義します。一時クラスターは、ワークロードに基づいてこのスコープ内でスケールアップおよびスケールダウンされます。
    Databricksのドキュメンテーションによると、自動スケーリングはDatabricksランタイムバージョン3.0以降で最適に機能します。
  2. このチェックボックスをオフにすると、自動スケーリングは無効になります。次に、一時クラスターに必要なワーカーの数を定義します。この数には、Sparkドライバーノードは含まれません。
3. [Node type] (ノードタイプ)および[Driver node type] (ドライバーノードタイプ): ワーカーとSparkドライバーノードのノードタイプを選択します。これらのタイプは、ノードの容量とDatabricksによる価格設定を決定します。
  これらのノードタイプと使用するDatabricksユニットの詳細は、DatabricksドキュメンテーションでSupported Instance Types (サポートされるインスタンスタイプ)をご覧ください。
4. [Elastic disk] (エラスティックディスク): このチェックボックスをオンにすると、Sparkワーカーのディスクスペースが不足している時に、一時クラスターが自動的にディスクスペースを拡大できるようになります。
  このエラスティックディスク機能の詳細は、Databricksのドキュメンテーションでローカルストレージの自動スケーリングに関するセクションを検索してください。
5. [SSH public key] (SSHパブリックキー): クラスターにSSHアクセスがセットアップされている場合は、生成されたSSHキーペアのパブリックキーを入力します。このパブリックキーは、一時クラスターの各ノードに自動的に追加されます。SSHアクセスが設定されていない場合、このフィールドは無視してください。
  クラスターへのSSHアクセスの詳細は、公式DatabricksドキュメンテーションからSSH access to clusters(クラスターへのSSHアクセス)をご覧ください。
6. [Configure cluster log] (クラスターログの設定): このチェックボックスをオンにすると、Sparkログを長期間保存する場所を定義できます。このストレージシステムは、S3またはDBFSです。
[Do not restart the cluster when submitting] (送信時にクラスターを再起動しない): Studioがジョブを送信する時にStudioがクラスターを再起動しないようにする場合は、このチェックボックスをオンにします。ただし、ジョブで変更を行う場合は、このチェックボックスをオフにして、Studioがクラスターを再起動してこれらの変更が考慮されるようにします。

タスクの結果

ジョブを耐障害性にする必要がある場合は、[Activate checkpointing] (チェックポイントを有効化)チェックボックスをオンにしてSparkチェックポイントオペレーションを有効にします。メタデータなどの計算のコンテキストデータ、この計算から生成されたRDDをSparkが保存する、クラスターのファイルシステム内のディレクトリーを、表示されたフィールドに入力します。

Sparkチェックポイント操作に関する詳細は、の公式Sparkドキュメンテーションをご覧ください。