Zenn
💨

リモート関数と Translation API で翻訳を試す。with 用語集

2024/10/21に公開
tech
Google Cloud
BigQuery
translation
tech

はじめに

こんにちは。クラウドエース 第三開発部所属の金です。

前回 Cloud Translation API と Cloud Functions を利用して Cloud Storage にファイルが格納されたら自動検知して翻訳する機能を試してみました。
Cloud Translation APIとCloud Functionsを利用して自動翻訳を試してみた

本記事で紹介する内容は、リモート関数を使用して BigQuery 上簡単に翻訳する機能を試してみます。

リモート関数とは？

リモート関数は、BigQuery 内で定義された関数ですが、その処理の実行は BigQuery 外部で行われます。つまり、外部のコードやサービス (例: Cloud Run 関数、Python スクリプト) を呼び出して処理を実行し、その結果を BigQuery に返します。

リモート関数のメリット

  • 外部のサービスやライブラリを活用 : BigQuery に組み込まれていない機能やライブラリを利用できます。
  • 複雑な処理の外部化 : 複雑なロジックや処理を外部のコードに委譲することで、クエリのパフォーマンスを向上させ、クエリの複雑さを軽減できます。
  • コードの再利用 : 同じロジックを複数のクエリで利用する場合、リモート関数にまとめることでコードの再利用性を高められます。
  • 柔軟なデータ処理: リモート関数を使用することで、さまざまな形式のデータ (JSON、CSVなど) を処理したり、外部 API と連携したりできます。

今回やりたいこと

  1. リモート関数作成
  2. Cloud Translation API（用語集利用）で翻訳（JA => EN）
  3. BigQuery 上クエリで翻訳実行
    1
    <構成図>

事前準備

  1. API を有効にします。

  2. ロールを付与します。

    • BigQuery データオーナー（roles/bigquery.dataOwner）
      • BigQuery データセットに対する完全な管理権限です。
    • BigQuery Connection 管理者（roles/bigquery.connectionAdmin）
      • BigQuery Connection に関する管理権限です。今回はリモート関数が対象になります。
    • Cloud Functions デベロッパー（roles/cloudfunctions.developer）
      • Cloud Functions 関数の作成、削除、更新、デプロイ、管理を行うための権限です。

  3. Compute Engine のデフォルトサービスアカウントに必要なロールを付与します。

    <PROJECT_NUMBER>-compute@developer.gserviceaccount.com
    • Google Cloud コンソールの IAM ページに移動します。
    • プロジェクトを選択する。
    • [アクセス権を付与] をクリックし、[新しいプリンシパル] フィールドに、先ほどコピーした Compute Engine のデフォルトサービスアカウントを貼り付けます。
    • [ロールを割り当てる] のリストで、[Cloud Translation API ユーザー] を見つけて選択します。
    • [保存] をクリックします。

Cloud Run 関数を作成（用語集なし）

  1. 次の仕様で Cloud Run 関数を作成します。

    • [環境] には、[第 2 世代] を選択します。
    • [関数名] に「translation-handler」と入力します。
    • [リージョン] で、[us-central1] を選択します。
      2

    [ランタイム, ビルド, 接続, セキュリティの設定]

    • [インスタンスの最大数] に「10」と入力します。(チュートリアルでは、デフォルトよりも小さい値を使用)
      3
    • 左下の[次へ]ボタンを押します。
    • [ランタイム] では [Python 3.10] を選択します。
    • [エントリポイント] に「handle_translation」と入力します。
      4

  2. ファイルリストで main.py を選択し、コードを貼り付けます。
    5

main.py
from __future__ import annotations


import flask
import functions_framework
from google.api_core.retry import Retry
from google.cloud import translate

# Construct a Translation Client object
translate_client = translate.TranslationServiceClient()


# Register an HTTP function with the Functions Framework
@functions_framework.http
def handle_translation(request: flask.Request) -> flask.Response:
    """BigQuery remote function to translate input text.

    Args:
        request: HTTP request from BigQuery
        https://cloud.google.com/bigquery/docs/reference/standard-sql/remote-functions#input_format

    Returns:
        HTTP response to BigQuery
        https://cloud.google.com/bigquery/docs/reference/standard-sql/remote-functions#output_format
    """
    try:
        # Parse request data as JSON
        request_json = request.get_json()
        # Get the project of the query
        caller = request_json["caller"]
        project = extract_project_from_caller(caller)
        if project is None:
            return flask.make_response(
                flask.jsonify(
                    {
                        "errorMessage": (
                            'project can\'t be extracted from "caller":' f" {caller}."
                        )
                    }
                ),
                400,
            )
        # Get the target language code
        context = request_json.get("userDefinedContext", {})
        target = context.get("target_language", "en")

        calls = request_json["calls"]
        translated = translate_text([call[0] for call in calls], project, target)

        return flask.jsonify({"replies": translated})
    except Exception as err:
        return flask.make_response(
            flask.jsonify({"errorMessage": f"Unexpected error {type(err)}:{err}"}),
            400,
        )


def extract_project_from_caller(job: str) -> str:
    """Extract project id from full resource name of a BigQuery job.

    Args:
        job: full resource name of a BigQuery job, like
          "//bigquery.googleapi.com/projects/<project>/jobs/<job_id>"

    Returns:
        project id which is contained in the full resource name of the job.
    """
    path = job.split("/")
    return path[4] if len(path) > 4 else None


def translate_text(
    calls: list[str], project: str, target_language_code: str
) -> list[str]:
    """Translates the input text to specified language using Translation API.

    Args:
        calls: a list of input text to translate.
        project: the project where the translate service will be used.
        target_language_code: The ISO-639 language code to use for translation
          of the input text. See
          https://cloud.google.com/translate/docs/advanced/discovering-supported-languages-v3#supported-target
            for the supported language list.

    Returns:
        a list of translated text.
    """
    location = "us-central1"
    parent = f"projects/{project}/locations/{location}"
    # Call the Translation API, passing a list of values and the target language
    response = translate_client.translate_text(
        request={
            "parent": parent,
            "contents": calls,
            "target_language_code": target_language_code,
            "mime_type": "text/plain",
        },
        retry=Retry(),
    )
    # Convert the translated value to a list and return it
    return [translation.translated_text for translation in response.translations]
  1. ファイルリストで requirements.txt を選択し、次のテキストを貼り付けます。
    6
requirements.txt
Flask==2.2.2
functions-framework==3.5.0
google-cloud-translate==3.11.1
Werkzeug==2.3.7

  1. [デプロイ] をクリックし、関数がデプロイされるのを待ちます。

  2. [トリガー] タブをクリックします。

  • [トリガー URL] の値をコピーし、後で使用できるように保存します。BigQuery リモート関数を作成するときに、この URL を使用する必要があります。
    7

BigQuery データセットを作成

  • [データセット ID] に[remote_function_test]と入力します。
  • [ロケーションタイプ] で [マルチリージョン] を選択します。
  • [マルチリージョン] で [US（米国の複数のリージョン）] を選択します。
    8

BigQuery サービスアカウントに権限を付与

  1. Cloud Run ページに移動します。
  2. プロジェクトを選択します。
  3. translation-handler の横のチェックボックスをクリックします。
  4. [権限] パネルで、[プリンシパルを追加] をクリックします。
  5. [新しいプリンシパル] フィールドに、前の手順でコピーしたサービスアカウント ID を入力します。
  6. [ロールを割り当てる] のリストで、[Cloud Run 起動元] を見つけて選択します。
  7. [保存] をクリックします。
    9

BigQuery リモート関数を作成

  1. Google Cloud コンソールで BigQuery ページに移動します。
  2. クエリエディタで以下のクエリを入力します。
CREATE OR REPLACE FUNCTION `remote_function_test.translate_text`(x STRING)
RETURNS
STRING
    REMOTE WITH CONNECTION `us.remote-function-connection`
OPTIONS (
    endpoint = 'TRIGGER_URL',
    max_batching_rows = 10);
  1. クエリを実行すると、次のようなメッセージが表示されます。
This statement created a new function named
your_project.remote_function_test.translate_text.

ここまでで準備が整ったので、実際にクエリを実行して翻訳が正しく動作するか試してみましょう。

BigQuery リモート関数を呼び出す

  1. BigQuery クエリエディタで次のクエリを入力し、実行します。
    翻訳する内容は前回の記事と同じものにします。
SELECT
  remote_function_test.translate_text('私の名前は金です。')
    AS translated_text;

10

SELECT
  remote_function_test.translate_text('私はスイカが好きです')
    AS translated_text;

11
翻訳は問題なく出ました。少し驚いたのは、以前の記事で試したときに固有名詞（名前）を固有名詞だと認識せずに金をGOLDと翻訳していたのに対し、今回は名前として翻訳してくれました。前回の検証時よりも成長したかもですね！

  1. テーブル単位でのリモート関数テストもできます。
    事前に<PROJECT_ID>.test.test_translate テーブルを作成しました。
<PROJECT_ID>.test.test_translate
CREATE OR REPLACE TABLE `<PROJECT_ID>.test.test_translate`
(
    ja STRING
);

INSERT INTO `<PROJECT_ID>.test.test_translate` VALUES ("私はスイカが好きです。");
INSERT INTO `<PROJECT_ID>.test.test_translate` VALUES ("私は名前は金です。");

SELECT
    ja,
    remote_function_test.translate_text(ja) AS translated_text
FROM
    (SELECT ja FROM `<PROJECT_ID>.test.test_translate`);

12

問題なく翻訳されました！
それでは本題の用語集を使った翻訳を試してみましょう。

Cloud Run 関数を修正（用語集あり）

  1. 作成した Cloud Run 関数に戻り、ソース編集ボタンを押します。
    13
  2. 用語集に必要なコードを追加、保存して再デプロイボタンを押します。
  • translate_text 関数の引数に glossary_id を追加します。
def translate_text(
    calls: list[str], project: str, source_language_code: str,target_language_code: str,glossary_id: str = "glossary_ja_en"
) -> list[str]:
  • translate_text 関数内で、glossaryの設定を追加します。
location = "us-central1"
    parent = f"projects/{project}/locations/{location}"

    # Glossaryの設定
    glossary = translate_client.glossary_path(
        project, location, glossary_id
    )
    glossary_config = translate.TranslateTextGlossaryConfig(glossary=glossary)
    
    # Call the Translation API, passing a list of values and the target language
    response = translate_client.translate_text(
        request={
            "parent": parent,
            "contents": calls,
            "source_language_code": source_language_code,
            "target_language_code": target_language_code,
            "mime_type": "text/plain",
            "glossary_config": glossary_config,  # Glossaryの設定を追加
        },
        retry=Retry(),
    )
  • handle_translation 関数内で、translate_text 関数の呼び出しを更新します。
 # Get the target language code
        context = request_json.get("userDefinedContext", {})
        source = context.get("source_language", "ja")
        target = context.get("target_language", "en")
        glossary_id = context.get("glossary_id", "glossary_ja_en")

        calls = request_json["calls"]
        translated = translate_text([call[0] for call in calls], project, source, target, glossary_id)
  • 結果をglossary_translationsに変更します。
# 結果をglossary_translationsに変更
    return [translation.translated_text for translation in response.glossary_translations]

全体コードは下記になります。

main.py 全体コード
from __future__ import annotations


import flask
import functions_framework
from google.api_core.retry import Retry
from google.cloud import translate

# Construct a Translation Client object
translate_client = translate.TranslationServiceClient()


# Register an HTTP function with the Functions Framework
@functions_framework.http
def handle_translation(request: flask.Request) -> flask.Response:
    """BigQuery remote function to translate input text.

    Args:
        request: HTTP request from BigQuery
        https://cloud.google.com/bigquery/docs/reference/standard-sql/remote-functions#input_format

    Returns:
        HTTP response to BigQuery
        https://cloud.google.com/bigquery/docs/reference/standard-sql/remote-functions#output_format
    """
    try:
        # Parse request data as JSON
        request_json = request.get_json()
        # Get the project of the query
        caller = request_json["caller"]
        project = extract_project_from_caller(caller)
        if project is None:
            return flask.make_response(
                flask.jsonify(
                    {
                        "errorMessage": (
                            'project can\'t be extracted from "caller":' f" {caller}."
                        )
                    }
                ),
                400,
            )
        # Get the target language code
        context = request_json.get("userDefinedContext", {})
        source = context.get("source_language", "ja")
        target = context.get("target_language", "en")
        glossary_id = context.get("glossary_id", "glossary_ja_en")

        calls = request_json["calls"]
        translated = translate_text([call[0] for call in calls], project, source, target, glossary_id)

        return flask.jsonify({"replies": translated})
    except Exception as err:
        return flask.make_response(
            flask.jsonify({"errorMessage": f"Unexpected error {type(err)}:{err}"}),
            400,
        )


def extract_project_from_caller(job: str) -> str:
    """Extract project id from full resource name of a BigQuery job.

    Args:
        job: full resource name of a BigQuery job, like
          "//bigquery.googleapi.com/projects/<project>/jobs/<job_id>"

    Returns:
        project id which is contained in the full resource name of the job.
    """
    path = job.split("/")
    return path[4] if len(path) > 4 else None


def translate_text(
    calls: list[str], project: str, source_language_code: str,target_language_code: str,glossary_id: str = "glossary_ja_en"
) -> list[str]:
    """Translates the input text to specified language using Translation API.

    Args:
        calls: a list of input text to translate.
        project: the project where the translate service will be used.
        target_language_code: The ISO-639 language code to use for translation
          of the input text. See
          https://cloud.google.com/translate/docs/advanced/discovering-supported-languages-v3#supported-target
            for the supported language list.

    Returns:
        a list of translated text.
    """
    location = "us-central1"
    parent = f"projects/{project}/locations/{location}"

    # Glossaryの設定
    glossary = translate_client.glossary_path(
        project, location, glossary_id
    )
    glossary_config = translate.TranslateTextGlossaryConfig(glossary=glossary)
    
    # Call the Translation API, passing a list of values and the target language
    response = translate_client.translate_text(
        request={
            "parent": parent,
            "contents": calls,
            "source_language_code": source_language_code,
            "target_language_code": target_language_code,
            "mime_type": "text/plain",
            "glossary_config": glossary_config,  # Glossaryの設定を追加
        },
        retry=Retry(),
    )
    
    # 結果をglossary_translationsに変更
    return [translation.translated_text for translation in response.glossary_translations]

用語集には以下の CSV を利用しました。

glossary.csv
ja,en
金,kim
スイカ,made in Hokkaido water melon
  1. クエリで実行します。
SELECT
    ja,
    remote_function_test.translate_text(ja) AS translated_text
FROM
    (SELECT ja FROM `<PROJECT_ID>.test.test_translate`);

14

用語集に基づいて翻訳した結果、翻訳内容が変更されていることを確認しました。

おわりに

今回はリモート関数と用語集を活用した翻訳を試してみました。
リモート関数と用語集を活用することで、正確な翻訳を実現することができました。適切なユースケースでリモート関数を活用することで、データ分析の効率性と柔軟性を高められると感じています。

今後は、ML.TRANSLATE 関数を使った翻訳も試してみたいと考えています。この関数を利用することで、より自然な翻訳を実現し、翻訳作業の効率化に役立つと考えています。

最後まで読んでいただきありがとうございました。

関連記事

Discussion