このトピックでは、AnalyticDB for MySQL 3.0 Writer でサポートされるデータ型およびパラメーター、およびコードレスユーザーインターフェイス (UI) とコードエディターを使用した設定方法について説明します。

AnalyticDB for MySQL 3.0 Writerを設定する前に、接続を設定しておく必要があります。

データ型

下表に、 AnalyticDB for MySQL 3.0 Writer でサポートされるデータ型を示します。
カテゴリー AnalyticDB for MySQL 3.0 のデータ型
整数 INT、INTEGER、TINYINT、 SMALLINT、および BIGINT
浮動小数点 FLOAT、DOUBLE、DECIMAL
文字列 VARCHAR
日付と時刻 DATE、DATETIME、TIMESTAMP、および TIME
ブール値 BOOLEAN

パラメーター

パラメーター 説明 必須 / 任意 デフォルト値
datasource 接続の名前を設定します。 追加された接続の名前と同じである必要があります。 コードエディターで接続を追加できます。 必須 なし
table 書き込み先テーブルの名前を設定します。 必須 なし
writeMode 書き込みモードを設定します。 設定可能な値は、insert into および replace into です。
  • INSERT INTO:主キーまたは一意のインデックスの競合が発生した場合、データは競合している行に書き込まれず、ダーティデータと見なされます。
  • REPLACE INTO:主キーの競合や一意のインデックスの競合が発生していない場合、INSERT INTOと同様に動作します。 競合が発生すると、元の行が削除され、新しい行が挿入されます。 つまり、元の行のすべてのフィールドが置き換えられます。
任意 insert into
column 書き込み先テーブルでデータが書き込まれる列を指定します。 複数入力する場合は、コンマ (,) で区切ります (例:"column": ["id", "name", "age"])。 データを書き込みテーブルのすべての列に書き込むには、値をアスタリスク (*) に設定します。 その場合、"column":["*"] のように設定します。
フィールド名に select が使用されている場合、フィールド名をグレイヴアクセント (`) で囲みます。 (例:`item_select_no`)
必須 なし
preSql 同期ノードを実行する前に実行する SQL ステートメントを設定します。 たとえば、データ同期の前に古いデータを消去できます。 現在、コードレス UI では 1 つの SQL ステートメントのみ実行でき、コードエディターでは複数の SQL ステートメントを実行できます。
コードエディターで複数の SQL ステートメントを指定した場合でも、それらが同じトランザクションで実行されることは保証されません。
任意 なし
postSql 同期ノードを実行した後に実行する SQL ステートメントを設定します。 たとえば、データ同期後にタイムスタンプを追加できます。 現在、コードレス UI では 1 つの SQL ステートメントのみ実行でき、コードエディターでは複数の SQL ステートメントを実行できます。
コードエディターで複数の SQL ステートメントを指定した場合でも、それらが同じトランザクションで実行されることは保証されません。
任意 なし
batchSize 一度に書き込まれるデータレコードの数を設定します。 このパラメーターを設定すると、Data Integration と AnalyticDB for MySQL 3.0 データベース間のネットワーク上の通信が大幅に減少し、スループットが向上します。 ただし、値が大きすぎると、データ同期プロセス中に OOM (メモリ不足) エラーが発生する可能性があります。 任意 1024

コードレス UI を使用した AnalyticDB for MySQL 3.0 Writer の設定

  1. 接続を設定します。

    同期ノードの接続元および接続先を設定します。

    パラメーター 説明
    接続 前述の datasource パラメーターを設定します。 接続タイプを選択し、DataWorks に設定されている接続の名前を入力します。
    テーブル 前述の table パラメーターを設定します。
    書き込み前実行ステートメント 前述の preSqlパラメーターを設定します。 同期ノードを実行する前に実行する SQL ステートメントを設定します。
    書き込み後実行ステートメント 前述の postSql パラメーターを設定します。 同期ノードを実行した後に実行する SQL ステートメントを設定します。
    プライマリキー違反の解決方法 前述の writeMode パラメーターを設定します。 想定される書き込みモードを選択します。
    書き込みごとのデータレコード 一度に書き込まれるデータレコードの数を設定します。 前述の batchSize パラメーターを設定します。 このパラメーターは、wirteMode が insert to に設定されている場合のみ有効です。
  2. フィールドのマッピング(前述の column パラメーター)を設定します。 左側のソーステーブルと右側の書き込み先テーブルのフィールドは 1 対 1 のマッピングです。
  3. チャンネル制御ポリシーを設定します。チャンネル
    パラメーター 説明
    最大同時実行数の期待値 同期ノード内のデータストレージにデータを読み書きする同時スレッド数の上限を設定します。 コードレス UI でノードの同時実行数を設定できます。
    帯域幅スロットリング 帯域幅スロットリングを有効にするかどうかを指定します。 帯域幅スロットリングを有効化し、最大転送速度を設定すると、ソース側の読み取り作業負荷が過剰になることを回避できます。 帯域幅スロットリングを有効化し、最大伝送速度を適切な値に設定することを推奨します。
    許容されるダーティーデータレコード 許容されるエラーまたはダーティーデータレコード数の上限を設定します。
    リソースグループ ノードが実行されるサーバーを指定します。 デフォルトのリソースグループで多数のタスクが実行されている場合、一部のタスクはリソース不足のために実行が遅くなる場合があります。 このような場合、データ統合用の専用リソースグループを購入するか、またはカスタムリソースグループを追加することを推奨します。 詳細については、「DataWorks の排他的リソース」および「カスタムリソースグループの追加」をご参照ください。

コードエディターを使用した AnalyticDB for MySQL 3.0 Writer の設定

以下のコードでは、ノードが AnalyticDB for MySQL 3.0 データベースにデータを書き込むよう設定されます。 パラメーターの詳細については、前述のパラメーターの説明をご参照ください。
{
    "type": "job",
    "steps": [
        {
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "analyticdb_for_mysql", // ライターのタイプ。
             "parameter": {
                "postSql": [], // 同期ノード実行後に実行される SQL ステートメント。
                 "tableType": null, // 予約フィールド。 規定値: null。
                 "datasource": "hangzhou_ads", // 接続名。
                 "column": [ // データが書き込まれる宛先テーブルの列。
                     "id",
                    "value"
                ],
                "guid": null,
                "writeMode": "insert", // 書き込みモード。 詳細については、writeMode パラメーターの説明をご参照ください。
                 "batchSize": 2048, // 一度に書き込むデータレコードの数。 詳細については、batchSize パラメーターの説明をご参照ください。
                 "encoding": "UTF-8", // エンコード形式。
                 "table": "t5", // ターゲットテーブル名。
                 "preSql": [] // 同期ノード実行前に実行される SQL ステートメント。
             },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "version": "2.0", // バージョン番号。
     "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {
            "record":"0" // 許可されるダーティデータレコードの最大数。
         },
        "speed": {
            "concurrent": 2, // 同時実行スレッド数の上限。
             "throttle": false // 帯域幅スロットリングを使用するかどうかを指定。 値が false の場合は、帯域幅が調整されていないことを示します。 値が trueの場合は、帯域幅が調整されていることを示します。  最大伝送速度は、このパラメーターを true に設定した場合にのみ有効になります。
         }
    }
}