BCPユーティリティを使用したMS SQL Serverへのバルクローディング

概要

TDV ServerからMicrosoft SQL Serverデータソース内のテーブルにBCP（Bulk Copy Program）ユーティリティを使用してデータのバルクローディングを行う方法について説明します。

TDVにおいては、Microsoft SQL Serverデータソースについてはネイティブ ロード パフォーマンス オプションとして「BCP を使用したバルク ローディング」がサポートされています。
参考）データ ソースのネイティブ ロード パフォーマンス オプションについて

本記事では、TDV Serverの実行環境がWindowsの場合とLinuxの場合、それぞれの手順について説明します。

検証環境

Windows環境

Linux環境

bcpユーティリティ

TDV Server実行環境において、bcpコマンドが実行できる状態にあることを確認します。

Windows環境

コマンドプロンプトを起動し、

bcp -v

を実行して以下のようにバージョンが表示されることを確認します。

BCP - Microsoft SQL Server の一括コピー プログラム。
Copyright (C) Microsoft Corporation. All Rights Reserved.
バージョン: 16.0.1000.6

また、

where bcp

を実行することで以下のようにbcp.exeのパスを確認することができます。

C:\Program Files\Microsoft SQL Server\Client SDK\ODBC\170\Tools\Binn\bcp.exe

このパスは後続のTDV Server設定で使用します。
環境によって上記のパスとは異なる可能性もありますので、必ず確認してください。

Linux環境

/opt/mssql-tools18/bin/bcp -v

のようにコマンドを実行して以下のようにバージョンが表示されることを確認します。

BCP - Bulk Copy Program for Microsoft SQL Server.
Copyright (C) Microsoft Corporation. All Rights Reserved.
Version: 18.4.0001.1

TDV Server設定

bcpユーティリティを使用したMS SQL Serverへのバルクローディングを有効にするための設定を行います。

TDV Studioの管理 > 構成 > Microsoft BCPユーティリティパス にbcpのパスを設定します。

Windows環境

Linux環境

Trust server certificate

TDV 8.8.1より、MS SQL Serverデータソース専用のTrust server certificateというBCP一括ロードコマンドによる-uオプションの使用を制御するための設定が導入されています。デフォルト値はtrueです。bcpのバージョンが18より前のバージョンで、かつ、バルクローディングが動作しない場合は、この設定値をfalseに設定することで動作する可能性があります。

データソース設定

バルクローディング先のテーブルが存在するデータソース（本記事ではMS SQL Serverデータソース）に対し、以下のように設定します。

「データシップの下限」と「データシップの上限」は、本件（MS SQL ServerデータソースへのBCPによるバルクローディング）には関係のない設定値であるため、デフォルト設定のままで問題ありません。

データロード用プロシージャ

OracleデータソースのORDERSテーブル（1,700万行）とSQL ServerデータソースのSales.Productsテーブル（1万件）を結合した結果セットを、SQL ServerデータソースのSales.Order_Analysisテーブルにロードするプロシージャ（Load_Order_Analysis）を以下のとおり作成します。

Oracleデータソース、SQL ServerデータソースのいずれもTDV Server実行環境であるマシンとは異なるマシン上で実行されています。

/* * プロシージャ名: Load_Order_Analysis
 * 説明: 分析用データをターゲットテーブルにロードし、実行ログを出力する
 * 実行結果には所要時間(Duration)が含まれます。
 */
PROCEDURE Load_Order_Analysis(
    OUT result CURSOR(
        ProcessName VARCHAR(255), 
        StartTime   TIMESTAMP, 
        EndTime     TIMESTAMP, 
        Duration    VARCHAR(255), -- 所要時間（文字列表記）
        RowCount    INTEGER, 
        Status      VARCHAR(50), 
        Message     VARCHAR(1000)
    )
)
BEGIN
    -- 変数宣言
    DECLARE v_startTime     TIMESTAMP;
    DECLARE v_endTime       TIMESTAMP;
    DECLARE v_interval      INTERVAL DAY TO SECOND;
    DECLARE v_hours         INTEGER;
    DECLARE v_minutes       INTEGER;
    DECLARE v_seconds       DECIMAL(10, 2);
    DECLARE v_durationStr   VARCHAR(255);
    DECLARE v_rowCount      INTEGER;
    DECLARE v_msg           VARCHAR(1000);

    -- 1. 開始時間の取得
    SET v_startTime = CURRENT_TIMESTAMP;

    -- ログ出力 (TDV Studioのコンソールに出力されます)
    CALL PRINT('[INFO] Start Load_Order_Analysis at ' || CAST(v_startTime AS VARCHAR));

    -- 2. 事前に件数を取得 (ログおよび確認用)
    SELECT 
        COUNT(*)
    INTO v_rowCount
    FROM /shared/BCP/bcp_disabled/Views/Order_Analysis_Source;

    BEGIN
        -- 3. INSERT処理の実行
        -- ターゲット: SQL Server (Order_Analysis)
        -- ソース: 作成したビュー
        INSERT INTO /shared/BCP/bcp_disabled/DataSources/SQLServer/TDV_DB/TDV_DB/Sales/Order_Analysis
            (OrderID, OrderDate, ProductID, ProductName, Quantity, TotalAmount) 
        SELECT 
            OrderID,
            OrderDate,
            ProductID,
            ProductName,
            Quantity,
            TotalAmount
        FROM /shared/BCP/bcp_disabled/Views/Order_Analysis_Source;

        -- 4. 終了時間の取得
        SET v_endTime = CURRENT_TIMESTAMP;

        -- 5. 所要時間の計算とフォーマット
        SET v_interval = v_endTime - v_startTime;
        SET v_hours = EXTRACT(HOUR FROM v_interval);
        SET v_minutes = EXTRACT(MINUTE FROM v_interval);
        SET v_seconds = EXTRACT(SECOND FROM v_interval);

        IF v_hours  0 THEN
            SET v_durationStr = CAST(v_hours AS VARCHAR) || ' hours ' || CAST(v_minutes AS VARCHAR) || ' minutes ' || CAST(CAST(v_seconds AS INTEGER) AS VARCHAR) || ' seconds';
        ELSEIF v_minutes  0 THEN
            SET v_durationStr = CAST(v_minutes AS VARCHAR) || ' minutes ' || CAST(v_seconds AS VARCHAR) || ' seconds';
        ELSE 
            SET v_durationStr = CAST(v_seconds AS VARCHAR) || ' seconds';
        END IF;

        -- 6. 成功ログの出力
        SET v_msg = 'Successfully inserted ' || CAST(v_rowCount AS VARCHAR) || ' rows.';
        CALL PRINT('[INFO] ' || v_msg);
        CALL PRINT('[INFO] End Load_Order_Analysis. Duration: ' || v_durationStr);

        -- 7. 結果カーソルのオープン (成功時)
        OPEN result FOR SELECT 
            'Order Analysis Load' AS ProcessName,
            v_startTime           AS StartTime,
            v_endTime             AS EndTime,
            v_durationStr         AS Duration,
            v_rowCount            AS RowCount,
            'SUCCESS'             AS Status,
            v_msg                 AS Message;
    EXCEPTION
        ELSE
            -- エラーハンドリング
            SET v_endTime = CURRENT_TIMESTAMP;

            -- エラー時も所要時間を計算
            SET v_interval = v_endTime - v_startTime;
            SET v_seconds = EXTRACT(SECOND FROM v_interval); -- エラー時は簡易的に秒のみでも可
            SET v_durationStr = CAST(EXTRACT(MINUTE FROM v_interval) AS VARCHAR) || ' min ' || CAST(v_seconds AS VARCHAR) || ' sec';
            SET v_msg = 'Error: ' || CURRENT_EXCEPTION.MESSAGE;

            -- エラーログ出力
            CALL PRINT('[ERROR] ' || v_msg);

            -- 結果カーソルのオープン (エラー時)
            OPEN result FOR SELECT 
                'Order Analysis Load' AS ProcessName,
                v_startTime           AS StartTime,
                v_endTime             AS EndTime,
                v_durationStr         AS Duration,
                0                     AS RowCount,
                'FAILED'              AS Status,
                v_msg                 AS Message;
    END;
END

データロードプロシージャとそれ以外のデータソースやビュー等のリソース一覧をTDV Studioで開くと以下のようになります。

動作確認

プロシージャを実行し、バルクローディングを実行します。

Windows環境

4分28秒でバルクローディングが完了しました。

実行計画は以下のとおりです。

Linux環境

2分52秒でバルクローディングが完了しました。

実行計画は以下のとおりです。

Row by Row INSERTとの比較

Row by Row INSERT（バルクローディングではない1行ずつのINSERT）の場合との比較表は以下のとおりです。

参考）SQL Serverに対して実行されたINSERTクエリの確認

SQL Serverのsys.query_store_query_textにクエリを実行することで、どういったINSERTが実行されたかが確認できます。

Row by Row（1行ごとのINSERT）の場合は以下のようなクエリが確認できます。

(@P0 bigint, @P1 datetime2(7), @P2 int, @P3 nvarchar(4000), @P4 int, @P5 bigint)
INSERT INTO [TDV_DB].[Sales].[Order_Analysis] 
    ([OrderID], [OrderDate], [ProductID], [ProductName], [Quantity], [TotalAmount]) 
VALUES
    ( @P0 , @P1 , @P2 , @P3 , @P4 , @P5 )

BCPによるバルクローディングの場合は以下のようなクエリが確認できます。

insert bulk [TDV_DB].[Sales].[Order_Analysis]
(
    [OrderID] bigint,
    [OrderDate] datetime,
    [ProductID] int,
    [ProductName] nvarchar(100) collate Japanese_CI_AS,
    [Quantity] int,
    [TotalAmount] bigint
)
with(TABLOCK, KEEP_NULLS, ROWS_PER_BATCH=2000)