SparkジョブのAzure Databricks接続パラメーターの定義 - 7.3

Sparkバッチ

EnrichVersion
Cloud
7.3
EnrichProdName
Talend Big Data
Talend Big Data Platform
Talend Data Fabric
Talend Open Studio for Big Data
Talend Real-Time Big Data Platform
EnrichPlatform
Talend Studio
task
ジョブデザインと開発 > ジョブデザイン > ジョブフレームワーク > Sparkバッチ

ジョブの[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クラスターで並行して実行できます。そうしないと、実行ごとにクラスターが自動的に再起動されるため、並行して起動されたジョブが互いに割り込みや実行エラーが発生します。

手順

  1. 基本的な設定情報を入力します。
    [Use dataset API in migrated components] (移行したコンポーネントでデータセットAPIを使用) コンポーネントにRDD APIではなくDataset APIを使用させる場合は、このチェックボックスを選択します。
    重要: ジョブにtDeltaLakeInputコンポーネントとtDeltaLakeOutputコンポーネントが含まれている場合は、このチェックボックスを選択する必要があります。
    [Use timestamp for dataset components] (データセットコンポーネントにタイムスタンプを使用) このチェックボックスをオンにすると、日付でjava.sql.Timestamp使われます。
    注: このチェックボックスをオフのままにしておくと、パターンに応じてjava.sql.Timestampjava.sql.Dateが使われます。
    [Use local timezone] (ローカルタイムゾーンを使用) このチェックボックスをオンにすると、Sparkはシステムで指定されているローカルタイムゾーンを使います。
    注:
    • このチェックボックスがオフのままであれば、SparkではUTCタイムゾーンが使われます。
    • コンポーネントによっては[Use local timezone for date] (日付にローカルタイムゾーンを使用)チェックボックスもあります。コンポーネントのチェックボックスがオフのままであれば、Sparkで設定されているタイムゾーンが継承されます。
  2. [Cloud provider] (クラウドプロバイダー)ドロップダウンリストでAzureを選択します。
  3. Databricksの基本的な接続情報を入力します。

    Standalone (スタンドアロン)

    • [Use pool] (プールを使用): Databricksプールを活用する場合はこのチェックボックスをオンにします。選択した場合は、[Spark Configuration] (Spark設定)でクラスターIDではなくプールIDを指定する必要があります。また、[Use transient cluster] (一時クラスターを使用)チェックボックスをオンにする必要もあります。

    • [Endpoint] (エンドポイント)フィールドに、Azure DatabricksワークスペースのURLアドレスを入力します。このURLは、AzureポータルのDatabricksワークスペースページの[Overview] (概要)のブレードにあります。この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] (ユーザー設定)ページで生成または検索できます。詳細は、Azureドキュメンテーション[Personal access tokens] (パーソナルアクセストークン)を参照してください。

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

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