$bizdate 変数による増分データ同期の設定 - DataWorks - Alibaba Cloud - DataWorks

スケジューリングパラメータは、データ同期ノード設定で動的な値として機能します。静的な日付、テーブル名、またはフィールド値を、ノードのデータタイムスタンプから実行時に算出される値に置き換えます。このトピックでは、増分データの同期、動的なテーブル名またはファイル名の処理、宛先フィールドの定義、履歴データのデータバックフィルの 4 つの一般的なシナリオについて説明します。

事前準備

バッチ同期ノードの設定タブの上部ツールバーで [実行] または [パラメータを指定して実行] をクリックする前に、ノードコードで参照される変数に手動で定数を割り当ててください。Data Integration は、開発環境でのバッチ同期ノードのスモークテストをサポートしていません。

スケジューリングパラメータが期待どおりに置き換えられることを確認するには、次のいずれかの方法を使用してください。

同じスケジューリングパラメータを使用する SQL ノードでスモークテストを実行します。SQL ノードが合格した場合、パラメータは正しく設定されています。詳細については、「スケジューリングパラメータの設定と使用」をご参照ください。
開発環境でノードをオペレーションセンターにコミットします。

シナリオの概要

シナリオ	サポートされる Reader プラグイン (例)	詳細
増分データの同期	MySQL Reader、LogHub Reader、Kafka Reader	シナリオ1およびシナリオ4
動的な名前を持つテーブルまたはファイルからのデータ同期	Object Storage Service (OSS) Reader、FTP Reader、MySQL Reader	シナリオ2
宛先フィールドへの定数または時間ベースの値の割り当て	データソースによって異なります	シナリオ3

シナリオ1：増分データの同期

フィルター条件内のスケジューリングパラメータは、動的な時間境界として機能します。実行時に、各表現がノードのデータタイムスタンプから派生した実際の日付に置き換えられます。これにより、各実行ではターゲットの時間枠内に生成されたデータのみを読み取り、完全な再読み込みを回避できます。フィルター条件の構文がソースでサポートされていることを確認してください。

詳細については、「シナリオ：増分オフラインデータ同期タスクの設定」をご参照ください。

重要

フィルター条件で使用される時間範囲は、左閉右開区間です。

例1：10分ごとに LogHub データを T-1 の MaxCompute パーティションに同期

ノードは 10 分ごとにスケジューリングインスタンスを生成します。$bizdate はデータ同期ノードのデータタイムスタンプを指定し、パーティションフィルターの式は実行時に対応する 10 分間のウィンドウ境界になります。

タスク設定の要点: [データソース] を [LogHub] に、[ログ開始時間] を ${startTime} に、[ログ終了時間] を ${endTime} に設定します。[データ宛先] を [MaxCompute (ODPS)] に、[パーティション] を pt = ${bizdate} に、[書き込みモード] を [上書き挿入] に設定します。スケジューリング設定では、次のパラメーターを定義します。

bizdate=$bizdate
startTime=${yyyymmddh24miss-10/24/60}
endTime=${yyyymmddh24miss}

[スケジュール周期]を[日]に、[実行時刻]を [00:05] に設定します。

例2：毎日 00:00 に前日の LogHub データを T-1 の MaxCompute パーティションに同期

フィルター条件は $bizdate を使用して、前日に生成されたすべてのデータをキャプチャします。時間範囲は左閉右開区間です。

オフラインバッチ同期タスクで、スケジューリングパラメータをデータソースに関連付けるには、次の設定を完了してください。

右側のスケジューリング設定パネルの上部に 3 つのスケジューリングパラメータを追加します：startTime=${yyyymmdd}、endTime=${yyyymmdd+1}、bizdate=${yyyymmdd}。
[データソース] (Logstore) で、[ログ開始時刻] を ${startTime}000000 に、[ログ終了時刻] を ${endTime}000000 に、[バッチサイズ] を 256 に設定します。
[データ宛先] (MaxCompute)：[パーティション] を pt=${bizdate} に、[クリーニングルール] を [挿入して上書き] に設定します。
スケジューリング設定: [スケジューリング周期] を [日] に、[スケジュール時刻] を 00:00 (cron 式: 00 00 00 * * ?) に、[インスタンス生成] を [T+1 (翌日に生成)] に設定します。

シナリオ2：動的な名前を持つテーブルまたはファイルからのデータ同期

スケジューリングパラメータをソーステーブル名またはファイルパスに直接埋め込むことで、ノードは各実行時に正しいテーブルまたはファイルから自動的に読み取ります。

説明

一部の Reader プラグインでは、コードエディターを使用して動的な名前のスケジューリングパラメータを設定する必要があります。設定で変数が使用されている場合、データプレビューはサポートしていません。

例1：日付ベースの名前を持つ OSS ファイルを MaxCompute に同期

ソースオブジェクトパスをスケジューリングパラメータで設定することで、ノードは毎日正しい日付のフォルダーから読み取り、対応するパーティションにデータを書き込みます。

データ同期タスクの [データソースの選択] ステップで、次のように設定します。

[データソース]: OSS データソースを選択します。テキストタイプを CSV、ファイルパスを /xx/user_log_${var1}.txt、列区切り文字をコンマ、エンコーディングを UTF-8 に設定します。${var1} は、ファイル名の日付に動的に一致させるために使用されるスケジューリングパラメーターです。
[データ宛先]: MaxCompute (ODPS) データソースを選択します。ターゲットテーブルを ods_raw_log_d に、パーティションを dt=${var1} に、書き込みモードを [上書き挿入] に、[空の文字列を NULL 値に変換] を [いいえ] に設定します。

[スケジューリングパラメーター] 設定パネルで、次のパラメーターを設定します。

パラメータ名 bizdate、パラメータ値 $bizdate、ソース：手動で追加。
パラメータ名 var1、パラメータ値 ${yyyymmdd-1}、ソース：手動で追加。

例2：日付ベースの名前を持つ MySQL テーブルを MaxCompute に同期

ソーステーブル名をスケジューリングパラメータで設定することで、ノードは毎日正しい日付のテーブルから読み取ります。コードエディターの設定とスケジューリングパラメータの設定を以下に示します。


"version": "2.0",
"steps": [
    {
        "stepType": "mysql",
        "parameter": {
            "envType": 0,
            "useSpecialSecret": false,
            "column": [
                "id"
            ],
            "tableComment": "",
            "connection": [
                {
                    "datasource": "xc_workshop_public",
                    "table": [
                        "person_${var1}"
                    ]
                }
            ],
            "where": "",
            "splitPk": "id",
            "encoding": "UTF-8"
        },
        "name": "Reader",
        "category": "reader"
    },
    {
        "stepType": "odps",
        "parameter": {
            "partition": "ds=${var1}",
            "truncate": true,
            "datasource": "odps_first",
            "isSupportThreeModel": false,
            "column": [
                "id"

[パラメーター] 設定エリアで、スケジューリングパラメーターの設定を確認します。パラメーター bizdate の値は $bizdate (手動で追加) で、パラメーター var1 の値は ${yyyymmdd-1} (コード解析による自動生成) です。設定が完了したら、[パラメーターのプレビュー] をクリックして、パラメーター置換結果をプレビューします。

シナリオ3：宛先フィールドの定義

フィールドマッピングを設定する際、ノードのデータタイムスタンプから派生した定数を宛先フィールドに割り当てます。システムは、スケジューリングパラメータの値に基づいて、実行ごとにそのフィールドを自動的に更新します。サポートはデータソースのタイプによって異なります。

たとえば、次の設定では、データタイムスタンプを宛先テーブルの ds フィールドに割り当て、毎日更新されます。

[フィールドマッピング] エリアで、宛先フィールドにタイプが [定数] の行を追加し、値を ${bizdate} に設定します。右側の [スケジューリング] パネルのパラメーターエリアで、パラメーター bizdate = $bizdate を追加して、このスケジューリングパラメーターがフィールドマッピングで定数値として有効になるようにします。

シナリオ4：履歴データの同期

スケジューリングパラメータは、ノードのデータタイムスタンプとスケジューリングパラメータの値の形式に基づいて、特定の値に自動的に置き換えられます。これにより、ノードスケジューリングの動的なパラメータ設定が可能になります。データバックフィルの実行では、各パラメータはバックフィル操作で指定されたデータタイムスタンプを使用して置き換えられます。

オペレーションセンターのデータバックフィル機能を使用して、指定された過去の時間範囲のデータを生成してください。詳細については、「データバックフィルインスタンスのO&M」をご参照ください。

例：増分 MySQL データを時間でパーティション分割された MaxCompute テーブルにバックフィル

次の例は、データバックフィルを使用して、MySQL ソースからの増分データを MaxCompute の指定された時間パーティションに書き込む方法を示しています。

オフラインバッチ同期タスクでは、bizdate スケジューリングパラメータを次の 3 つの場所で設定する必要があります。

データソースの [データフィルター] 条件で、${bizdate} を参照します。例: STR_TO_DATE('${bizdate}','%Y%m%d') <= gmt_modify_time AND gmt_modify_time < DATE_ADD(STR_TO_DATE('${bizdate}','%Y%m%d'), interval 1 day)
データ宛先の[パーティション] フィールドで、値をpt=${bizdate}に設定し、クリーニングルールを[Insert Overwrite]に設定します。
右側の [スケジューリング] パネルの [パラメーター] エリアで bizdate=$bizdate を設定し、[インスタンス生成] を [T+1 (翌日に生成)] に設定します。