DataWorks:バッチ同期に関するよくある質問

• 以前に接続テストが成功した場合は、再度テストして、リソースグループとデータベースが現在接続されていること (およびデータベース側で変更が行われていないこと) を確認してください。

• 接続テストに合格したリソースグループが、タスクの実行に使用されるリソースグループと同じであるか確認してください。

  タスクが使用するリソースグループを確認します：

  • タスクがデフォルトのリソースグループで実行される場合、ログには次の情報が含まれます：running in Pipeline[basecommon_ group_xxxxxxxxx]

  • タスクがデータ統合専用リソースグループで実行される場合、ログには次の情報が含まれます：running in Pipeline[basecommon_S_res_group_xxx]

  • タスクがサーバーレスリソースグループで実行される場合、ログには次の情報が含まれます：running in Pipeline[basecommon_Serverless_res_group_xxx]

• タスクが早朝のスケジューリング中に時々失敗し、再実行後に成功する場合は、エラーが発生した時点でのデータベースの負荷を確認してください。

バッチ同期タスクが断続的に失敗する場合、原因はホワイトリストの設定が不完全である可能性があります。データベースのホワイトリストが完全に設定されているか確認してください。

データ統合専用リソースグループを使用する場合：

• 以前にデータ統合専用リソースグループの Elastic Network Interface (ENI) の IP アドレスをデータソースのホワイトリストに追加し、その後リソースグループがスケールアウトされた場合は、スケールアウトされたリソースグループの ENI の IP アドレスを含めるようにデータソースのホワイトリストを更新してください。

• リソースグループがスケールアウトするたびにホワイトリストを更新する必要がないように、データ統合専用リソースグループに関連付けられている vSwitch の CIDR ブロックをデータベースのホワイトリストとして追加することを推奨します。詳細については、「ホワイトリストの追加」をご参照ください。

サーバーレスリソースグループを使用する場合：「サーバーレスリソースグループのネットワーク接続」を参照して、リソースグループのホワイトリスト設定を確認し、ネットワークが正しく設定されていることを確認してください。

ホワイトリストが正しく設定されている場合、データベースの負荷が高すぎて接続が中断される可能性があるかどうかを確認してください。

• 考えられる原因：

  同時実行数が高すぎるため、リソースが不足しています。

• 解決策：

  バッチ同期タスクの同時実行数を減らします。

  • ウィザードモードでバッチ同期タスクを設定する場合、チャネル制御の タスクの希望最大同時実行数 の値を減らします。詳細については、「ウィザードモードでのチャネル制御の設定」をご参照ください。

  • スクリプトモードでバッチ同期タスクを設定する場合、チャネル制御の concurrent パラメーターの値を減らします。詳細については、「スクリプトモードでのチャネル制御の設定」をご参照ください。

このエラーを解決するには：

1. プラグイン設定が batchsize や maxfilesize などのパラメーターをサポートしている場合は、対応する値を減らします。

   各プラグインが上記のパラメーターをサポートしているかどうかを確認できます。「サポートされているデータソースと Reader/Writer」トピックに移動し、対応するプラグインをクリックしてパラメーターの詳細を表示します。

2. 同時実行数を減らします。

   • ウィザードモードでバッチ同期タスクを設定する場合、チャネル制御の タスクの希望最大同時実行数 の値を減らします。詳細については、「ウィザードモードでのチャネル制御の設定」をご参照ください。

   • スクリプトモードでバッチ同期タスクを設定する場合、チャネル制御の concurrent パラメーターの値を減らします。詳細については、「スクリプトモードでのチャネル制御の設定」をご参照ください。

3. OSS ファイルなどのファイルを同期している場合は、読み取るファイル数を減らします。

4. タスク設定の 実行中のリソース セクションで、リソース使用量 (CU) の値を適切に増やします。他の実行中のタスクに影響を与えないように、CU の値は慎重に設定してください。

• エラーメッセージ：Error updating database. Cause: com.mysql.jdbc.exceptions.jdbc4.MySQLIntegrityConstraintViolationException: Duplicate entry 'cfc68cd0048101467588e97e83ffd7a8-0' for key 'uk_uk_op'。

• 考えられる原因：データ統合では、同じノードの異なるインスタンス (つまり、同じ JSON 設定を持つ同期タスク) を同時に実行することはできません。たとえば、同期タスクが 5 分間隔で実行され、上流の遅延により 00:00 のインスタンスと 00:05 のインスタンスの両方が 00:05 にトリガーされた場合、いずれかのインスタンスは開始できません。これは、タスクインスタンスがまだ実行中にデータをバックフィルしたり、インスタンスを再実行したりする場合にも発生する可能性があります。

• 解決策：インスタンスの実行時間をずらします。時間単位または分単位の間隔でスケジュールされたタスクについては、現在のインスタンスが前のサイクルのインスタンスが完了した後にのみ開始されるように、自己依存を設定することを推奨します。レガシー DataStudio での設定については、「自己依存」をご参照ください。新しい DataStudio での設定については、「自己依存の設定」をご参照ください。

• エラーの詳細：データ同期タスク中に、タスクが次のエラーで失敗します：MongoDBReader$Task - operation exceeded time limitcom.mongodb.MongoExecutionTimeoutException: operation exceeded time limit。

• 考えられる原因：完全なデータプルが大きすぎます。

• 解決策：

  • 同時実行数を増やします。

  • BatchSize を減らします。

  • Reader パラメーターセクションに cursorTimeoutInMs 設定を追加し、3600000 ms などの大きな値を設定します。

• 読み取りエラー

  • 症状：

    データの読み取り時に、次のエラーが発生します：Communications link failure The last packet successfully received from the server was 7,200,100 milliseconds ago. The last packet sent successfully to the server was 7,200,100 milliseconds ago. - com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link failure

  • 考えられる原因：

    データベースが SQL クエリをゆっくり実行するため、MySQL の読み取りタイムアウトが発生します。

  • 解決策：

    • where フィルター条件が設定されているかどうかを確認し、フィルター列にインデックスが付けられていることを確認します。

    • ソーステーブルにデータが多すぎるかどうかを確認します。多すぎる場合は、タスクを複数のタスクに分割します。

    • ログを調べて、ブロッキングの原因となった SQL ステートメントを見つけ、データベース管理者に相談して問題を解決します。

• 書き込みエラー

  • 症状：

    データの書き込み時に、次のエラーが発生します：Caused by: java.util.concurrent.ExecutionException: ERR-CODE: [TDDL-4614][ERR_EXECUTE_ON_MYSQL] Error occurs when execute on GROUP 'xxx' ATOM 'dockerxxxxx_xxxx_trace_shard_xxxx': Communications link failure The last packet successfully received from the server was 12,672 milliseconds ago. The last packet sent successfully to the server was 12,013 milliseconds ago. More...

  • 考えられる原因：

    スロークエリが SocketTimeout を引き起こします。TDDL 接続のデフォルトの SocketTimeout は 12 秒です。SQL ステートメントが MySQL での実行に 12 秒以上かかると、4614 エラーが報告されます。このエラーは、データ量が大きい場合やサーバーがビジーな場合に時々発生する可能性があります。

  • 解決策：

    • データベースが安定するまで待ってから、同期タスクを再実行します。

    • データベース管理者に連絡して、タイムアウト値を調整します。

考えられる原因 2：データ統合タスクの実行リソースを待機している

解決策 2：ログに長時間の WAIT ステータスが表示される場合、現在のタスクで使用されているデータ統合専用リソースグループには、タスクを実行するのに十分な利用可能な同時実行数がありません。原因と解決策の詳細については、「リソースグループの同時実行数の問題のトラブルシューティング」をご参照ください。

• シナリオ例

  実行された SQL は次のとおりです：

  SELECT bid,inviter,uid,createTime FROM `relatives` WHERE createTime>='2016-10-2300:00:00' AND reateTime<'2016-10-24 00:00:00';

  実行は 2016-10-25 11:01:24.875 に開始され、結果は 2016-10-25 11:11:05.489 に返され始めました。同期プログラムはデータベースが SQL クエリ結果を返すのを待ち、MaxCompute は実行を進める前に長時間待機する必要がありました。

• 根本原因分析

  WHERE 句の createTime 列にインデックスがないため、全表スキャンが発生します。

• 解決策

  where 句では、パフォーマンスを向上させるためにインデックス付きの列を使用することを推奨します。必要に応じてインデックスを追加することもできます。

レガシー DataStudio：

DataStudio のバッチ同期タスク詳細ページでデバッグに使用するリソースグループを変更できます。また、オペレーションセンター でスケジューリング中に使用されるデータ統合タスク実行リソースグループを変更することもできます。詳細については、「データ統合リソースグループの切り替え」をご参照ください。

新しい DataStudio：

DataStudio でデータ統合タスクのデバッグに使用するリソースグループを変更できます。また、オペレーションセンター でスケジューリング中に使用されるデータ統合タスク実行リソースグループを変更することもできます。詳細については、「データ統合リソースグループの切り替え」をご参照ください。

ダーティデータ：例外が原因で送信先データソースへの書き込みに失敗したレコード。

タスクは実行中にダーティデータレコードの数を累積します。数が設定されたダーティデータのしきい値を超えると、タスクは直ちに終了します。

• データ保持：タスクが終了する前に送信先に正常に書き込まれたデータは保持されます。ロールバックは実行されません。

• ゼロトレランスポリシー：ダーティデータのしきい値が 0 に設定されている場合、システムはゼロトレランスポリシーを採用します。これは、最初のダーティデータレコードを検出すると、タスクが直ちに失敗して停止することを意味します。

• エラーメッセージ：

  データに絵文字が含まれている場合、同期中にダーティデータエラーが発生することがあります：[13350975-0-0-writer] ERROR StdoutPluginCollector - Dirty data {"exception":"Incorrect string value: '\\xF0\\x9F\\x98\\x82\\xE8\\xA2...' for column 'introduction' at row 1","record":[{"byteSize":8,"index":0,"rawData":9642,"type":"LONG"}],"type":"writer"} 。

• 考えられる原因：

  • データベースのエンコーディングが utf8mb4 に設定されていないため、絵文字の同期時にエラーが発生します。

  • ソースデータ自体に文字化けが含まれています。

  • データベースとクライアントのエンコーディングが一致していません。

  • ブラウザのエンコーディングが異なるため、プレビューの失敗や文字化けが発生します。

• 解決策：

  文字化けの原因に応じて適切な解決策を選択してください：

  • 元のデータに文字化けが含まれている場合は、同期タスクを実行する前にデータを修正してください。

  • データベースとクライアントのエンコード形式が一致しない場合は、まずエンコード形式を変更してください。

  • ブラウザのエンコーディングがデータベースまたはクライアントのエンコーディングと一致しない場合は、データをプレビューする前にエンコード形式を統一してください。

  次のことを試すことができます：

  1. JDBC 形式で追加されたデータソースの場合、utf8mb4 を次のように変更します：jdbc:mysql://xxx.x.x.x:3306/database?com.mysql.jdbc.faultInjection.serverCharsetIndex=45。

  2. インスタンス ID で追加されたデータソースの場合、データベース名の後に次を追加します：database?com.mysql.jdbc.faultInjection.serverCharsetIndex=45。

  3. データベースのエンコード形式を utf8mb4 に変更します。たとえば、RDS コンソールで RDS データベースのエンコード形式を変更します。

     説明

     RDS データソースのエンコード形式を設定するコマンド：set names utf8mb4。RDS データベースのエンコード形式を確認するコマンド：show variables like 'char%'。

• シナリオ：MaxCompute (ODPS) にデータを同期すると、ソースフィールドが 8 MB を超えているため、タスクが失敗するか、書き込まれたデータが切り捨てられます。

• 原因：MaxCompute を送信先として使用する同期タスクの場合、MaxCompute (ODPS) Writer はフィールドごとに 8 MB のサイズ制限を適用します。

• 解決策：

  • MaxCompute (ODPS) Writer 設定の高度な設定で、長すぎるフィールドの処理ポリシー (overLengthRule) を設定して、大きすぎるフィールドの処理方法を指定します：フィールドを 8 MB に切り捨てる、フィールドを NULL に設定する、または切り捨てずにフィールドをそのまま書き込む。

  • ファイルやログなどのラージオブジェクトを含むシナリオでは、生データを Object Storage Service (OSS) に保存し、データベースには URL のみを保持します。その後、DataWorks を使用して、ラージオブジェクト自体の代わりに URL を同期します。

  • 前処理ステップとして、ソースで大きすぎるフィールドを複数の小さなサブフィールドに分割します。たとえば、7 MB のセグメントに分割します。サブフィールドを個別に同期し、送信先で再度連結します。

送信先テーブルを作成する際、DataWorks はソーステーブルから列名、データの型、コメントのみを保持します。デフォルト値、制約 (非 NULL 制約やインデックスを含む) は保持しません。

バッチ同期タスクは、複合主キーを分割キーとして使用することをサポートしていません。

• シナリオ：増分同期タスクが、設定された分割キー (splitPk) 列が無効であることを示すエラーで失敗します。

• 考えられる原因：分割キーの設定が要件を満たしていません。たとえば、複数の列が分割キーとして設定されている、設定された列のデータの型がサポートされていない、または設定された列がテーブルに存在しないなどです。

• 解決策：

  1. ソーステーブルのマッピングを更新して、システムが自動的に提案する分割キー列を表示します。

  2. 分割キーとして単一の列のみが設定されており、その列のデータの型が整数型であることを確認してください。このドキュメントで前述したように、splitPk は整数ベースのデータシャーディングのみをサポートし、文字列、浮動小数点、日付、その他の型はサポートしていません。

  3. 自動的に提案された列がこれらの条件を満たさない場合は、分割キーを要件を満たす整数型の単一の列に手動で変更します。

分割キーが同期パフォーマンスに与える影響の詳細については、「長時間実行されるバッチ同期タスクのトラブルシューティング方法」をご参照ください。

データ同期後にデータ品質の問題が発生した場合は、「同期後のデータ品質問題のトラブルシューティング」を参照して詳細なトラブルシューティングを行ってください。

Q：「Task have SSRF attacks」というエラーの対処方法

原因：クラウドのセキュリティを確保するため、DataWorks はタスクがパブリック IP アドレスを介して内部クラウドネットワークアドレスにアクセスすることを禁止しています。プラグイン設定 (HTTP Reader など) の URL が内部 IP アドレスまたは VPC ドメイン名を指している場合、このセキュリティチェックがトリガーされます。

正しいアプローチ：

• 設定で 10.x.x.x や 192.168.x.x などの内部 IP アドレスを使用しないでください。

• サーバーレスリソースグループ (推奨) またはデータ統合専用リソースグループを購入し、VPC 接続を設定して内部データソースに安全にアクセスします。

解決策：内部データソースにアクセスするタスクについては、共有リソースグループの使用を停止し、安全なサーバーレスリソースグループ (推奨) またはデータ統合専用リソースグループに切り替えます。

同期タスクをスクリプトモードに切り替え、タスク設定ページの setting セクションに次の設定を追加します：

"common": {
  "column": {
    "dateFormat": "yyyyMMdd",
    "datetimeFormatInNanos": "yyyyMMdd HH:mm:ss.SSS"
  }
}

ここで：

• dateFormat は、ソースの DATE (時間なし) 型データをテキストに変換する際に使用される日付形式を指定します。

• datetimeFormatInNanos は、ソースの DATETIME/TIMESTAMP (時間あり) 型データをテキストに変換する際に使用される日付形式を指定します。ミリ秒までの精度を指定できます。

1. 定数を入力できます。値は 'abc' や '123' のように単一引用符で囲む必要があります。

2. '${bizdate}' のようなスケジューリングパラメーターを使用できます。スケジューリングパラメーターの使用方法については、「スケジューリングパラメーターの設定」をご参照ください。

3. pt のように、同期するパーティション列を入力できます。

4. 入力された値が解析できない場合、型は 'Custom' と表示されます。

5. ODPS 関数はサポートされていません。

6. 手動で追加された列が Custom と表示される場合 (たとえば、MaxCompute のパーティション列やデータプレビューに表示されない LogHub の列など)、実際のタスク実行には影響しません。

列マッピングリストで、ソーステーブルの列の下にある 行の追加 または フィールドの追加 をクリックし、パーティション列名 (pt など) を入力して、送信先テーブルの列へのマッピングを設定します。

MaxCompute Writer を設定することで、MaxCompute 自体がサポートしていない列のフィルタリング、並べ替え、null 埋め込み操作を実装できます。たとえば、すべての列をインポートするには、"column": ["*"] と設定します。

MaxCompute テーブルに a、b、c の 3 つの列があり、c と b の列のみを同期したい場合は、列リストを "column": ["c","b"] と設定します。これは、Reader からの最初の列と 2 番目の列が MaxCompute テーブルの c 列と b 列にインポートされ、MaxCompute テーブルに新しく挿入された a 列が null に設定されることを意味します。

データの書き込みの信頼性を確保し、余分な列データが失われることによるデータ品質の問題を回避するため、MaxCompute Writer は余分な列が書き込まれた場合にエラーを報告します。たとえば、MaxCompute テーブルに a、b、c の列があり、MaxCompute Writer が 3 つ以上の列を書き込もうとすると、エラーを報告します。

• シナリオ：MaxCompute (ODPS) への書き込みを行うバッチ同期タスクが失敗し、トラブルシューティングの結果、送信先テーブルに JSON 型の列が含まれていることが判明しました。

• 考えられる原因：MaxCompute Writer は、すべての場合において JSON 型の送信先列への書き込みをサポートしていない可能性があります。

• 解決策：送信先の MaxCompute テーブルのスキーマを確認し、JSON 型の列が含まれているかどうかを確認します。含まれている場合は、次のいずれかの方法を試してください：

  • MaxCompute 側で列の型を STRING に変更します。たとえば、ALTER TABLE ADD COLUMN または同等のスキーマ変更を実行します。

  • フィールドマッピングから JSON 型の列を除外し、同期中に書き込まれないようにします。

MaxCompute Writer は、最終レベルのパーティションへの書き込みのみをサポートし、列に基づくパーティションルーティングはサポートしていません。テーブルに 3 つのレベルのパーティションがある場合、パーティション設定で正確な第 3 レベルのパーティションを指定する必要があります。たとえば、第 3 レベルのパーティションにデータを書き込むには、pt=20150101, type=1, biz=2 のように設定します。pt=20150101, type=1 や pt=20150101 のようには設定できません。

MaxCompute Writer は、"truncate": true を設定することで書き込みのべき等性を保証します。書き込みが失敗して再実行されると、MaxCompute Writer は以前のデータをクリアして新しいデータをインポートし、再実行ごとにデータの整合性を確保します。実行中に他の例外によりタスクが中断された場合、データの原子性は保証されません。データはロールバックされたり、自動的に再実行されたりしません。べき等性の特徴を活用してタスクを再実行し、データの完全性を確保してください。

• エラーメッセージ：

  Code:DATAX_R_ODPS_005:Failed to read ODPS data, Solution:[Please contact the ODPS administrator]. RequestId=202012091137444331f60b08cda1d9, ErrorCode=StatusConflict, ErrorMessage=The download session is expired.

• 考えられる原因：

  バッチ同期が MaxCompute データを読み取る際、MaxCompute の tunnel コマンドを使用してデータをアップロードおよびダウンロードします。Tunnel セッションのサーバー側の有効期間は 24 時間です。したがって、バッチ同期タスクが 24 時間以上実行されると失敗します。tunnel の詳細については、「Tunnel の概要」をご参照ください。

• 解決策：

  バッチ同期タスクの同時実行数を増やし、データ量を適切に計画して、タスクが 24 時間以内に完了するようにします。

• エラーメッセージ：

  Code:[OdpsWriter-09], Description:[Failed to write data to the ODPS destination table.]. - Failed to write block:0 to the ODPS destination table, uploadId=[202012081517026537dc0b0160354b]. Please contact the ODPS administrator for assistance. - java.io.IOException: Error writing request body to server。

• 考えられる原因：

  • 考えられる原因 1：データの型の例外、つまりソースデータが ODPS のデータの型仕様に準拠していない。たとえば、ODPS の decimal(18,10) データ型に値 4.2223 を書き込む場合。

  • 考えられる原因 2：ODPS ブロックまたは通信の例外。

• 解決策：

  データの型を変換し、データの型仕様に準拠したデータを使用します。

• シナリオ：MaxCompute への全データベースのバッチ同期タスクが ErrorCode=MethodNotAllowed, ErrorMessage=cdc mode not supported で失敗します。

• 考えられる原因：送信先の MaxCompute テーブルのトランザクションまたは変更データキャプチャ (CDC) 属性が、同期タスクが使用する書き込みモードと一致しない可能性があります。

• 解決策：送信先テーブルが、現在の同期タスクの書き込みモードと互換性のない CDC またはトランザクション属性で作成されているかどうかを確認します。テーブルの属性が不明な場合は、テクニカルサポートに連絡して詳細を確認してください。

• シナリオ：MaxCompute を送信先としてバッチ同期タスクを設定する際、送信先設定の Tunnel リソースグループセレクターに、購入したサーバーレスリソースグループが表示されません。

• 原因：Tunnel リソースグループと、サーバーレスリソースグループなどの同期タスクを実行するリソースグループは、目的が異なる 2 つの独立した設定項目です。Tunnel リソースグループは MaxCompute のデータアップロードおよびダウンロード転送にのみ使用され、デフォルトでパブリック転送リソース、つまり MaxCompute Tunnel クォータを使用します。タスクを実行するリソースグループは、ソースの読み取り、データ処理、スケジューリングなど、同期タスク自体を実行するためにのみ使用されます。2 つの設定は目的が異なるため、独立しており、相互に交換することはできません。

設定については、次のドキュメントを参照してください：シャーディングされた MySQL テーブルを MaxCompute に同期する。

接続文字列を使用してデータソースを追加します。JDBC URL を次のように変更することを推奨します：jdbc:mysql://xxx.x.x.x:3306/database?com.mysql.jdbc.faultInjection.serverCharsetIndex=45。詳細については、「MySQL データソースの追加」をご参照ください。

• エラーの原因：

  • net_read_timeout：DataX は SplitPk に基づいて MySQL データを複数の同じサイズの SELECT ステートメントに分割します。実行中、いずれかの SQL ステートメントが RDS 側で許可される最大実行時間を超えます。

  • net_write_timeout：クライアントにブロックを送信するのを待つためのタイムアウトが小さすぎます。

• 解決策：

  データソースの URL 接続にパラメーターを追加し、net_write_timeout/net_read_timeout をより大きな値に設定するか、RDS コンソールでパラメーターを調整します。

• 改善提案：

  タスクが再実行可能な場合は、エラー時にタスクが自動的に再実行されるように設定します。

例：jdbc:mysql://192.168.1.1:3306/lizi?useUnicode=true&characterEncoding=UTF8&net_write_timeout=72000

エラーの原因：

MySQL パラメーター wait_timeout のデフォルトは 8 時間です。このタイムアウトに達したときにまだデータがフェッチされている場合、同期タスクは中断されます。

解決策：

MySQL 設定ファイル my.cnf (Windows では my.ini) を変更します。MySQL モジュールの下にパラメーターを追加します (秒単位)：wait_timeout=2592000 interactive_timeout=2592000。その後、MySQL を再起動してログインし、次のステートメントを実行して確認します：show variables like '%wait_time%'。

CPU 使用率は正常でもメモリ使用量が高い場合、接続が切断される可能性があります。

タスクが自動的に再実行できることを確認した場合は、[エラー時に自動再実行] を有効にすることを推奨します。詳細については、「自動再実行の設定」をご参照ください。

• シナリオ：バッチ同期ツールが PostgreSQL データを同期する際に、次のエラーが発生します：org.postgresql.util.PSQLException: FATAL: terminating connection due to conflict with recovery

• 考えられる原因：このエラーは、データベースからのデータプルに時間がかかりすぎるために発生します。max_standby_archive_delay と max_standby_streaming_delay の値を増やしてください。詳細については、「スタンバイサーバーイベント」をご参照ください。

• シナリオ：AWS PostgreSQL から MaxCompute へのリアルタイム同期タスクが、ソースの PostgreSQL ユーザーに REPLICATION 権限がないために失敗します。

• 考えられる原因：ソースの PostgreSQL ユーザーに REPLICATION 権限がありません。一部のマネージド AWS RDS PostgreSQL インスタンスでは、この属性は ALTER ROLE を使用して付与できません。

• 解決策：バッチ (オフライン、スケジュール) 同期には REPLICATION 権限は必要ありません。AWS PostgreSQL インスタンスで REPLICATION 権限を付与できない場合は、回避策としてリアルタイム同期の代わりに定期的なスケジュールを持つバッチ同期タスクを使用してください。

• シナリオ：PostgreSQL の timestamp (タイムゾーンなし) 列を MaxCompute の DATETIME 列に同期した後、結果の値が期待値と比較して、たとえば 2 時間ずれます。

• 考えられる原因：PostgreSQL の timestamp (タイムゾーンなし) 型は、ローカル時刻の値をそのまま保存します。MaxCompute の DATETIME 型は、値を協定世界時 (UTC) で保存し、プロジェクトまたはセッションのタイムゾーンに基づいて表示用に変換します。この保存と変換の動作の違いが、タイムゾーンのずれを引き起こす可能性があります。

• 解決策：

  1. SHOW timezone; を実行してクエリできる PostgreSQL サーバーのタイムゾーンが、MaxCompute プロジェクトのタイムゾーンと一致するかどうかを確認します。プロジェクトのタイムゾーンは、MaxCompute プロジェクトの基本情報ページで表示できます。

  2. タイムゾーンが異なる場合は、バッチ同期タスクの高度な設定でタイムゾーンを設定します。同期タスクのタイムゾーンを PostgreSQL サーバーのタイムゾーンと一致するように設定すると、timestamp の値が MaxCompute に書き込まれる前にそのタイムゾーンで解析され、ずれが解消されます。

     ソースと送信先のタイムゾーンの不一致は、通常、すべての時間フィールドで固定の N 時間のオフセットとして現れます。たとえば、両端が Asia/Bangkok に設定されているのに、同期された値が 1 時間異なる場合や、リソースグループがドイツで実行され、デフォルトで Europe/Berlin になっているのに、ビジネス要件では値を UTC で保存する必要がある場合などです。このような場合は、タスクの高度な設定で使用するタイムゾーンを選択してください。

  説明

  スケジューリングのタイムゾーンを変更しても、データ統合プロセスが使用するタイムゾーンには影響しません。2 つの設定は独立しています。スケジューリングのタイムゾーンを調整しても日付列が予期しない値を返す場合は、バッチ同期タスクの高度な設定でもタイムゾーンを明示的に設定してください。

  タイムゾーン設定のデフォルトは GMT+8 です。時間フィールドが正しく同期される場合はそのままにしておき、時間オフセットが発生した場合にのみ、ビジネスで期待されるタイムゾーンに設定してください。スクリプトモードでは、この設定は次の設定と同等です：

  "common":{"column":{"timeZone":"Asia/Bangkok"}}

• シナリオ：PostgreSQL からの増分同期のために WHERE 条件またはカスタムの querySql ステートメントを設定する際に、比較のために文字列形式の時間パラメーターをタイムスタンプに変換する必要があります。

• 解決策：PostgreSQL と MySQL は日付と時刻の文字列を変換するために異なる SQL 方言を使用するため、MySQL の STR_TO_DATE 関数の代わりに、標準の PostgreSQL TO_TIMESTAMP 関数を使用します。例：

  TO_TIMESTAMP('${start_time}', 'YYYYMMDDHH24')
  TO_TIMESTAMP('${end_time}', 'YYYYMMDDHH24')

  TO_TIMESTAMP は、指定された形式の文字列をタイムスタンプオブジェクトに変換します。これを WHERE 条件または querySql ステートメントで時間範囲内の行をフィルタリングするために使用できます。

• シナリオ：バッチ同期タスクが WHERE 条件付きで Oracle からデータを読み取る際に、タスクが ORA-00932: inconsistent datatypes で失敗します。

• 考えられる原因：WHERE 句が Oracle の DATE 列を NUMBER 値と直接比較しています。たとえば、プレーンな数値として書かれた日付リテラルなどです。この比較は Oracle でデータの型の不一致を引き起こします。

• 解決策：比較の前に、数値の日付値を明示的に DATE 型に変換します。たとえば、列をプレーンな数値と比較する代わりに、WHERE 条件で TO_DATE('20250611', 'YYYYMMDD') またはリテラル DATE '2025-06-11' を使用します。

Amazon RDS に接続して Host is blocked エラーを受け取った場合は、Amazon ロードバランサーのヘルスチェックを無効にしてください。無効にすると、ブロックの問題は発生しなくなります。

MongoDB データソースを追加する際は、同期するテーブルを含むデータベースで作成されたユーザーを使用してください。root ユーザーはサポートされていません。

たとえば、name テーブルをインポートしたい場合で、name テーブルが test データベースにある場合、データベース名は test になり、test データベースで作成されたユーザーのユーザー名を使用する必要があります。

代入ノードを使用して、まず日付型の値をタイムスタンプに変換し、その値を MongoDB データ同期タスクの入力パラメーターとして渡すことができます。

MongoDB Reader の設定でタイムゾーンを設定します。詳細については、「MongoDB Reader」をご参照ください。

クエリ条件を変更せずに、遅延後にタスクを再起動できます。つまり、設定を変更せずにタスクの実行時間を遅らせます。

データを読み取る際、ユーザーが設定した Column.name は大文字と小文字を区別します。設定が正しくないと、読み取られたデータが null になります。例：

• MongoDB ソースデータ：

  {
      "MY_NAME": "zhangsan"
  }

• 同期タスクの列設定：

  {
      "column":
      [
          {
              "name": "my_name"
          }
      ]
  }

列設定の大文字と小文字がソースデータと一致しないため、データの読み取りに失敗します。

タイムアウト設定パラメーターは cursorTimeoutInMs で、デフォルトは 600000 ms (10 分) です。このパラメーターは、データ転送時間を除き、MongoDB Server がクエリの実行に費やす合計時間を指定します。完全なデータ読み取りが大きい場合、次のエラーが発生する可能性があります：MongoDBReader$Task - operation exceeded time limitcom.mongodb.MongoExecutionTimeoutException: operation exceeded time limit。

現在、DataWorks の同期タスクはセカンダリノードからのデータ読み取りをサポートしていません。読み取り用にセカンダリノードを設定すると、次のエラーが発生します：no master。

• 根本原因分析：

  カーソルのタイムアウトが原因です。

• 解決策：

  cursorTimeoutInMs パラメーターの値を増やします。

タスクの同時実行数と読み取り BatchSize を増やします。

• 考えられる原因：

  デフォルトでは、同期タスクはタスクのシャーディングに splitVector コマンドを使用します。一部の MongoDB バージョンは splitVector コマンドをサポートしていないため、no such cmd splitVector エラーが発生します。

• 解決策：

  1. 同期タスク設定ページに移動し、上部にある [スクリプトに変換] ボタンをクリックします。タスクをスクリプトモードに変更します。

  2. MongoDB パラメーター設定で、次のパラメーターを追加します：

     "useSplitVector" : false

     これにより、splitVector の使用が回避されます。

• エラーメッセージ：

  同期タスクで、ウィザードモードを例にとると、書き込みモード (上書き) が はい に設定され、_id 以外の列が ビジネスプライマリキー として設定されている場合に、この問題が発生することがあります。

  同期タスクの送信先設定で、MongoDB データソース (インスタンス名：xc_mongo_rds) を選択し、コレクション名を xc_timestamp に設定し、上書きモードを有効にし (「上書き」を「はい」に設定)、my_id をビジネスプライマリキーとして指定します。

• 考えられる原因：

  書き込まれるデータに、_id が設定された ビジネスプライマリキー (上記の例では my_id など) と一致しないレコードが含まれています。

• 解決策：

  • オプション 1：バッチ同期タスクを変更して、設定された ビジネスプライマリキー が _id と同じになるようにします。

  • オプション 2：データ同期中に _id をビジネスプライマリキーとして使用します。

• 原因：

  Redis がストレージにハッシュモードを使用する場合、ハッシュの属性と値はペアで出現する必要があります。例：odpsReader: "column":[ "id", "name", "age", "address" ]。送信先で、RedisWriter が "keyIndexes":[ 0, 1] と設定されている場合、Redis では id と name がキーとして機能し、age が属性として、address がハッシュ型の値として機能します。ODPS ソースで 2 つの列しか設定されていない場合、Redis ストレージにハッシュモードを使用できず、この例外がスローされます。

• 解決策：

  2 つの列のみを使用したい場合は、ストレージに Redis の String モードを設定します。ハッシュモードを使用する必要がある場合は、ソース側で少なくとも 3 つの列を設定します。

• 症状：

  OSS や FTP などのファイルストレージからデータを読み取るバッチ同期タスクを設定する際に、ファイルが CSV 形式で、列区切り文字として複数の文字 (たとえば |,、##、または ;;) を使用している場合、タスクがダーティデータエラーで失敗することがあります。ランタイムログには、ダーティデータとともに IndexOutOfBoundsException エラーが表示されます。

• 根本原因分析：

  DataWorks の組み込み csv リーダー ("fileFormat": "csv") は、複数文字の区切り文字を処理する際に制限があり、データ行の列分割が不正確になります。

• 解決策：

  • ウィザードモード：テキストタイプを text に切り替え、複数文字の区切り文字を明示的に指定します。

  • スクリプトモード："fileFormat": "csv" を "fileFormat": "text" に変更し、区切り文字を正しく設定します： "fieldDelimiter":"<multi-char delimiter>", "fieldDelimiterOrigin":"<multi-char delimiter>"。

バッチ同期自体は、OSS Reader プラグインが読み取るファイル数を制限しません。主な制限は、タスクが消費する CU リソースに由来します。一度に多くのファイルを読み取ると、メモリ不足エラーが発生しやすくなります。したがって、OutOfMemoryError: Java heap space エラーを防ぐため、object パラメーターを * と設定することは推奨しません。

OSS Writer は、オブジェクト名を使用してディレクトリをシミュレートすることでファイル名を書き込みます。OSS にはオブジェクト名に関する制限があります。「object」：「datax」を使用すると、書き込まれるオブジェクトは datax で始まり、ランダムな文字列のサフィックスが追加されます。ファイル数は、実際の分割タスクの数によって決まります。

ランダムな UUID サフィックスが不要な場合は、「writeSingleObject」：「true」を設定します。詳細については、OSS Writer ドキュメントの writeSingleObject パラメーターの説明をご参照ください。

• 原因：

  データソースに設定された AccessKey に、バケットに対する権限がありません。

• 解決策：

  OSS データソースに設定された AccessKey アカウントに、バケットの読み取り権限を付与します。

• 根本原因分析：

  mapred.task.timeout パラメーターが低すぎるため、Hadoop がタスクを終了して一時ディレクトリをクリーンアップし、一時データが利用できなくなる可能性があります。

• 解決策：

  バッチ同期タスクのデータソースセクションで、Hive の読み取りメソッド が [Hive JDBC に基づくデータ読み取り (条件付きフィルタリングをサポート)] に設定されている場合、セッション設定 で mapred.task.timeout パラメーター値を設定します。たとえば、mapred.task.timeout=600000 です。

• エラーメッセージ：

  ERROR JobContainer - Exception when job runcom.alibaba.datax.common.exception.DataXException: Code:[DatahubWriter-04], Description:[Failed to write data.]. - com.aliyun.datahub.exception.DatahubServiceException: Record count 12498 exceed max limit 10000 (Status Code: 413; Error Code: TooLargePayload; Request ID: 20201201004200a945df0bf8e11a42)

• 考えられる原因：

  このエラーは、DataX が単一のバッチで DataHub に送信するデータ量が DataHub の制限を超えたために発生します。DataHub に送信されるデータ量に影響を与える主な設定パラメーターは次のとおりです：

  • maxCommitSize：累積バッファーデータサイズを指定します。累積データが maxCommitSize (MB 単位) に達すると、バッチで送信先に送信されます。デフォルトは 1 MB (1,048,576 バイト) です。

  • batchSize：DataX-On-Flume の累積バッファーデータレコード数を指定します。累積レコード数が batchSize に達すると、データはバッチで送信先に送信されます。

• 解決策：

  maxCommitSize と batchSize パラメーターの値を減らします。

このプラグインは列名の大文字と小文字を区別します。LogHub Reader の列設定を確認してください。

データ統合は、データが LogHub に入った時刻を使用します。LogHub コンソールで、メタデータ列 receive_time がタスクに設定された時間範囲内にあるかどうかを確認してください。

この問題が発生した場合は、UI で列設定を手動で編集してください。

バッチ同期タスクで設定された開始時刻と終了時刻は、Reader が SLS GetCursor API を呼び出して開始カーソルと終了カーソルを特定するために使用されます。この時間は、SLS サーバー側の受信時刻に基づいて読み取り範囲を特定するために使用されます。タスクは実際にカーソル範囲内のデータを読み取りますが、これは出力列 __time__ によるフィルタリングとは異なります。

出力列 __time__ は、各ログエントリの log.getTime() から取得され、ログ自体のログ時刻を表します。SLS コンソールのクエリは通常、クエリ時間範囲、クエリ文、およびインデックスキー列を統計に使用し、一般的にログ時刻 __time__ に基づいています。したがって、同期タスクとコンソールが同じ時間値を使用しても、両者が異なる時間メトリックを使用している場合、__time__ の範囲やレコード数が異なる場合があります。

適用シナリオ：

1. ログ収集や配信が遅延したり、過去のログがバックフィルされたり、クライアントの時計が不正確だったりすると、ログ時刻 __time__ が SLS サーバー側の受信時刻より早くなったり遅くなったりすることがあります。同期タスクはサーバー側の受信時刻に基づいてカーソルを特定しますが、コンソールは __time__ に基づいてクエリするため、結果が異なる場合があります。

2. SLS データ変換を介して別の LogStore にデータが書き込まれる場合、変換ステートメントが明示的に __time__ を設定しない限り、ターゲットログの __time__ は通常、変換実行時刻ではなくソースログの時刻を保持します。この場合、同期タスクは、変換がターゲット LogStore に書き込む時間範囲内でこのデータバッチを読み取る可能性があります。しかし、変換実行時刻または現在の時間範囲でターゲット LogStore コンソールをクエリすると、これらのログは見つからない場合があります。ログの実際の __time__ 範囲でクエリする必要があります。

3. コンソールのクエリ文、インデックスキー列、時間範囲、および同期タスクのルールフィルタリングステートメント (SPL) が一致しない場合、時間メトリックが同じであってもレコード数が異なる場合があります。

トラブルシューティングの提案：

1. コンソールのクエリ時間範囲、クエリ文、インデックスキー列、および同期タスクの開始/終了時刻とルールフィルタリングステートメント (SPL) が一致しているかどうかを確認します。

2. __time__ (ログ時刻) と __tag__:__receive_time__ (SLS サーバー側受信時刻の観測可能なフィールド。このフィールドがログタグに存在する必要があります) の両方を column 設定に含めて、ログ時刻とサーバー側受信時刻を比較します。

3. データが SLS データ変換から来ている場合は、変換ステートメントが明示的に __time__ を設定しているかどうかを確認し、実際の __time__ に基づいてターゲット LogStore のコンソールクエリ時間範囲を調整します。

4. 下流でログ時刻による厳密な照合が必要な場合は、送信先に書き込んだ後、__time__ でフィルタリングまたは集約します。

例：ソースログの __time__ は 2026-06-01 10:00:00 です。SLS データ変換タスクは、2026-06-12 10:00:00 にこのログをターゲット LogStore に書き込みますが、__time__ を明示的に変更しません。ターゲットログの __time__ は 2026-06-01 10:00:00 のままです。同期タスクの開始時刻と終了時刻が 2026-06-12 10:00:00 をカバーしている場合、タスクはこのログを読み取る可能性があります。しかし、2026-06-12 10:00:00 前後にターゲット LogStore をコンソールでクエリし、__time__ をフィルターとして使用すると、このログは見つからない場合があります。この場合、コンソールのクエリ時刻を 2026-06-01 10:00:00 前後に調整するか、必要に応じてデータ変換中にターゲットログの __time__ を明示的に設定します。

Reader は、column 設定に基づいて、実際にプルされたログコンテンツフィールド、Reader 組み込みのメタフィールドマッピング、および LogTag から列名を照合します。列名は大文字と小文字を区別します。一致が見つからない場合、エラーなしで null が出力されます。

一般的な原因は次のとおりです：

1. column で設定された列名の大文字と小文字が、元のログフィールドキーと異なります。

2. コンソールには、クエリ分析、インデックスフィールド、または JSON 展開フィールドからのエイリアスが表示されますが、これらは Reader が実際に取得する元のログキーとは異なります。

3. 列は実際には LogTag から来ており、__tag__:<tagKey> として設定する必要があります。

4. ルールフィルタリングステートメント (SPL) または変換を設定した後、出力フィールド名が column 設定と完全に一致しません。

トラブルシューティングの際は、まずビジュアルページのソーステーブルの列とデータプレビューを確認して、Reader が実際に識別するフィールドを確認します。スクリプトモードでは、一時的に column を ["*"] に設定して、Reader が取得する実際のログコンテンツフィールドキーを確認し、その後、元のキーに基づいて column を設定することもできます。

動作は API の書き込みロジックと同じです：同じ行と列のデータは上書きされ、他のデータは変更されません。

curl コマンドを使用して ES インデックスマッピングを取得し、マッピングからすべての列を抽出します。

• クエリ用のシェルコマンド：

  //es7
  curl -u username:password --request GET 'http://esxxx.elasticsearch.aliyuncs.com:9200/indexname/_mapping'
  //es6
  curl -u username:password --request GET 'http://esxxx.elasticsearch.aliyuncs.com:9200/indexname/typename/_mapping'

• 結果から列を取得する：

  {
      "indexname": {
          "mappings": {
              "typename": {
                  "properties": {
                      "field1": {
                          "type": "text"
                      },
                      "field2": {
                          "type": "long"
                      },
                      "field3": {
                          "type": "double"
                      }
                  }
              }
          }
      }
  }

  応答の properties の下の列と属性定義は、インデックスのすべての列です。たとえば、上記のインデックスには field1、field2、field3 の 3 つの列が含まれています。

インデックス設定に日付スケジューリングパラメーターを追加して、異なる日付に基づいてインデックス文字列を自動的に計算し、Elasticsearch Reader のインデックス名を自動的に変更できます。設定には、日付パラメーターの定義、インデックスパラメーターの設定、タスクのデプロイと実行の 3 つのステップが含まれます。

1. 日付パラメーターの定義：同期タスクのスケジュール設定で、パラメーターを追加して日付パラメーターを定義します。以下の var1 設定はタスクの実行時間 (当日) を表し、var2 はビジネス日付 (前日) を表します。

2. インデックスパラメーターの設定：タスクをスクリプトモードに切り替え、以下のように ${variable_name} の形式で Elasticsearch Reader インデックスを設定します。

   {
       "type": "job",
       "version": "2.0",
       "steps": [
           {
               "stepType": "elasticsearch",
               "parameter": {
                   "retryCount": 30,
                   "scroll": "10m",
                   "column": [
                       "col18",
                       "col17"
   
                   ],
                   "index": "esstress_1_${var1}_${var2}",
                   "pageSize": 100,
                   "sort": {
                       "_id": "asc"
                   },

3. タスクのデプロイと実行：検証後、タスクをオペレーションセンターに送信してデプロイし、定期的なスケジュールまたはバックフィルデータタスクとして実行します。

   1. パラメータ付きで実行 ボタンをクリックして、検証のためにタスクを直接実行します。パラメーター付きで実行すると、タスク設定で使用されるスケジューリングシステムパラメーターが置き換えられます。実行後、ログをチェックして、同期されたインデックスが期待どおりであるかを確認します。

      説明

      パラメーター付きで実行する場合、置換テストのためにパラメーター値を直接入力します。

   2. 前のステップで期待どおりに検証された場合、タスク設定は完了です。保存 をクリックし、次に コミット をクリックして、同期タスクを本番環境に送信します。

      標準モードのワークスペースの場合、発行 をクリックしてデプロイメントセンターに移動し、同期タスクを本番環境にデプロイします。

4. 結果：以下に設定と実際のランタイムインデックスの結果を示します。

   スクリプトインデックス設定："index": "esstress_1_${var1}_${var2}"。

   ランタイムインデックスは次のように解決されました：esstress_1_20230106_20230105。

   ],
   "full":false,
   "gmtCreate":"2022-07-18 14:47:18",
   "gmtModified":"2022-07-18 14:47:18",
   "index":"esstress_1_20230106_20230105",
   "instanceId":"es-cn-2r42se1je001zwmt0",
   "ownerId":"1224800975333052",
   "pageSize":100,
   "password":"********",
   "privateNetworkIpWhiteList":[
       "0.0.0.0/0"
   ],

オブジェクトフィールドのプロパティを同期するには、スクリプトモードのみを使用できます。スクリプトモードでは、multi を次のように設定し、column を attribute.sub-attribute 形式で指定します。

"multi":{
   "multi":true 
 }

設定については、次の例を参照してください：

#例：
##Elasticsearch のデータ
"hits": [
    {
        "_index": "mutiltest_1",
        "_type": "_doc",
        "_id": "7XAOOoMB4GR_1Dmrrust",
        "_score": 1.0,
        "_source": {
            "level1": {
                "level2": [
                    {
                        "level3": "testlevel3_1"
                    },
                    {
                        "level3": "testlevel3_2"
                    }
                ]
            }
        }
    }
]
##Reader の設定
"parameter": {
  "column": [
      "level1",
      "level1.level2",
      "level1.level2[0]"
  ],
  "multi":{
        "multi":true
    }
}
##Writer の結果：1 行 3 列、列の順序は Reader の設定と一致
COLUMN              VALUE
level1:             {"level2":[{"level3":"testlevel3_1"},{"level3":"testlevel3_2"}]}
level1.level2:      [{"level3":"testlevel3_1"},{"level3":"testlevel3_2"}]
level1.level2[0]:   {"level3":"testlevel3_1"}

1. 文字の前後に表示される余分な二重引用符は、Kibana の表示上の問題です。実際のデータには、これらの先頭と末尾の二重引用符はありません。curl コマンドまたは Postman を使用して実際のデータを表示してください。データを取得するための curl コマンドは次のとおりです：

   //es7
   curl -u username:password --request GET 'http://esxxx.elasticsearch.aliyuncs.com:9200/indexname/_mapping'
   //es6
   curl -u username:password --request GET 'http://esxxx.elasticsearch.aliyuncs.com:9200/indexname/typename/_mapping'

2. ES の書き込み列の型を nested に設定することで、ODPS からの JSON 型の文字列データを ES に nested 形式で同期できます。次の例では、name 列を ES に nested 形式で同期します。

   • 同期設定：name の型を nested に設定します。

   • 同期結果：name は nested オブジェクト型です。

     "total": {
         "value": 1,
         "relation": "eq"
     },
     "max_score": 1.0,
     "hits": [
         {
             "_index": "test",
             "_type": "_doc",
             "_id": "bb5oqoUBlqHPyI16REEQ",
             "_score": 1.0,
             "_source": {
                 "name": [
                     {
                         "fields1": "value"
                     }
                 ]
             }
         }
     ]
     }

ES に配列型を書き込むための設定方法は 2 つあります。ソースデータの形式に基づいて、対応する同期方法を選択してください。

• ソースデータを JSON として解析して ES に配列型として書き込みます。たとえば、ソースデータが「[1,2,3,4,5]」の場合、json_array=true を設定してソースデータを解析し、ES の列に配列として書き込みます。ColumnList に json_array=true を設定します。

  • ウィザードモードの設定。

  • スクリプトモードの設定：

    "column":[
      {
        "name":"docs",
        "type":"keyword",
        "json_array":true
      }
    ]

• 区切り文字でソースデータを解析して ES に配列型として書き込みます。たとえば、ソースデータが「1,2,3,4,5」の場合、区切り文字 splitter="," を設定してデータを解析し、ES の列に配列として書き込みます。

  • 制限事項：

    • タスクは 1 つの区切り文字のみをサポートします。splitter はグローバルに一意であり、異なる配列列に異なる区切り文字をサポートしません。たとえば、ソース列 col1="1,2,3,4,5" , col2="6-7-8-9-10" の場合、splitter は各列に個別に設定できません。

    • splitter は正規表現として設定できます。たとえば、ソース列の値が「6-,-7-,-8+,*9-,-10」の場合、splitter:".,." と設定でき、これはウィザードモードでサポートされています。

  • ウィザードモードの設定：splitter: デフォルトは "-,-"

  • スクリプトモードの設定：

    "parameter" : {
          "column": [
            {
              "name": "col1",
              "array": true,
              "type": "long"
            }
          ],
          "splitter":","
    }

• 根本原因分析：

  HttpClient は、接続が確立されるたびに、最初に認証されていないリクエストを行うことを規定しています。サーバーが認証要件を返した後 (応答に基づいて認証方式を指定)、認証されたリクエストが行われます。各 ES データ書き込みには接続の確立が必要なため、すべてのデータ書き込みで 1 つの認証されていないリクエストが生成され、それが監査ログに記録されます。

• 解決策：

  スクリプトモードで "preemptiveAuth":true 設定を追加します。

日付の書き込みを設定するには 2 つの方法があります。ニーズに応じて適切なものを選択してください。

• Reader から読み取った内容に基づいて ES の Date 列に直接書き込みます：

  • origin:true を設定して、読み取った内容を直接 ES に書き込みます。

  • ES 書き込みでマッピングを作成する際に、列の format 属性を指定するために "format" を設定します。

    "parameter" : {
        "column": [
            {
                "name": "col_date",
                "type": "date",
                "format": "yyyy-MM-dd HH:mm:ss",
                "origin": true
            }
              ]
    }

• タイムゾーン変換：データ統合にタイムゾーン変換を実行させる必要がある場合は、Timezone パラメーターを追加します。

  "parameter" : {
      "column": [
          {
              "name": "col_date",
              "type": "date",
              "format": "yyyy-MM-dd HH:mm:ss",
              "Timezone": "UTC"
          }
            ]
  }

• type:version が設定されていますが、ES は外部バージョンの指定をサポートしていません。

      "column":[
                              {
                                  "name":"id",
                                  "type":"version"
                              },
    ]

• 解決策：

  "type":"version" 設定を削除します。Elasticsearch Writer は外部バージョンの指定をサポートしていません。

• 根本原因分析：

  対応する ES の日付列のマッピングに format が設定されていないため、Elasticsearch Reader は日付型の列の日付形式を解析できません。

• 解決策：

  • dateFormat パラメーターを ES の日付列の形式と同じ形式で設定し、区切り文字として "||" を使用します。形式には、すべての日付型の形式を含める必要があります。例：

    "parameter" : {
          "column": [
         			"dateCol1",
            	"dateCol2",
              "otherCol"
          ],
         "dateFormat" : "yyyy-MM-dd||yyyy-MM-dd HH:mm:ss",
    }

  • ES データベースのすべての日付列のマッピング形式を設定します。

• 根本原因分析：

  fastjson のキーワード制限により、インデックスまたは列に $ref などのキーワードが含まれている可能性があります。

• 解決策：

  Elasticsearch Reader は、列名に $ref キーワードを含むインデックスの同期をサポートしていません。詳細については、「Elasticsearch Reader」をご参照ください。

• 根本原因分析：

  これにより ES の楽観的ロックメカニズムがトリガーされました。現在のバージョン番号は 1 つの値であるべきですが、更新コマンドによって渡されたバージョン番号が異なるため、バージョンの競合が発生しました。更新中に、他の誰かがインデックスデータを削除していました。

• 解決策：

  1. データ削除操作が発生しているかどうかを確認します。

  2. タスクの同期方法を Update から Index に変更します。

• 根本原因分析：

  列に similarity や properties などの高度な属性を設定する場合、プラグインがそれらを認識するためには other_params が必要です。

  "parameter":{
          "__datasource__type":"elasticsearch",
          "actionType":"index",
          "aliasMode":"append",
          "batchSize":1024,
          "cleanup":false,
          "column":[
              {
                  "name":"id",
                  "type":"long"
              },
              {
                  "name":"dim1_code",
                  "type":"keyword"
              },
              {
                  "analyzer":"china",
                  "name":"dim1_name",
                  "similarity":"len_similarity",
                  "type":"text"
              },
              {
                  "analyzer":"china",
                  "name":"dim1_val",
                  "similarity":"len_similarity",
                  "type":"text"
              },
              {
                  "name":"dim1_sort",
                  "type":"long"
              }

• 解決策：

  列設定で other_params を設定し、other_params の中に similarity を追加します。次のようになります：

  {"name":"dim2_name",...,"other_params":{"similarity":"len_similarity"}}

• 根本原因分析：

  現在、Elasticsearch へのバッチ同期書き込みは dense_vector 型をサポートしていません。次の型のみがサポートされています：

  ID,PARENT,ROUTING,VERSION,STRING,TEXT,KEYWORD,LONG,
  INTEGER,SHORT,BYTE,DOUBLE,FLOAT,DATE,BOOLEAN,BINARY,
  INTEGER_RANGE,FLOAT_RANGE,LONG_RANGE,DOUBLE_RANGE,DATE_RANGE,
  GEO_POINT,GEO_SHAPE,IP,IP_RANGE,COMPLETION,TOKEN_COUNT,OBJECT,NESTED;

• 解決策：

  Elasticsearch Writer がサポートしていない型については、次のように処理します：

  • Elasticsearch Writer を使用してインデックスマッピングを作成することは推奨しません。代わりにカスタムマッピングを使用してください。

  • 対応する型を NESTED に変更します。

  • 設定を dynamic = true、cleanup=false に変更します。

• 原因：

  #不正な設定
  "settings": {
    "index": {
      "number_of_shards": 1,
      "number_of_replicas": 0
    }
  }
  #正しい設定
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0
  }

• 解決策：

  設定はインデックスが作成されるときにのみ有効になります。これには、インデックスが存在しない場合、または cleanup=true の 2 つのケースが含まれます。cleanup=true の場合、設定には「index」を含める必要はありません。

#元のマッピング
{
  "name":"box_label_ret",
  "properties":{
    "box_id":{
      "type":"keyword"
    }
}
#cleanup=true で再構築した後、次のようになります
{
    "box_label_ret": {
      "properties": {
        "box_id": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }}}}
}

• 根本原因分析：

  ネストされた型の場合、Elasticsearch Writer はトップレベルのマッピングのみを使用し、ES にネストされた複雑な型を自動適応させます。属性の型が text に変わり、fields:keyword が追加されるのは ES の自動適応動作であり、ES の使用には影響しません。特定のマッピング形式が必要な場合は、「Elasticsearch Writer」をご参照ください。

• 解決策：

  同期前に期待される ES インデックスマッピングを作成し、ES 同期タスクで cleanup を false に設定してタスクを実行します。

Kafka Reader はデータをバッチで読み取ります。データのバッチで、いずれかのレコードが endDateTime を超えると、同期は停止します。ただし、そのバッチ内の endDateTime を超えるデータは、依然として送信先データソースに書き込まれます。

• skipExceedRecord 設定を使用して、超過したデータを同期するかどうかを指定することもできます。詳細な使用方法については、「Kafka Reader」をご参照ください。[これを同期しないように設定すると、データ損失の原因となる可能性があるため、推奨されません。]

• Kafka の max.poll.records パラメーターを設定して、各バッチでプルされるデータ量を指定できます。同時実行数と組み合わせることで、制限を超える可能性のあるデータ量を制御できます。超過データ量 < max.poll.records × 同時実行数。

• 根本原因分析：

  データ量が少ない場合やデータが不均等に分散している場合、一部の Kafka パーティションに新しいデータが入らないか、新しいデータが指定された終了オフセットに達しないことがあります。タスクの終了条件はすべてのパーティションが指定された終了オフセットに達することであるため、これらの「アイドル」パーティションは条件を満たせず、タスク全体の正常な完了を妨げます。

• 解決策：

  同期終了ポリシーを1 分間新しいデータを読み取らないに設定します (スクリプトモードでは、stopWhenPollEmpty を true に、stopWhenReachEndOffset を true に設定します)。タスクは、すべてのパーティションから最新のオフセットデータを読み取った後に終了し、アイドリングを回避します。ただし、タスク終了後に書き込まれる、設定された終了オフセットより前のタイムスタンプを持つレコードは消費されません。

RestAPI Writer は 2 つの書き込みモードを提供します。複数のレコードを同期する場合は、dataMode を multiData に設定し、スクリプトにパラメーター dataPath:"data.list" を追加します。詳細については、「RestAPI Writer」をご参照ください。

1. OTS Writer の設定には、次の 2 つの要件が含まれている必要があります：

   "newVersion": "true",
   "enableAutoIncrement": "true",

2. 自動採番主キー列名は OTS Writer で設定しないでください。

3. OTS Writer で設定された primaryKey のエントリ数 + column のエントリ数は、上流の OTS Reader データの列数と等しくなければなりません。

例：データレコードには 3 つのタグがあります：[phone=xiaomi, RAM=8G, camera=LEICA]。

• データエクスポートの例 (OTS Reader)

  • 上記のタグを単一の列にマージしてエクスポートしたい場合は、次のように設定します：

    "column": [
          {
            "name": "_tags",
          }
        ],

    DataWorks は、タグを次の形式の単一のデータ列としてエクスポートします：

    ["phone=xiaomi","camera=LEICA","RAM=8G"]

  • phone タグと camera タグを別々の列としてエクスポートしたい場合は、次のように設定します：

    "column": [
          {
            "name": "phone",
            "is_timeseries_tag":"true",
          },
          {
            "name": "camera",
            "is_timeseries_tag":"true",
          }
        ],

    DataWorks は、次の形式の 2 つのデータ列をエクスポートします：

    xiaomi, LEICA

• データインポートの例 (OTS Writer)

  上流のデータソース (Reader) には 2 つのデータ列があります：

  • 1 つの列には ["phone=xiaomi","camera=LEICA","RAM=8G"] が含まれています。

  • もう 1 つの列には 6499 が含まれています。

  両方の列をタグに追加するには、書き込み後の期待されるタグフィールド形式は次のようになります：次のように設定します：

  "column": [
        {
          "name": "_tags",
        },
        {
          "name": "price",
          "is_timeseries_tag":"true",
        },
      ],

  • 最初の列設定は、["phone=xiaomi","camera=LEICA","RAM=8G"] を全体としてタグフィールドにインポートします。

  • 2 番目の列設定は、price=6499 を個別にタグフィールドにインポートします。

テーブル名が orders_20170310、orders_20170311、orders_20170312 のように規則的なパターンに従っている場合 (テーブルが日付で区別され、同じ構造を共有している場合)、スケジューリングパラメーター (スクリプトモードでの同期タスクの設定) を使用してテーブル名をカスタマイズし、毎朝早くソースデータベースから前日のテーブルデータを自動的に読み取ることができます。

スクリプトモードでは、ソーステーブル名を orders_${tablename} のような変数に変更します。テーブルは日付で区別され、毎日前日のデータを読み取る必要があるため、タスクパラメーター設定で変数値として tablename=${yyyymmdd} を割り当てます。

同期タスク設定ページに移動し、列マッピングを変更してタスク設定で変更された列を更新し、変更を有効にするためにタスクを再送信して実行します。

バッチ同期ノードを設定する際、ソースセクションには、選択したデータソースから最初の 25 テーブルのみがデフォルトで表示されます。テーブルがさらにある場合は、テーブル名を入力して検索するか、開発にスクリプトモードを使用できます。

• エラーの原因：列設定に予約キーワードが含まれているか、列設定に数値で始まる列が含まれています。

• 解決策：データ統合同期タスクをスクリプトモードに切り替え、列設定で特殊な列をエスケープします。スクリプトモードでのタスクの設定については、「スクリプトモードでの同期タスクの設定」をご参照ください。

  • MySQL のエスケープ文字は `keyword` です。

  • Oracle と PostgreSQL のエスケープ文字は "keyword" です。

  • SQL Server のエスケープ文字は [keyword] です。

  MySQL の例：

  {
      "stepType": "mysql",
      "parameter": {
          "envType": 0,
          "datasource": "wpw_test_mysql",
          "column": [
              "id",
              "`order`",
              "`add`"
          ],
          "connection": [
              {
                  "datasource": "wpw_test_mysql",
                  "table": [
                      "abc"
                  ]
              }

• MySQL データソースを例にとります：

  1. 次のステートメントを実行して、aliyun という名前のテーブルを作成します：create table aliyun (`table` int ,msg varchar(10));

  2. 次のステートメントを実行してビューを作成し、テーブル列にエイリアスを割り当てます：create view v_aliyun as select `table` as col1,msg as col2 from aliyun;

     説明

     • table は MySQL のキーワードです。データ同期中に、連結されたコードがエラーを引き起こします。ビューを作成し、テーブル列にエイリアスを割り当てます。

     • キーワードをテーブル列名として使用することは推奨しません。

  3. 上記のステートメントを実行した後、同期タスクを設定する際に aliyun テーブルの代わりに v_aliyun ビューを使用します。

このエラーは、同期タスクの列マッピングが正しく設定されていないか、プラグインに列が正しく設定されていないために発生する可能性があります。

1. 列マッピングが設定されているかどうかを確認します。

2. プラグインに列が正しく設定されているかどうかを確認します。

• 症状：

  データのプレビュー をクリックすると、列のバイトサイズが制限を超えていることを示す次のようなメッセージが表示されます。

  エラーメッセージは「Maximum column length of 1,000 exceeded in column 6 in record 1. Set the SafetySwitch property to false if you're expecting column lengths greater than 100,000 characters to avoid this error.」です。

• 原因：OOM を防ぐため、データソースサービスはデータプレビューリクエストを処理する際に列の長さをチェックします。単一の列が 1000 バイトを超えると、上記のメッセージが表示されます。このメッセージは実際のタスク実行には影響しません。このエラーを無視して、バッチ同期タスクを直接実行できます。

  説明

  ファイルが存在し、接続が正常な場合でも、次の状況でデータプレビューが失敗することがあります：

  • ファイル内の単一行が 10 MB のバイトサイズ制限を超えています。この場合、上記のメッセージと同様にデータは表示されません。

  • ファイル内の単一行が 1000 列の列数制限を超えています。この場合、最初の 1000 列のみが表示され、1001 番目の列にメッセージが表示されます。

TTL はテーブルレベルで設定されます。同期タスクの設定には TTL オプションはありません。

API ベースの同期は、ソース側の関数の使用をサポートしていません。インポートする前に、まずソース側の関数を使用してデータを処理してください。