DataWorks:オープンソースエンジンからのタスクのエクスポート

制限

• バージョンの制限: DataWorks エクスポートツールを使用して、DolphinScheduler 1.3.x、2.x、および 3.x の DolphinScheduler ノードのみをエクスポートし、DataWorks にインポートできます。

• ノードタイプの変換の制限:

  • SQL ノード: SQL ノードの変換は、一部のタイプのコンピューティングエンジンでのみサポートされています。ノードタイプの変換中に、SQL コードの構文は変換されず、SQL コードは変更されません。

  • Cron 式: 特定のシナリオでは、Cron 式が削除されるか、Cron 式がサポートされない場合があります。構成されているスケジュール時間がビジネス要件を満たしているかどうかを確認する必要があります。スケジュール時間については、「時間プロパティを構成する」をご参照ください。

  • Python ノード: DataWorks は Python ノードを提供していません。DataWorks は、DolphinScheduler の Python ノードを Python ファイルリソースと、Python ファイルリソースを参照する Shell ノードに変換できます。ただし、Python ノードのスケジュールパラメータが渡されるときに問題が発生する可能性があります。したがって、デバッグとチェックが必要です。スケジュールパラメータについては、「スケジュールパラメータを構成して使用する」をご参照ください。

  • 依存ノード: DataWorks は、サイクル間スケジュール依存関係が構成されている依存ノードを変換できません。依存ノードに対してサイクル間スケジュール依存関係が構成されている場合、DataWorks は依存関係を DataWorks 内のマップされた自動トリガーノードの同一サイクルスケジュール依存関係に変換します。同一サイクルスケジュール依存関係の構成方法については、「同一サイクルスケジュール依存関係を構成する」をご参照ください。

タスクタイプマッピングを構成する

このセクションの手順を実行して、DataWorks エクスポートツールをダウンロードし、タスクタイプマッピングを構成できます。

1. エクスポートツールをダウンロードします。

   エクスポートツールのソースコード をダウンロードします。

2. タスクタイプマッピングを構成します。

   エクスポートツールのディレクトリに移動し、lib、bin、および conf ディレクトリを表示します。conf ディレクトリの dataworks-transformer-config.json ファイルのマッピングを変更する必要があります。

   パラメータの説明:

   次のコードは、DolphinScheduler ノードを ODPS タイプのオブジェクトに変換するために、dataworks-transformer-config.json ファイルで構成する必要があるパラメータを示しています。

   {
     "format": "WORKFLOW",
     "locale": "zh_CN",
     "skipUnSupportType": true,
     "transformContinueWithError": true,
     "specContinueWithError": true,
     "processFilter": {
       "releaseState": "ONLINE",
       "includeSubProcess": true
     }
     "settings": {
       "workflow.converter.shellNodeType": "DIDE_SHELL",
       "workflow.converter.commandSqlAs": "ODPS_SQL",
       "workflow.converter.sparkSubmitAs": "ODPS_SPARK",
       "workflow.converter.target.unknownNodeTypeAs": "DIDE_SHELL",
       "workflow.converter.mrNodeType": "ODPS_MR",
       "workflow.converter.target.engine.type": "ODPS",
       "workflow.converter.dolphinscheduler.sqlNodeTypeMapping": {
         "POSTGRESQL": "POSTGRESQL",
         "MYSQL": "MYSQL"
       }
     },
     "replaceMapping": [
       {
         "taskType": "SHELL",
         "desc": "$[yyyyMMdd-1]",
         "pattern": "\$\[yyyyMMdd-1\]",
         "target": "\${dt}",
         "param": "dt=$[yyyyMMdd-1]"
       },
        {
         "taskType": "PYTHON",
         "desc": "$[yyyyMMdd-1]",
         "pattern": "\$\[yyyyMMdd-1\]",
         "target": "\${dt}",
         "param": "dt=$[yyyyMMdd-1]"
       }
     ]
   }

   パラメータ		説明
   format		データを移行する場合は、移行先ワークスペースが新バージョンの DataStudio のパブリックプレビューに参加しているかどうかによって、このパラメータを構成する必要があります。 移行先ワークスペースが新バージョンの DataStudio のパブリックプレビューに参加している場合は、このパラメータを WORKFLOW に設定する必要があります。 移行先ワークスペースが新バージョンの DataStudio のパブリックプレビューに参加していない場合は、このパラメータを SPEC に設定する必要があります。	必須。
   locale		言語環境。デフォルト値: zh_CN。
   skipUnSupportType		タスクタイプの変換中に、サポートされていないタイプをスキップするかどうかを指定します。有効な値: true および false。 このパラメータを true に設定すると、サポートされていないタイプはスキップされます。 このパラメータを false に設定すると、変換は失敗します。
   transformContinueWithError		タスクタイプの変換中にエラーが発生した場合に、変換を続行するかどうかを指定します。有効な値: true および false。 このパラメータを true に設定すると、変換は続行されます。 このパラメータを false に設定すると、変換は停止します。
   specContinueWithError		タスクタイプの変換が失敗した場合に、変換を続行するかどうかを指定します。有効な値: true および false。 このパラメータを true に設定すると、変換は続行されます。 このパラメータを false に設定すると、変換は停止します。
   processFilter	releaseState	状態が ONLINE のフローを処理するために使用されるフィルタ条件。	タスクタイプの変換中のフィルタリングがサポートされています。DolphinScheduler ノードを DataWorks にインポートする前にフィルタリングする場合は、このパラメータを構成する必要があります。
   	includeSubProcess	状態が ONLINE のサププロセスを処理するかどうかを指定します。
   settings	workflow.converter.shellNodeType	ソースシステムの Shell ノードが移行先システムでマップされるオブジェクトタイプ。値の例: DIDE_SHELL。	必須。
   	workflow.converter.commandSqlAs	移行先システムで SQL ノードを実行するために使用されるエンジンのタイプ。値の例: ODPS_SQL。
   	workflow.converter.sparkSubmitAs	移行先システムで Spark エンジンを使用して送信されたノードを実行するために使用されるエンジンのタイプ。値の例: ODPS_SPARK。
   	workflow.converter.target.unknownNodeTypeAs	不明なタイプのノードがマップされるデフォルトのオブジェクトタイプ。値の例: DIDE_SHELL。
   	workflow.converter.mrNodeType	移行先システムで MapReduce ノードを実行するために使用されるエンジンのタイプ。値の例: ODPS_MR。
   	workflow.converter.target.engine.type	デフォルトで使用されるエンジンのタイプ。値の例: ODPS。
   	workflow.converter.dolphinscheduler.sqlNodeTypeMapping	DolphinScheduler の SQL ノードのデータベースと移行先システムのデータベースのマッピング。値の例: "POSTGRESQL": "POSTGRESQL" "MYSQL": "MYSQL"
   replaceMapping	taskType	ルールを適用できるタスクタイプ。値の例: Shell または Python。	正規表現に基づいて一致するノードの内容を置き換えることができます。
   	desc	説明。これは情報フィールドであり、処理には関与しません。このパラメータは空のままにすることができます。
   	pattern	置き換える必要のある正規表現のパターン。値の例: $[yyyyMMdd-1]。
   	target	置き換え後に取得される移行先文字列。値の例: ${dt}。
   	param	移行先文字列に割り当てる値。たとえば、コード変数の値をノード変数 (dt=$[yyyyMMdd-1] など) に割り当てます。

DolphinScheduler からノードをエクスポートする

エクスポートツールを使用して、DolphinScheduler ジョブを ZIP ファイルとしてエクスポートできます。次のコードはサンプルコマンドを示しています。次のコードを使用する場合は、ビジネス要件に基づいてパラメータを構成する必要があります。次のコマンドを実行した後、DolphinScheduler のリソースは、-f で指定されたファイル名を使用してカレントディレクトリに保存されます。

python3 bin/reader.py \
-a dolphinscheduler \
-e http://xxx.xxx.xxx.xxx:12345 \
-t {token} \
-v 1.3.9  \
-p 123456,456256 \
-f project_dp01.zip\
-sr false;