Microsoft SQL Server

Foundry を Microsoft SQL Server に接続して、SQL Server データベースと Foundry の間でデータの読み取りと同期を行います。

サポートされている機能

セットアップ

1. Data Connection アプリケーションを開き、画面の右上にある + 新規ソースを選択します。
2. 利用可能なコネクターの種類から MS SQL Server を選択します。
3. 以下のセクションの情報を使用して、追加の設定プロンプトに従い、コネクターのセットアップを続行します。

Foundry でのコネクターのセットアップについて詳しくはこちら。

認証

SQL Server では、次の方法で認証できます。

1. ユーザー名とパスワード： ユーザー名とパスワードを入力します。個々のユーザーの資格情報ではなく、サービスの資格情報を使用することを推奨します。
2. Active Directory Msi*：このオプションは、JDBC 設定 authentication=ActiveDirectoryMSI を使用します。必要に応じて msiClientId を指定することもできます。
3. Active Directory パスワード*：このオプションは、JDBC 設定 authentication=ActiveDirectoryPassword を使用します。この設定を使用するには、Active Directory ユーザーのユーザー名とパスワードを入力する必要があります。
4. Active Directory サービスプリンシパル*：このオプションは、JDBC 設定 authentication=ActiveDirectoryServicePrincipal を使用します。プリンシパルID（アプリケーションIDまたはクライアントIDと呼ばれることもあります）と、そのプリンシパルIDのシークレットを指定する必要があります。

これらの認証モードの詳細については、公式ドキュメント ↗を参照してください。どの認証オプションでも、指定したユーザーとロールに、対象データベースに対する必要な権限と、対象テーブルの読み取りまたは書き込み権限があることを確認してください。

* Azure Active Directory は現在、Microsoft Entra ID ↗ と呼ばれています。ただし、Microsoft が公開している SQL Server ドライバーの JDBC オプションには、Active Directory を指す元の名前が引き続き使用されています。

ネットワーク

Microsoft SQL Server コネクターには、接続先の SQL Server インスタンスへのネットワークアクセスが必要です。SQL Server 接続では通常、ホスト名を使用してポート1433に接続します。

SQL Server 接続では、Data Connection アプリケーションでソースをセットアップするときに、適切なエグレスポリシーを適用する必要があります。

Azure SQL や AWS RDS などのクラウドサービスでホストされている SQL Server インスタンスの場合は、クラウドプロバイダーのコンソールから取得したホスト名のエグレスポリシーを追加する必要があります。

• Azure SQL インスタンスのホスト名を確認するには、Azure ポータルの設定 > プロパティページに移動し、サーバー名の項目を探します。Azure SQL のホスト名ベースのエグレスポリシーの例：<your-database-name>.database.windows.net (port 1433)
• 名前解決された IP アドレスを確認するには、コマンドラインから nslookup <your-database-name>.database.windows.net を実行できます。最終的な結果は、このホスト名が Azure で名前解決される IP アドレスです。Azure SQL の IPv4 ベースのエグレスポリシーの例は、x.x.x.x (port 1433) です。Azure SQL は複数のホスト間で負荷分散を行うため、nslookup コマンドを数回実行し、名前解決されたすべての IP アドレスを追加する必要がある場合があります。

接続情報

変更データキャプチャ

Microsoft SQL Server ソースは変更データキャプチャ同期をサポートしています。

Microsoft SQL Server で変更データキャプチャを有効にするには、次のようなコマンドを実行して、データベースで CDC を有効にする必要があります。

USE <database>
GO
EXEC sys.sp_cdc_enable_db
GO

次に、変更履歴を記録する必要がある各テーブルに対して、別のコマンドを実行します。

EXEC sys.sp_cdc_enable_table
    @source_schema = N'<schema>'
  , @source_name = N'<table_name>'
  , @role_name = NULL
  , @capture_instance = NULL
  , @supports_net_changes = 0
  , @filegroup_name = N'PRIMARY';
GO

Foundry に同期するテーブルで変更データキャプチャを有効にしたら、概要ページに移動し、+ CDC 同期を作成を選択して、新しい変更データキャプチャ同期の作成を開始できます。

変更データキャプチャ同期を作成するには、探索用ランタイムが動作している必要があります。ランタイムがまだ初期化中の場合は、数秒待ってページを再読み込みしてから同期を作成する必要がある場合があります。

これらのコマンドと Microsoft SQL Server での変更データキャプチャ（CDC）の使用について詳しくは、使用している SQL Server のバージョンの公式ドキュメント ↗を参照してください。

変更データキャプチャの権限設定

CDC データを正常に読み取るには、データベースユーザーに十分な権限を付与していることを確認する必要があります。

• 次のクエリで権限を確認できます：SELECT HAS_PERMS_BY_NAME('cdc', 'SCHEMA', 'EXECUTE') AS HasExecutePermission;。結果は True の場合は1、False の場合は0を返します。
• ソースシステム内で次のクエリを実行すると、不足している権限を付与できます：GRANT EXECUTE ON SCHEMA::cdc TO <USER>;

トラブルシューティング

Active Directory Password 認証を使用するとログインに失敗する

Active Directory Password 認証の選択肢を使用したときに Login failed for user ''. エラーが表示される場合、通常は SQL Server インスタンスが Kerberos または NTLM 認証のみをサポートし、Azure Active Directory 認証を受け付けないことを示しています。

ActiveDirectoryPassword 認証方式は、Kerberos または NTLM プロトコルを使用するのではなく、Azure Active Directory に対して認証を行います。SQL Server が Kerberos または NTLM 認証のみを受け付けるように設定されている場合、Azure AD 認証の試行は拒否されます。

この問題を解決するには、カスタム JDBC 接続を作成し、JDBC パラメーター authenticationScheme=NTLM を設定して、代わりに NTLM 認証を使用します。

コードで Microsoft SQL Server ソースを使用する

これらの例は、外部トランスフォームで pymssql ↗ Python パッケージを使用して、Microsoft SQL Server ソースに接続する方法を示しています。

これらの例は、次のスキーマで作成された Fruits テーブルに基づいています。

Copied!
1
2
3
4
5
CREATE TABLE Fruits (
    FruitName VARCHAR(50) PRIMARY KEY,
    Inventory INT NOT NULL,
    PricePerKg DECIMAL(10, 2) NOT NULL
);

外部トランスフォームで MSSQL から読み取る

この例では、Fruits テーブルから、Inventory の値が 60 未満のデータに絞り込んで読み取ります。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
import logging
from pandas import DataFrame
from transforms.external.systems import ResolvedSource
from transforms.api import lightweight, Output, transform_pandas
from transforms.external.systems import external_systems, Source
import pymssql
import pandas as pd

logger = logging.getLogger(__name__)

@lightweight
@external_systems(
    mssql_source=Source("<source_rid>")
)
@transform_pandas(
    Output("<dataset_rid>")
)
def compute(mssql_source: ResolvedSource) -> DataFrame:
    # 在庫のしきい値パラメーター（入力 DataFrame から読み込むこともできます）
    INVENTORY_THRESHOLD = 60

    connection_parameters = {
        "server": "<your_server_name>",
        "database": "<your_database_name>",
        "port": "1433",
        "user": "<your_user_name>",
        "password": mssql_source.get_secret("MSSQL_BASIC_AUTH_PASSWORD"),
        "encryption": "require",
        "timeout": 30
    }
    try:
        with pymssql.connect(**connection_parameters) as connection:
            df = pd.read_sql(
                'SELECT * FROM Fruits WHERE Inventory < %s',
                connection,
                params=(INVENTORY_THRESHOLD,)
            )
    except Exception as e:
        logger.error(f"Error querying MSSQL Fruits table: {e}")
        raise RuntimeError(f"Failed to fetch Fruits data from MSSQL: {e}") from e
    return df

外部トランスフォームで MSSQL に書き込む

この例では、入力データセットを使用して Fruits テーブルを更新します。FruitName ごとに実行された操作（更新または挿入）をまとめたデータセットを返します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import logging
from transforms.api import lightweight, Output, transform_pandas, Input
from transforms.external.systems import external_systems, Source
import pymssql
import pandas as pd

logger = logging.getLogger(__name__)

@lightweight
@external_systems(
    mssql_source=Source("<source_rid>")
)
@transform_pandas(
    Output("<dataset_rid>"),
    fruits_df=Input("<dataset_rid>") # スキーマ [FruitName: String, Inventory: Integer, PricePerKg: Double] を持つ DataFrame
)
def compute(mssql_source, fruits_df):
    # 接続パラメーター
    connection_parameters = {
        "server": "<your_server_name>",
        "database": "<your_database_name>",
        "port": "1433",
        "user": "<your_user_name>",
        "password": mssql_source.get_secret("MSSQL_BASIC_AUTH_PASSWORD"),
        "encryption": "require",
        "timeout": 30
    }
    results = []
    try:
        with pymssql.connect(**connection_parameters) as connection:
            with connection.cursor() as cursor:
                for _, row in fruits_df.iterrows():
                    fruit_name = row["FruitName"]
                    inventory = row["Inventory"]
                    price_per_kg = row["PricePerKg"]

                    # 更新を試み、更新された行がなければ挿入します
                    update_sql = """
                        UPDATE Fruits
                        SET Inventory = %s, PricePerKg = %s
                        WHERE FruitName = %s
                    """
                    cursor.execute(update_sql, (inventory, price_per_kg, fruit_name))
                    if cursor.rowcount == 0:
                        insert_sql = """
                            INSERT INTO Fruits (FruitName, Inventory, PricePerKg)
                            VALUES (%s, %s, %s)
                        """
                        cursor.execute(insert_sql, (fruit_name, inventory, price_per_kg))
                        results.append({"FruitName": fruit_name, "action": "inserted"})
                    else:
                        results.append({"FruitName": fruit_name, "action": "updated"})
                connection.commit()
    except Exception as e:
        logger.error(f"Error updating MSSQL Fruits table: {e}")
        raise RuntimeError(f"Failed to update Fruits data in MSSQL: {e}") from e
    # 操作をまとめた DataFrame を返します
    return pd.DataFrame(results, columns=["FruitName", "action"])