Qwen NLP 統合によるオフライン同期タスクの構築 - DataWorks

AI 支援処理は、大規模言語モデル (LLM) の推論を DataWorks のデータ統合パイプラインに直接統合します。データをそのまま移動する代わりに、パイプラインがリアルタイムで AI モデルを呼び出し、ソースから送信先に転送される各レコードを分析、エンリッチ、または変換します。これにより、AI 呼び出しコードを記述することなく、非構造化テキストから価値を引き出すことができます。

この機能は、データ同期中にデータに対して高度な分析と処理を実行する必要があるエンタープライズユーザー向けに設計されています。特に、AI を活用してデータ品質を向上させ、データから価値を引き出したい企業にとって有用です。

サポートされる NLP タスク

タスク	機能
感情分析	テキストの感情を分類します
要約生成	長いドキュメントを要点に要約します
キーワード抽出	最も関連性の高い用語やフレーズを抽出します
テキスト翻訳	テキストをある言語から別の言語に変換します

ユースケース

業界	AI 処理の活用方法
カスタマーサービス / E コマース	ユーザーコメントやサポートチケットの感情を分析し、中心的な問題や主要なフィードバックポイントを抽出します
コンプライアンス / 法務 / 科学研究	同期中にポリシー文書、法的契約、研究論文から要約を生成し、重要な情報を抽出します
製造 / サプライチェーン / ヘルスケア	デバイスログ、サプライチェーンのフィードバック、診療記録を分析し、脅威アラートやサービス品質のシグナルを明らかにします
多言語コラボレーション	ソーシャルメディアのコメント、ニュース記事、ビジネス文書を単一の言語に翻訳し、一元的に分析します

前提条件

開始する前に、以下が準備できていることを確認してください：

Data Studio (新バージョン) を使用するワークスペース
サポートされているプロバイダーのいずれかから設定された大規模モデルサービス：
- Alibaba Cloud DataWorks モデルサービス — モデルサービス管理でモデルをデプロイし、モデルサービスを開始します
- Alibaba Cloud Model Studio — Alibaba Cloud Model Studio を有効化し、API キーを取得します
- PAI モデルギャラリー — Platform for AI (PAI) を有効化し、モデルサービスのトークンを取得します
手動で設定された、またはオフライン同期タスクの既存のデータソースとして設定されたデータソース情報
データソースへの接続性を持つワークスペースにアタッチされたリソースグループ

課金

DataWorks のサブスクリプション料金とリソースグループ料金に加えて、AI 支援処理にはモデル推論料金が発生します。

例：Hologres から Hologres への同期中に顧客フィードバックを翻訳

この例では、ある Hologres テーブルから別のテーブルにデータをコピーするオフライン同期タスクで AI 支援処理を設定する手順を説明します。このタスクは feedback_info カラムの各値を英語に翻訳し、その結果を送信先テーブルに書き込みます。

この例における AI 処理の動作：

ソースフィールド	処理の説明	出力フィールド
`feedback_info`	`Translate '#{feedback_info}' into English`	`feedback_processed`

ステップ 1：ソースデータの準備

ソーステーブルを作成し、サンプルレコードを挿入します：

CREATE TABLE customer_feedback (
    id BIGINT PRIMARY KEY,
    device STRING,
    feedback_info STRING,
    pt INT
)
PARTITIONED BY (pt)
DISTRIBUTED BY HASH(id)
WITH (table_type='Duplication');

INSERT INTO customer_feedback (id, device, feedback_info, pt)
VALUES
(8, 'Huawei MateBook D14', 'Affordable, suitable for students, performance is adequate', 2020),
(1, 'iphone', 'This product is okay, I have used it for 1 year', 2013),
(10, 'Bose QuietComfort 35 II', 'A classic among noise-canceling headphones, maximum comfort', 2021);

ステップ 2：オフライン同期タスクの作成

DataWorks コンソールのワークスペースページに移動します。上部のナビゲーションバーでリージョンを選択します。対象のワークスペースを見つけ、操作列で ショートカット > Data Studio を選択します。
左側のナビゲーションウィンドウでをクリックして Data Studio を開きます。ワークスペースディレクトリ の右側にあるをクリックし、ノードの作成 > データ統合 > バッチ同期 を選択します。新しいノード ダイアログボックスが開きます。
ノードのパス、データソースと送信先、名前を設定し、OK をクリックします。

ステップ 3：同期タスクの設定

ノードが作成されると、タスク設定ページが開きます。各セクションを順番に設定します。

データソース

タイプ — ノード作成時に設定され、変更できません。異なるデータソースタイプを使用するには、新しいノードを作成してください。
データソース — ドロップダウンリストから既存のデータソースを選択するか、データソースの追加 をクリックして新規作成します。

ランタイムリソース

同期タスクの リソースグループ を選択します。サーバーレスリソースグループの場合、リソース使用量 (CU) フィールドで CU 数を指定します。
リソースグループを選択すると、Data Integration はソースと送信先への接続性を自動的にチェックします。接続性チェック をクリックして手動でチェックを実行することもできます。

ソース

ソーステーブルの設定を行います：スキーマ、テーブル、パーティション、データフィルタリング 条件。データプレビュー をクリックして、同期されるレコードをプレビューします。

データ処理

データ処理セクションで、有効化 を切り替えてデータ処理をオンにします。これには追加のコンピューティングリソースが必要です。
ノードの追加 をクリックし、AI プロセス を選択します。

AI プロセスノードを設定します。主要なパラメーターは以下の通りです：

パラメーター	説明
モデルプロバイダー	プロバイダーを選択します：DataWorks モデルサービス Aliyun Bailian、、または PAI モデルギャラリー
モデルエンドポイント	モデルプロバイダーが PAI モデルギャラリーの場合に適用されます。モデル呼び出しエンドポイントを入力します。エンドポイントの取得方法については、「サービスの呼び出しをテストする
モデル名	インテリジェントなデータ処理に使用されるモデルです。リストから 1 つ選択します
API キー	モデルにアクセスするための認証情報です。Alibaba Cloud Model Studio の場合、「Model Studio API キーの取得」をご参照ください。PAI モデルギャラリーの場合、デプロイされた EAS タスクに移動し、オンラインデバッグを開始し、アクセストークンを API キーとして使用します
処理の説明	ソースフィールドをどのように処理するかを記述する自然言語の命令です。フィールド名は `#{column_name}` のフォーマットで参照します。この例では、`Translate '#{feedback_info}' into English`
出力フィールド	処理結果が格納されるフィールドです。フィールドが存在しない場合は自動的に作成されます。この例では、`feedback_processed`

データ処理セクションの右上隅にある データ出力プレビュー をクリックして、処理された出力をプレビューします。
(オプション) さらに処理ノードを追加します。複数のデータ処理ノードは順次実行されます。

宛先

送信先テーブルを設定します：スキーマ、テーブル、パーティション。
- ターゲットテーブルスキーマの生成 をクリックして、スキーマを自動的に生成します。
- 既存のテーブルがある場合は、それを選択します。
書き込みモード と 書き込み競合戦略 を設定します。
同期前に Hologres テーブルの既存データを削除するかどうかを設定します。
(オプション) 最大接続数 を設定します。

最大接続数 は、書き込みモード が SQL(INSERT INTO) の場合にのみ適用されます。1 つのタスクで最大 9 つの接続を使用できます。タスクを開始する前に、Hologres インスタンスに十分なアイドル接続があることを確認してください。

送信先フィールドマッピング

ソース、データ処理、送信先を設定した後、ソーステーブルと送信先テーブル間のフィールドマッピングが表示されます。デフォルトでは、フィールドは名前と位置によってマッピングされます。

この例では、id、device、feedback_info、pt のデフォルトマッピングに加えて、ソースの AI 出力フィールドである feedback_processed を送信先テーブルの translate_feedback に手動でマッピングします。

ステップ 4：タスクのテスト

右側のパネルで 実行設定 をクリックします。このテスト実行のために リソースグループ と任意の スクリプトパラメーター を設定します。
ツールバーで保存をクリックし、次に実行をクリックします。タスクが終了したら、結果が成功したことを確認し、送信先テーブルをチェックしてデータが正しいことを確認します。

ステップ 5：スケジューリングの設定

同期タスクをスケジュールに従って実行するには、ページ右側の スケジューリング セクションで スケジューリングポリシー を設定し、ノードのスケジューリングプロパティを設定します。

ステップ 6：ノードの公開

ツールバーの公開アイコンをクリックして、公開フローを開始します。これにより、タスクが本番環境に公開されます。定期的なスケジューリングは、ノードが公開された後にのみ有効になります。

次のステップ

ノードが公開された後、既存データのバックフィルや、オペレーションセンターでのタスクの監視・管理ができます。

[データバックフィル] — 現在のノードのデータをバックフィルします。高度なバックフィルワークフローについては、オペレーションセンターをご利用ください。詳細については、「データバックフィルタスクの実行およびデータバックフィルインスタンスの表示（新規）」をご参照ください。
O&M の実行 — 公開後、タスクはオペレーションセンターによって自動的に管理されます。実行ステータスを表示したり、手動で実行をトリガーしたりできます。詳細については、「オペレーションセンター」をご参照ください。

DataWorks:バッチ同期のための AI 支援処理