Azure Synapse Analytics Dedicated SQL pool ( 専用 SQL プール) における Merge（Upsert）処理の性能検証を実施する方法

## 概要

Azure Synapse Analytics Dedicated SQL pool ( 専用 SQL プール) における Merge（Upsert）処理を、 TPC-H の LINEITEM テーブルのデータを用いて性能検証を実施する方法を共有します。


下記のリソースを利用します。Azure Datbricks にて TPC-H のデータを生成、および、TPC-H のParquet 化を実施します。その後、出力される SQL 文を Azure Synapse Analytics Dedicated SQL pool (専用 SQL プール)で実行します。

- Azure Synapse Analytics
- Azure Storage
- Azure Databricks


TPC-H の概要とデータ生成方法については、下記記事にて整理してます。

-   [データベースの性能検証に利用されるTPC-HとTPC-DSに関するざっくりとした整理 - Qiita](https://qiita.com/manabian/items/93d9330cdb607383c2b9)
-   [Databrick(Spark)にて、TPC-H、および、TPC-DSのデータを生成する方法 - Qiita](https://qiita.com/manabian/items/710cc4d6a3b54e75661b)

本手順後に、下記表のデータを生成できます。Merge 処理時のソースでは約600万レコードを想定しており、ファイルサイズは Snappy 形式で圧縮した Parquet 形式のファイルサイズを示しております。

| #    | 想定処理           | 処理後のレコード数 | ソースのレコード数 | ソースの新規レコード数 | ソースの既存レコード数 | ファイルサイズ（MB） | テキストファイル時のサイズ（MB） |
| ---- | ------------------ | ------------------ | ------------------ | ---------------------- | ---------------------- | -------------------- | -------------------------------- |
| 0    | 初期データのロード | 44,991,725         | 44,991,725         | 44,991,725             | 0                      |                      |                                  |
| 1    | １度目の処理       | 47,989,007         | 5,996,953          | 2,997,282              | 2,999,671              | 245                  | 774                              |
| 2    | ２度目の処理       | 50,988,593         | 5,997,887          | 2,999,586              | 2,998,301              | 245                  | 776                              |
| 3    | ３度目の処理       | 53,986,608         | 5,996,879          | 2,998,015              | 2,998,864              | 245                  | 778                              |
| 4    | ４度目の処理       | 56,982,414         | 5,997,357          | 2,995,806              | 3,001,551              | 245                  | 779                              |
| 5    | ５度目の処理       | 59,986,052         | 6,000,991          | 3,003,638              | 2,997,353              | 245                  | 779                              |

2022 年 10 月 31 日時点にて、Merge 文により Upsert 処理を実施する場合には、ターゲットのテーブルがハッシュ分散テーブルのみに制限されております。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/83024565-0ef3-2d3e-e016-c1f09a613086.png)

引用元：[MERGE (Transact-SQL) - SQL Server | Microsoft Learn](https://learn.microsoft.com/ja-jp/sql/t-sql/statements/merge-transact-sql?view=azure-sqldw-latest&preserve-view=true)

本手順で記述しているストレージは削除済みです。

## データと SQL 文の生成
### TPC-H のデータ生成

```pyhon
scale_factor = "10"  # Scale Factor数をセット
chunks_number = "20"  # 出力ファイル数と並列処理数をセット
 
path = "file:/tmp/tpch_data"
dbfs_path = "dbfs:/FileStore/load_test/raw"
```

```pyhon
# ディレクトリの初期化
dbutils.fs.rm(path, True)
dbutils.fs.rm(dbfs_path, True)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6911b87d-3084-22a6-3246-a06f7b255bbc.png)


```pyhon
import os
 
# shellスクリプトに渡す引数をセット
os.environ["scale_factor"] = scale_factor
print(os.getenv("scale_factor"))
 
os.environ["chunks_number"] = chunks_number
print(os.getenv("chunks_number"))
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/718d2acd-5c62-58a1-9414-a51d1b3a3881.png)


```pyhon
%sh
sudo apt-get install -y make gcc yacc
cd /tmp
rm -r ./tpch-dbgen -f
git clone https://github.com/databricks/tpch-dbgen.git
cd tpch-dbgen
make OS=LINUX WORKLOAD=TPCH
cd ..
rm -r ./tpch_data -f
mkdir tpch_data
cd tpch_data
 
for i in $( seq 1 ${chunks_number} ) ; do
    ../tpch-dbgen/dbgen -b ../tpch-dbgen/dists.dss -f -s ${scale_factor} -C ${chunks_number} -S ${i} -T L &
done
wait
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/faabd6ac-6c8f-40d4-aa3f-2afdef812457.png)

```pyhon
file_list = dbutils.fs.ls(path)
 
# ファイル一覧を表示
display(file_list)
 
# ファイルサイズを表示
spark.createDataFrame(file_list).groupBy().sum("size").display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b66ce242-fe40-215f-ec68-fcfcd931d5b4.png)


```pyhon
# dbfs 上にファイルをコピー
dbutils.fs.cp(path, dbfs_path, True)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2aadb68f-1335-c913-3af1-921bc3e42532.png)

```pyhon
# データ件数チェック（ 59,986,052 件のはず）
def create_tpch_lineitem_dataframe(
    filepath,
):
 
    schema = """
      L_ORDERKEY    INTEGER ,
      L_PARTKEY     INTEGER ,
      L_SUPPKEY     INTEGER ,
      L_LINENUMBER  INTEGER ,
      L_QUANTITY    DECIMAL(15,2) ,
      L_EXTENDEDPRICE  DECIMAL(15,2) ,
      L_DISCOUNT    DECIMAL(15,2) ,
      L_TAX         DECIMAL(15,2) ,
      L_RETURNFLAG  STRING ,
      L_LINESTATUS  STRING ,
      L_SHIPDATE    DATE ,
      L_COMMITDATE  DATE ,
      L_RECEIPTDATE DATE ,
      L_SHIPINSTRUCT STRING ,
      L_SHIPMODE     STRING ,
      L_COMMENT      STRING
    """
 
    df = spark.read.format("csv").schema(schema).option("sep", "|").load(filepath)
    return df
 
 
df = create_tpch_lineitem_dataframe(dbfs_path)
df.count()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/34f0acba-1a77-c9cd-d8c8-908931cec0df.png)


### TPC-H のデータの Parquet 形式への変換

```pyhon
parquet_file_paths = {
    "base": [
        f"{dbfs_path}/lineitem.tbl.1",
        f"{dbfs_path}/lineitem.tbl.2",
        f"{dbfs_path}/lineitem.tbl.3",
        f"{dbfs_path}/lineitem.tbl.4",
        f"{dbfs_path}/lineitem.tbl.5",
        f"{dbfs_path}/lineitem.tbl.6",
        f"{dbfs_path}/lineitem.tbl.7",
        f"{dbfs_path}/lineitem.tbl.8",
        f"{dbfs_path}/lineitem.tbl.9",
        f"{dbfs_path}/lineitem.tbl.10",
        f"{dbfs_path}/lineitem.tbl.11",
        f"{dbfs_path}/lineitem.tbl.12",
        f"{dbfs_path}/lineitem.tbl.13",
        f"{dbfs_path}/lineitem.tbl.14",
        f"{dbfs_path}/lineitem.tbl.15",
    ],
    "wiret_01": [
        f"{dbfs_path}/lineitem.tbl.1",
        f"{dbfs_path}/lineitem.tbl.16",
    ],
    "wiret_02": [
        f"{dbfs_path}/lineitem.tbl.4",
        f"{dbfs_path}/lineitem.tbl.17",
    ],
    "wiret_03": [
        f"{dbfs_path}/lineitem.tbl.7",
        f"{dbfs_path}/lineitem.tbl.18",
    ],
    "wiret_04": [
        f"{dbfs_path}/lineitem.tbl.10",
        f"{dbfs_path}/lineitem.tbl.19",
    ],
    "wiret_05": [
        f"{dbfs_path}/lineitem.tbl.13",
        f"{dbfs_path}/lineitem.tbl.20",
    ],
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a41c3b50-2c75-352e-3660-66d34fec862f.png)


```pyhon
base_dbfs_path = "dbfs:/FileStore/load_test"
 
parqeut_dbfs_path = f"{base_dbfs_path}/parquet"

def create_tpch_lineitem_dataframe(
    filepath,
):
 
    schema = """
      L_ORDERKEY    INTEGER ,
      L_PARTKEY     INTEGER ,
      L_SUPPKEY     INTEGER ,
      L_LINENUMBER  INTEGER ,
      L_QUANTITY    DECIMAL(15,2) ,
      L_EXTENDEDPRICE  DECIMAL(15,2) ,
      L_DISCOUNT    DECIMAL(15,2) ,
      L_TAX         DECIMAL(15,2) ,
      L_RETURNFLAG  STRING ,
      L_LINESTATUS  STRING ,
      L_SHIPDATE    DATE ,
      L_COMMITDATE  DATE ,
      L_RECEIPTDATE DATE ,
      L_SHIPINSTRUCT STRING ,
      L_SHIPMODE     STRING ,
      L_COMMENT      STRING
    """
 
    df = spark.read.format("csv").schema(schema).option("sep", "|").load(filepath)
    return df

```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/077c21de-9aa6-a929-dd57-917cc4c12cf9.png)


```pyhon
for path, src_path in parquet_file_paths.items():
    write_path = f"{parqeut_dbfs_path}/{path}"
 
    # 指定パスのデータフレームを作成
    raw_df = create_tpch_lineitem_dataframe(src_path)
 
    # Synapse docs に記載があるパラメータを設定
    spark.conf.set("spark.sql.parquet.writeLegacyFormat", "true")
 
    # parqeut 形式で書き込み
    raw_df.write.format("parquet").mode("overwrite").save(write_path)
 
    spark.conf.set("spark.sql.parquet.writeLegacyFormat", "false")
 
    print(f"{path} is finished.")
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7038d3cc-d4e1-e15d-9178-9e250f49e9f3.png)

```pyhon
# ファイル一覧を表示
file_list = dbutils.fs.ls(f"{parqeut_dbfs_path}")
display(file_list)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ec9c3fb7-e571-8e78-ab5a-a6b7e00a29b1.png)

### Parquet 形式のTPC-H のデータをストレージへの書き込み

```pyhon
storage_account = "synapsetest123456689343"
container_name = "synapse"
access_key = "09jIsfx97uwoW1Vc+CXalhXCuf2ONI+gJB0YyWXh5mGKVi8HmX8Q7HFV8cdXby7o9O4j181048B0+AStJWf7ug=="
spark.conf.set(
    f"fs.azure.account.key.{storage_account}.dfs.core.windows.net", access_key
)
 
# SAS キーで実施する場合には下記で実施
# storage_account = 'synapsetest123345'
# container_name = 'synapse-load'
# token = '?sv=2021-06-08&ss=b&srt=sco&sp=rwdlacyx&se=2022-11-04T18:45:33Z&st=2022-10-28T10:45:33Z&spr=https&sig=XJLZh9FVzHZvia2VYlTzfKHa7%2FYcKljfo%2F9z3Dx'
# spark.conf.set(f"fs.azure.account.auth.type.{storage_account}.dfs.core.windows.net", "SAS")
# spark.conf.set(f"fs.azure.sas.token.provider.type.{storage_account}.dfs.core.windows.net", "org.apache.hadoop.fs.azurebfs.sas.FixedSASTokenProvider")
# spark.conf.set(f"fs.azure.sas.fixed.token.{storage_account}.dfs.core.windows.net", token)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2445b356-7163-ddc8-1fb9-9c9f58143ed3.png)


```pyhon
# 接続確認
storage_dir = f"abfss://{container_name}@{storage_account}.dfs.core.windows.net"
dbutils.fs.ls(storage_dir)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f8c2815d-3194-2398-7afa-a92b5d0aa675.png)

```pyhon
tpch_dir = f"{storage_dir}/tpch"

# ストレージ上のディレクトリを初期化
dbutils.fs.rm(tpch_dir, True)
 
# `tpch` ディレクトリをストレージ側に作成後、dbfs 上の parquet ファイルをコピー
dbutils.fs.mkdirs(tpch_dir)
dbutils.fs.cp(parqeut_dbfs_path, tpch_dir, True)
 
# ストレージ上のファイルチェック
dbutils.fs.ls(tpch_dir)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ba95c15c-d4b2-1015-938b-b0e162611955.png)


### Synapse における検証用 SQL の生成

#### 変数をセット

```pyhon
storage_account = "synapsetest123456689343"
container_name = "synapse"
access_key = "09jIsfx97uwoW1Vc+CXalhXCuf2ONI+gJB0YyWXh5mGKVi8HmX8Q7HFV8cdXby7o9O4j181048B0+AStJWf7ug=="
spark.conf.set(
    f"fs.azure.account.key.{storage_account}.dfs.core.windows.net", access_key
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8e5d1ca1-f618-3ba2-4e7e-fe24fe885f4e.png)


```pyhon
# 接続確認
storage_dir = f"abfss://{container_name}@{storage_account}.dfs.core.windows.net"
dbutils.fs.ls(storage_dir)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a429c101-d098-8866-659b-37cc8ba79f17.png)


```pyhon
data_dir = "tpch"
base_path = f"{data_dir}/base/*.parquet"
wiret_01 = f"{data_dir}/wiret_01/*.parquet"
wiret_02 = f"{data_dir}/wiret_02/*.parquet"
wiret_03 = f"{data_dir}/wiret_03/*.parquet"
wiret_04 = f"{data_dir}/wiret_04/*.parquet"
wiret_05 = f"{data_dir}/wiret_05/*.parquet"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/852ef898-3fa6-e954-cfec-28eddeba9469.png)

#### ワークロードを作成

```pyhon
# ユーザー名を取得
sql = """
SELECT user_name();
"""
 
print(sql)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/74523dd0-c82b-1669-217f-c4812bf34a82.png)


```pyhon
# ユーザー名を設定
user_name = "dbo"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9f92790b-ba0d-be38-5d94-9484d19a5e0f.png)


```pyhon
sql = f"""
CREATE WORKLOAD GROUP DataLoads
WITH ( 
    MIN_PERCENTAGE_RESOURCE = 0
    ,CAP_PERCENTAGE_RESOURCE = 100
    ,REQUEST_MIN_RESOURCE_GRANT_PERCENT = 100
);
 
CREATE WORKLOAD CLASSIFIER [wgcELTLogin]
WITH (
    WORKLOAD_GROUP = 'DataLoads'
    ,MEMBERNAME = '{user_name}'
);
"""
 
print(sql)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e9877726-8358-5525-c220-6606352be2a8.png)


#### テーブルを作成

##### 複数列のハッシュ分散テーブルを作成

```pyhon
ddl = """
-- ハッシュ分散テーブル
ALTER DATABASE SCOPED CONFIGURATION SET DW_COMPATIBILITY_LEVEL = 9000;
 
IF OBJECT_ID (N'dbo.LINEITEM', N'U') IS NOT NULL  
DROP TABLE dbo.LINEITEM;
 
CREATE TABLE dbo.LINEITEM
(
    L_ORDERKEY      integer
   ,L_PARTKEY       integer
   ,L_SUPPKEY       integer
   ,L_LINENUMBER    integer
   ,L_QUANTITY      NUMERIC(15,2)
   ,L_EXTENDEDPRICE NUMERIC(15,2)
   ,L_DISCOUNT      NUMERIC(15,2)
   ,L_TAX           NUMERIC(15,2)
   ,L_RETURNFLAG    varchar(1)
   ,L_LINESTATUS    varchar(1)
   ,L_SHIPDATE      date
   ,L_COMMITDATE    date
   ,L_RECEIPTDATE   date
   ,L_SHIPINSTRUCT  varchar(25)
   ,L_SHIPMODE      varchar(10)
   ,L_COMMENT       varchar(44)
)
WITH
(
    DISTRIBUTION = HASH (L_ORDERKEY, L_LINENUMBER),
    HEAP
)
;
"""
 
print(ddl)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2d7de1e1-0201-c671-3b45-4463a20c4474.png)


##### 単一列のハッシュ分散テーブルを作成

```pyhon
ddl = """
IF OBJECT_ID (N'dbo.LINEITEM', N'U') IS NOT NULL  
DROP TABLE dbo.LINEITEM;
 
CREATE TABLE dbo.LINEITEM
(
    L_ORDERKEY      integer
   ,L_PARTKEY       integer
   ,L_SUPPKEY       integer
   ,L_LINENUMBER    integer
   ,L_QUANTITY      NUMERIC(15,2)
   ,L_EXTENDEDPRICE NUMERIC(15,2)
   ,L_DISCOUNT      NUMERIC(15,2)
   ,L_TAX           NUMERIC(15,2)
   ,L_RETURNFLAG    varchar(1)
   ,L_LINESTATUS    varchar(1)
   ,L_SHIPDATE      date
   ,L_COMMITDATE    date
   ,L_RECEIPTDATE   date
   ,L_SHIPINSTRUCT  varchar(25)
   ,L_SHIPMODE      varchar(10)
   ,L_COMMENT       varchar(44)
)
WITH
(
    DISTRIBUTION = HASH (L_LINENUMBER),
    HEAP
)
;
"""
 
print(ddl)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/dc812b01-78cc-ed94-99ab-fb1065ccc7f7.png)


##### ラウンドロビン分散テーブルを作成

```pyhon
ddl = """
-- ラウンドロビンテーブル
IF OBJECT_ID (N'dbo.LINEITEM', N'U') IS NOT NULL  
DROP TABLE dbo.LINEITEM;
CREATE TABLE dbo.LINEITEM
(
    L_ORDERKEY      integer
   ,L_PARTKEY       integer
   ,L_SUPPKEY       integer
   ,L_LINENUMBER    integer
   ,L_QUANTITY      NUMERIC(15,2)
   ,L_EXTENDEDPRICE NUMERIC(15,2)
   ,L_DISCOUNT      NUMERIC(15,2)
   ,L_TAX           NUMERIC(15,2)
   ,L_RETURNFLAG    varchar(1)
   ,L_LINESTATUS    varchar(1)
   ,L_SHIPDATE      date
   ,L_COMMITDATE    date
   ,L_RECEIPTDATE   date
   ,L_SHIPINSTRUCT  varchar(25)
   ,L_SHIPMODE      varchar(10)
   ,L_COMMENT       varchar(44)
)
WITH
(
    DISTRIBUTION = ROUND_ROBIN,
    HEAP
)  
;
"""
 
print(ddl)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2a8f9169-9c07-3cbf-e778-a2cc93a1899d.png)


#### 初期データをロード

```pyhon
sql = f"""
-- 44991725 レコード
IF OBJECT_ID(N'tempdb..#LINEITEM', N'U') IS NOT NULL
DROP TABLE #LINEITEM;
 
TRUNCATE TABLE dbo.LINEITEM;
 
COPY INTO #LINEITEM
FROM 'https://{storage_account}.blob.core.windows.net/{container_name}/{base_path}'
WITH (
    FILE_TYPE = 'PARQUET',
    CREDENTIAL = (IDENTITY= 'Storage Account Key', SECRET='{access_key}'),
    AUTO_CREATE_TABLE = 'ON'
)
;
 
INSERT INTO dbo.LINEITEM
SELECT * FROM #LINEITEM
 
select count(*) FROM dbo.LINEITEM;
"""
 
print(sql)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fc8b7468-a9b4-6a5d-2afd-72f556319d45.png)

#### Merge 処理

##### 1 回目のデータを更新

```pyhon
sql = f"""
IF OBJECT_ID(N'tempdb..#LINEITEM', N'U') IS NOT NULL
DROP TABLE #LINEITEM;
 
COPY INTO #LINEITEM
FROM 'https://{storage_account}.blob.core.windows.net/{container_name}/{wiret_01}'
WITH (
    FILE_TYPE = 'PARQUET',
    CREDENTIAL = (IDENTITY= 'Storage Account Key', SECRET='{access_key}'),
    AUTO_CREATE_TABLE = 'ON'
)
;
 
MERGE INTO 
    dbo.LINEITEM AS tgt  
USING
    #LINEITEM AS src  
ON 
(
    tgt.L_ORDERKEY = src.L_ORDERKEY
    and tgt.L_LINENUMBER = src.L_LINENUMBER
)
 WHEN MATCHED THEN
    UPDATE 
        SET
        tgt.L_ORDERKEY = src.L_ORDERKEY
        ,tgt.L_PARTKEY = src.L_PARTKEY
        ,tgt.L_SUPPKEY = src.L_SUPPKEY
        ,tgt.L_LINENUMBER = src.L_LINENUMBER
        ,tgt.L_QUANTITY = src.L_QUANTITY
        ,tgt.L_EXTENDEDPRICE = src.L_EXTENDEDPRICE
        ,tgt.L_DISCOUNT = src.L_DISCOUNT
        ,tgt.L_TAX = src.L_TAX
        ,tgt.L_RETURNFLAG = src.L_RETURNFLAG
        ,tgt.L_LINESTATUS = src.L_LINESTATUS
        ,tgt.L_SHIPDATE = src.L_SHIPDATE
        ,tgt.L_COMMITDATE = src.L_COMMITDATE
        ,tgt.L_RECEIPTDATE = src.L_RECEIPTDATE
        ,tgt.L_SHIPINSTRUCT = src.L_SHIPINSTRUCT
        ,tgt.L_SHIPMODE = src.L_SHIPMODE
        ,tgt.L_COMMENT = src.L_COMMENT
WHEN NOT MATCHED THEN
    INSERT
    (
        L_ORDERKEY
        ,L_PARTKEY
        ,L_SUPPKEY
        ,L_LINENUMBER
        ,L_QUANTITY
        ,L_EXTENDEDPRICE
        ,L_DISCOUNT
        ,L_TAX
        ,L_RETURNFLAG
        ,L_LINESTATUS
        ,L_SHIPDATE
        ,L_COMMITDATE
        ,L_RECEIPTDATE
        ,L_SHIPINSTRUCT
        ,L_SHIPMODE
        ,L_COMMENT
    )
    VALUES 
    (
        L_ORDERKEY
        ,L_PARTKEY
        ,L_SUPPKEY
        ,L_LINENUMBER
        ,L_QUANTITY
        ,L_EXTENDEDPRICE
        ,L_DISCOUNT
        ,L_TAX
        ,L_RETURNFLAG
        ,L_LINESTATUS
        ,L_SHIPDATE
        ,L_COMMITDATE
        ,L_RECEIPTDATE
        ,L_SHIPINSTRUCT
        ,L_SHIPMODE
        ,L_COMMENT
    )
;
"""
 
print(sql)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ec23a349-0f64-d707-779b-896d54812f4a.png)

##### 2 回目のデータを更新

```pyhon
sql = f"""
IF OBJECT_ID(N'tempdb..#LINEITEM', N'U') IS NOT NULL
DROP TABLE #LINEITEM;
 
COPY INTO #LINEITEM
FROM 'https://{storage_account}.blob.core.windows.net/{container_name}/{wiret_02}'
WITH (
    FILE_TYPE = 'PARQUET',
    CREDENTIAL = (IDENTITY= 'Storage Account Key', SECRET='{access_key}'),
    AUTO_CREATE_TABLE = 'ON'
)
;
 
MERGE INTO 
    dbo.LINEITEM AS tgt  
USING
    #LINEITEM AS src  
ON 
(
    tgt.L_ORDERKEY = src.L_ORDERKEY
    and tgt.L_LINENUMBER = src.L_LINENUMBER
)
 WHEN MATCHED THEN
    UPDATE 
        SET
        tgt.L_ORDERKEY = src.L_ORDERKEY
        ,tgt.L_PARTKEY = src.L_PARTKEY
        ,tgt.L_SUPPKEY = src.L_SUPPKEY
        ,tgt.L_LINENUMBER = src.L_LINENUMBER
        ,tgt.L_QUANTITY = src.L_QUANTITY
        ,tgt.L_EXTENDEDPRICE = src.L_EXTENDEDPRICE
        ,tgt.L_DISCOUNT = src.L_DISCOUNT
        ,tgt.L_TAX = src.L_TAX
        ,tgt.L_RETURNFLAG = src.L_RETURNFLAG
        ,tgt.L_LINESTATUS = src.L_LINESTATUS
        ,tgt.L_SHIPDATE = src.L_SHIPDATE
        ,tgt.L_COMMITDATE = src.L_COMMITDATE
        ,tgt.L_RECEIPTDATE = src.L_RECEIPTDATE
        ,tgt.L_SHIPINSTRUCT = src.L_SHIPINSTRUCT
        ,tgt.L_SHIPMODE = src.L_SHIPMODE
        ,tgt.L_COMMENT = src.L_COMMENT
WHEN NOT MATCHED THEN
    INSERT
    (
        L_ORDERKEY
        ,L_PARTKEY
        ,L_SUPPKEY
        ,L_LINENUMBER
        ,L_QUANTITY
        ,L_EXTENDEDPRICE
        ,L_DISCOUNT
        ,L_TAX
        ,L_RETURNFLAG
        ,L_LINESTATUS
        ,L_SHIPDATE
        ,L_COMMITDATE
        ,L_RECEIPTDATE
        ,L_SHIPINSTRUCT
        ,L_SHIPMODE
        ,L_COMMENT
    )
    VALUES 
    (
        L_ORDERKEY
        ,L_PARTKEY
        ,L_SUPPKEY
        ,L_LINENUMBER
        ,L_QUANTITY
        ,L_EXTENDEDPRICE
        ,L_DISCOUNT
        ,L_TAX
        ,L_RETURNFLAG
        ,L_LINESTATUS
        ,L_SHIPDATE
        ,L_COMMITDATE
        ,L_RECEIPTDATE
        ,L_SHIPINSTRUCT
        ,L_SHIPMODE
        ,L_COMMENT
    )
;
"""
 
print(sql)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/709d2f6e-992c-f7f2-3204-b121470a7a56.png)


##### 3 回目のデータを更新

```pyhon
sql = f"""
IF OBJECT_ID(N'tempdb..#LINEITEM', N'U') IS NOT NULL
DROP TABLE #LINEITEM;
 
COPY INTO #LINEITEM
FROM 'https://{storage_account}.blob.core.windows.net/{container_name}/{wiret_03}'
WITH (
    FILE_TYPE = 'PARQUET',
    CREDENTIAL = (IDENTITY= 'Storage Account Key', SECRET='{access_key}'),
    AUTO_CREATE_TABLE = 'ON'
)
;
 
MERGE INTO 
    dbo.LINEITEM AS tgt  
USING
    #LINEITEM AS src  
ON 
(
    tgt.L_ORDERKEY = src.L_ORDERKEY
    and tgt.L_LINENUMBER = src.L_LINENUMBER
)
 WHEN MATCHED THEN
    UPDATE 
        SET
        tgt.L_ORDERKEY = src.L_ORDERKEY
        ,tgt.L_PARTKEY = src.L_PARTKEY
        ,tgt.L_SUPPKEY = src.L_SUPPKEY
        ,tgt.L_LINENUMBER = src.L_LINENUMBER
        ,tgt.L_QUANTITY = src.L_QUANTITY
        ,tgt.L_EXTENDEDPRICE = src.L_EXTENDEDPRICE
        ,tgt.L_DISCOUNT = src.L_DISCOUNT
        ,tgt.L_TAX = src.L_TAX
        ,tgt.L_RETURNFLAG = src.L_RETURNFLAG
        ,tgt.L_LINESTATUS = src.L_LINESTATUS
        ,tgt.L_SHIPDATE = src.L_SHIPDATE
        ,tgt.L_COMMITDATE = src.L_COMMITDATE
        ,tgt.L_RECEIPTDATE = src.L_RECEIPTDATE
        ,tgt.L_SHIPINSTRUCT = src.L_SHIPINSTRUCT
        ,tgt.L_SHIPMODE = src.L_SHIPMODE
        ,tgt.L_COMMENT = src.L_COMMENT
WHEN NOT MATCHED THEN
    INSERT
    (
        L_ORDERKEY
        ,L_PARTKEY
        ,L_SUPPKEY
        ,L_LINENUMBER
        ,L_QUANTITY
        ,L_EXTENDEDPRICE
        ,L_DISCOUNT
        ,L_TAX
        ,L_RETURNFLAG
        ,L_LINESTATUS
        ,L_SHIPDATE
        ,L_COMMITDATE
        ,L_RECEIPTDATE
        ,L_SHIPINSTRUCT
        ,L_SHIPMODE
        ,L_COMMENT
    )
    VALUES 
    (
        L_ORDERKEY
        ,L_PARTKEY
        ,L_SUPPKEY
        ,L_LINENUMBER
        ,L_QUANTITY
        ,L_EXTENDEDPRICE
        ,L_DISCOUNT
        ,L_TAX
        ,L_RETURNFLAG
        ,L_LINESTATUS
        ,L_SHIPDATE
        ,L_COMMITDATE
        ,L_RECEIPTDATE
        ,L_SHIPINSTRUCT
        ,L_SHIPMODE
        ,L_COMMENT
    )
;
"""
 
print(sql)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/574f3c6e-d1ef-0b52-c18d-5feac13ee9f9.png)

##### 4 回目のデータを更新

```pyhon
sql = f"""
IF OBJECT_ID(N'tempdb..#LINEITEM', N'U') IS NOT NULL
DROP TABLE #LINEITEM;
 
COPY INTO #LINEITEM
FROM 'https://{storage_account}.blob.core.windows.net/{container_name}/{wiret_04}'
WITH (
    FILE_TYPE = 'PARQUET',
    CREDENTIAL = (IDENTITY= 'Storage Account Key', SECRET='{access_key}'),
    AUTO_CREATE_TABLE = 'ON'
)
;
 
MERGE INTO 
    dbo.LINEITEM AS tgt  
USING
    #LINEITEM AS src  
ON 
(
    tgt.L_ORDERKEY = src.L_ORDERKEY
    and tgt.L_LINENUMBER = src.L_LINENUMBER
)
 WHEN MATCHED THEN
    UPDATE 
        SET
        tgt.L_ORDERKEY = src.L_ORDERKEY
        ,tgt.L_PARTKEY = src.L_PARTKEY
        ,tgt.L_SUPPKEY = src.L_SUPPKEY
        ,tgt.L_LINENUMBER = src.L_LINENUMBER
        ,tgt.L_QUANTITY = src.L_QUANTITY
        ,tgt.L_EXTENDEDPRICE = src.L_EXTENDEDPRICE
        ,tgt.L_DISCOUNT = src.L_DISCOUNT
        ,tgt.L_TAX = src.L_TAX
        ,tgt.L_RETURNFLAG = src.L_RETURNFLAG
        ,tgt.L_LINESTATUS = src.L_LINESTATUS
        ,tgt.L_SHIPDATE = src.L_SHIPDATE
        ,tgt.L_COMMITDATE = src.L_COMMITDATE
        ,tgt.L_RECEIPTDATE = src.L_RECEIPTDATE
        ,tgt.L_SHIPINSTRUCT = src.L_SHIPINSTRUCT
        ,tgt.L_SHIPMODE = src.L_SHIPMODE
        ,tgt.L_COMMENT = src.L_COMMENT
WHEN NOT MATCHED THEN
    INSERT
    (
        L_ORDERKEY
        ,L_PARTKEY
        ,L_SUPPKEY
        ,L_LINENUMBER
        ,L_QUANTITY
        ,L_EXTENDEDPRICE
        ,L_DISCOUNT
        ,L_TAX
        ,L_RETURNFLAG
        ,L_LINESTATUS
        ,L_SHIPDATE
        ,L_COMMITDATE
        ,L_RECEIPTDATE
        ,L_SHIPINSTRUCT
        ,L_SHIPMODE
        ,L_COMMENT
    )
    VALUES 
    (
        L_ORDERKEY
        ,L_PARTKEY
        ,L_SUPPKEY
        ,L_LINENUMBER
        ,L_QUANTITY
        ,L_EXTENDEDPRICE
        ,L_DISCOUNT
        ,L_TAX
        ,L_RETURNFLAG
        ,L_LINESTATUS
        ,L_SHIPDATE
        ,L_COMMITDATE
        ,L_RECEIPTDATE
        ,L_SHIPINSTRUCT
        ,L_SHIPMODE
        ,L_COMMENT
    )
;
"""
 
print(sql)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9598468a-f604-3b63-ed7f-53a3cb249481.png)

##### 5 回目のデータを更新

```pyhon
sql = f"""
IF OBJECT_ID(N'tempdb..#LINEITEM', N'U') IS NOT NULL
DROP TABLE #LINEITEM;
 
COPY INTO #LINEITEM
FROM 'https://{storage_account}.blob.core.windows.net/{container_name}/{wiret_05}'
WITH (
    FILE_TYPE = 'PARQUET',
    CREDENTIAL = (IDENTITY= 'Storage Account Key', SECRET='{access_key}'),
    AUTO_CREATE_TABLE = 'ON'
)
;
 
MERGE INTO 
    dbo.LINEITEM AS tgt  
USING
    #LINEITEM AS src  
ON 
(
    tgt.L_ORDERKEY = src.L_ORDERKEY
    and tgt.L_LINENUMBER = src.L_LINENUMBER
)
 WHEN MATCHED THEN
    UPDATE 
        SET
        tgt.L_ORDERKEY = src.L_ORDERKEY
        ,tgt.L_PARTKEY = src.L_PARTKEY
        ,tgt.L_SUPPKEY = src.L_SUPPKEY
        ,tgt.L_LINENUMBER = src.L_LINENUMBER
        ,tgt.L_QUANTITY = src.L_QUANTITY
        ,tgt.L_EXTENDEDPRICE = src.L_EXTENDEDPRICE
        ,tgt.L_DISCOUNT = src.L_DISCOUNT
        ,tgt.L_TAX = src.L_TAX
        ,tgt.L_RETURNFLAG = src.L_RETURNFLAG
        ,tgt.L_LINESTATUS = src.L_LINESTATUS
        ,tgt.L_SHIPDATE = src.L_SHIPDATE
        ,tgt.L_COMMITDATE = src.L_COMMITDATE
        ,tgt.L_RECEIPTDATE = src.L_RECEIPTDATE
        ,tgt.L_SHIPINSTRUCT = src.L_SHIPINSTRUCT
        ,tgt.L_SHIPMODE = src.L_SHIPMODE
        ,tgt.L_COMMENT = src.L_COMMENT
WHEN NOT MATCHED THEN
    INSERT
    (
        L_ORDERKEY
        ,L_PARTKEY
        ,L_SUPPKEY
        ,L_LINENUMBER
        ,L_QUANTITY
        ,L_EXTENDEDPRICE
        ,L_DISCOUNT
        ,L_TAX
        ,L_RETURNFLAG
        ,L_LINESTATUS
        ,L_SHIPDATE
        ,L_COMMITDATE
        ,L_RECEIPTDATE
        ,L_SHIPINSTRUCT
        ,L_SHIPMODE
        ,L_COMMENT
    )
    VALUES 
    (
        L_ORDERKEY
        ,L_PARTKEY
        ,L_SUPPKEY
        ,L_LINENUMBER
        ,L_QUANTITY
        ,L_EXTENDEDPRICE
        ,L_DISCOUNT
        ,L_TAX
        ,L_RETURNFLAG
        ,L_LINESTATUS
        ,L_SHIPDATE
        ,L_COMMITDATE
        ,L_RECEIPTDATE
        ,L_SHIPINSTRUCT
        ,L_SHIPMODE
        ,L_COMMENT
    )
;
"""
 
print(sql)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e90972fc-d51f-55ef-3846-b664725d45ed.png)