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

Databricks

Version
7.2
Language
日本語 (日本)
Product
Talend Big Data
Talend Big Data Platform
Talend Data Fabric
Talend Open Studio for Big Data
Talend Real-Time Big Data Platform
Module
Talend Studio
Content
ジョブデザインと開発 > ジョブデザイン > Hadoopディストーション > Databricks
ジョブデザインと開発 > ジョブデザイン > Serverless > 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システムの管理者に問い合わせてご確認ください。

手順

  1. 基本的な設定情報を入力します:
    [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.Timestampjava.sql.Dateが使われます。
  2. [Cloud provider] (クラウドプロバイダー)ドロップダウンリストでAWSを選択します。
  3. Databricksの基本的な接続情報を入力します:

    スタンドアロン

    • [Endpoint] (エンドポイント)フィールドに、Databricks on AWSのワークスペースのURLアドレスを入力します。たとえば、このURLはhttps://<your_endpoint>.cloud.databricks.comのようになります。

    • [Cluster ID] (クラスターID)フィールドに、使用する DatabricksクラスターのIDを入力します。このIDは、Sparkクラスターのspark.databricks.clusterUsageTags.clusterIdプロパティの値です。このプロパティは、クラスターの[Spark UI]ビューの[Environment] (環境)タブのプロパティリストにあります。

      このIDはDatabricksクラスターのURLからでも容易に見つかります。このURLのcluster/の直後に置かれています。

      このフィールドは使用されないため、一時クラスターを使用している場合は使用できません。

    • [Token] (トークン)フィールドの横の[...]ボタンをクリックして、Databricksユーザーアカウントに生成された認証トークンを入力します。このトークンは、Databricksワークスペースの[User settings] (ユーザー設定)ページで生成または検索できます。詳細は、Databricksドキュメンテーションの「Token management(トークン管理)」を参照してください。

    • [DBFS dependencies folder] (DBFS依存項目フォルダー)フィールドに、ランタイムにDatabricks Filesystemの依存項目に関連するジョブを保存するために使用するディレクトリーを入力し、このディレクトリーの末尾にスラッシュ(/)を入力します。たとえば、/jars/と入力して、依存項目をjarsという名前のフォルダーに保管します。このフォルダーが存在しない場合、ここで作成されます。

      このディレクトリーにはDBFSに対するジョブ依存項目のみ保存されます。ジョブでtS3ConfigurationtDynamoDBConfigurationコンポーネント、または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. [Autoscale] (自動スケーリング): このチェックボックスをオンまたはオフにして、一時的なクラスターで使用するワーカーの数を定義します。
        1. このチェックボックスをオンにすると、自動スケーリングが有効になります。次に、[Min workers] (最小ワーカー)にワーカーの最小数、[Max workers] (最大ワーカー)に最大ワーカー数を定義します。一時的なクラスターは、ワークロードに基づいてこのスコープ内でスケールアップおよびスケールダウンされます。

          Databricksのドキュメンテーションによると、自動スケーリングはDatabricksランタイムバージョン3.0以降で最適に機能します。

        2. このチェックボックスをオフにすると、自動スケーリングは無効になります。次に、一時的なクラスターに必要なワーカーの数を定義します。この数には、Sparkドライバーノードは含まれません。
      2. [Node type] (ノードタイプ)および[Driver node type] (ドライバーノードタイプ): ワーカーとSparkドライバーノードのノードタイプを選択します。これらのタイプは、ノードの容量とDatabricksによる価格設定を決定します。

        これらのノードタイプと使用するDatabricksユニットの詳細は、Databricksドキュメントの「Supported Instance Types(サポートされるインスタンスタイプ)」 を参照してください。

      3. [Elastic disk] (エラスティックディスク): このチェックボックスをオンにすると、Sparkワーカーのディスクスペースが不足している時に、一時的なクラスターが自動的にディスクスペースを拡大できるようになります。

        このエラスティックディスク機能の詳細は、Databricksのドキュメンテーションでローカルストレージの自動スケーリングに関するセクションを検索してください。

      4. [SSH public key] (SSHパブリックキー): クラスターにSSHアクセスがセットアップされている場合は、生成されたSSHキーペアのパブリックキーを入力します。このパブリックキーは、一時的なクラスターの各ノードに自動的に追加されます。SSHアクセスが設定されていない場合、このフィールドは無視してください。

        クラスターへのSSHアクセスの詳細は、Databricksドキュメントの「SSH access to clusters(クラスターへのSSHアクセス)」を参照してください。

      5. [Configure cluster log] (クラスターログの設定): このチェックボックスをオンにすると、Sparkログを長期間保存する場所を定義できます。このストレージシステムは、S3またはDBFSです。
    • [Do not restart the cluster when submitting] (送信時にクラスターを再起動しない): Studioがジョブを送信する時にStudioがクラスターを再起動しないようにするには、このチェックボックスをオンにします。ただし、ジョブで変更を行う場合は、このチェックボックスをオフにして、Studioがクラスターを再起動してこれらの変更が考慮されるようにします。

タスクの結果

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

Sparkチェックポイントオペレーションの詳細は、http://spark.apache.org/docs/latest/streaming-programming-guide.html#checkpointingを参照してください。