Azure Data Factory または Synapse Analytics を使用して、Azure Database for PostgreSQL のデータをコピーして変換する

[アーティクル]
01/05/2024

適用対象: Azure Data Factory Azure Synapse Analytics

ヒント

企業向けのオールインワン分析ソリューション、Microsoft Fabric の Data Factory をお試しください。 Microsoft Fabric は、データ移動からデータサイエンス、リアルタイム分析、ビジネスインテリジェンス、レポートまで、あらゆるものをカバーしています。無料で新しい試用版を開始する方法について説明します。

この記事では、Azure Data Factory および Synapse Analytics パイプラインの Copy アクティビティを使用して、Azure Database for PostgreSQL との間でデータをコピーする方法、および Data Flow を使用して Azure Database for PostgreSQL のデータを変換する方法について説明します。詳細については、Azure Data Factory および Azure Synapse Analytics の概要記事を参照してください。

このコネクタは、Azure Database for PostgreSQL サービス用に特化しています。オンプレミスまたはクラウドにある汎用 PostgreSQL データベースからデータをコピーするには、PostgreSQL コネクタを使用します。

サポートされる機能

この Azure Database for PostgreSQL コネクタは、以下の機能でサポートされています。

サポートされる機能	IR	マネージドプライベートエンドポイント
Copy アクティビティ (ソース/シンク)	1.1	✓
マッピングデータフロー (ソース/シンク)	1	✓
Lookup アクティビティ	1.1	✓

① Azure 統合ランタイム ② セルフホステッド統合ランタイム

3 つのアクティビティは、すべての Azure Database for PostgreSQL デプロイオプションで機能します。

作業の開始

パイプラインでコピーアクティビティを実行するには、次のいずれかのツールまたは SDK を使用します。

UI を使用して Azure Database for PostgreSQL へのリンクサービスを作成する

次の手順を使用して、Azure portal UI で Azure database for PostgreSQL のリンクサービスを作成します。

Azure Data Factory または Synapse ワークスペースの [管理] タブに移動し、[リンクサービス] を選択して、[新規] をクリックします。
- Azure Data Factory
- Azure Synapse
PostgreSQL を検索し、Azure database for PostgreSQL コネクタを選択します。
サービスの詳細を構成し、接続をテストして、新しいリンクサービスを作成します。

コネクタの構成の詳細

以下のセクションでは、Azure Database for PostgreSQL コネクタに固有の Data Factory エンティティの定義に使用されるプロパティについて詳しく説明します。

リンクされたサービスのプロパティ

Azure Database for PostgreSQL のリンクされたサービスでは、次のプロパティがサポートされます。

プロパティ	Description	必須
type	type プロパティは、次のように設定する必要があります:AzurePostgreSql	はい
connectionString	Azure Database for PostgreSQL に接続するための ODBC 接続文字列。パスワードを Azure Key Vault に格納して、接続文字列から `password` 構成をプルすることもできます。詳細については、下記の例と、「Azure Key Vault への資格情報の格納」を参照してください。	はい
connectVia	このプロパティは、データストアに接続するために使用される統合ランタイムを表します。 Azure 統合ランタイムまたは自己ホスト型統合ランタイム (データストアがプライベートネットワークにある場合) を使用できます。指定されていない場合は、既定の Azure 統合ランタイムが使用されます。	いいえ

一般的な接続文字列は Server=<server>.postgres.database.azure.com;Database=<database>;Port=<port>;UID=<username>;Password=<Password> です。ケースごとにさらに多くのプロパティを設定できます。それらのプロパティを次に示します。

プロパティ	説明	Options	必須
EncryptionMethod (EM)	ドライバーとデータベースサーバー間で送信されるデータを暗号化するためにドライバーが使用するメソッド。たとえば、`EncryptionMethod=<0/1/6>;`	0 (暗号化なし) (既定) /1 (SSL)/6 (RequestSSL)	いいえ
ValidateServerCertificate (VSC)	SSL 暗号化が有効 (Encryption Method=1) になっているときに、データベースサーバーによって送信される証明書をドライバーが検証するかどうかを決定します。たとえば、`ValidateServerCertificate=<0/1>;`	0 (無効) (既定) / 1 (有効)	いいえ

例:

{
    "name": "AzurePostgreSqlLinkedService",
    "properties": {
        "type": "AzurePostgreSql",
        "typeProperties": {
            "connectionString": "Server=<server>.postgres.database.azure.com;Database=<database>;Port=<port>;UID=<username>;Password=<Password>"
        }
    }
}

例:

Azure Key Vault にパスワードを格納する

{
    "name": "AzurePostgreSqlLinkedService",
    "properties": {
        "type": "AzurePostgreSql",
        "typeProperties": {
            "connectionString": "Server=<server>.postgres.database.azure.com;Database=<database>;Port=<port>;UID=<username>;",
            "password": { 
                "type": "AzureKeyVaultSecret", 
                "store": { 
                    "referenceName": "<Azure Key Vault linked service name>", 
                    "type": "LinkedServiceReference" 
                }, 
                "secretName": "<secretName>" 
            }
        }
    }
}

データセットのプロパティ

データセットの定義に使用できるセクションとプロパティの一覧については、データセットに関する記事をご覧ください。このセクションでは、Azure Database for PostgreSQL がデータセットでサポートするプロパティの一覧を示します。

Azure Database for PostgreSQL からデータをコピーするには、データセットの type プロパティを AzurePostgreSqlTable に設定します。次のプロパティがサポートされています。

プロパティ	Description	必須
type	データセットの type プロパティは、AzurePostgreSqlTable に設定する必要があります	はい
tableName	テーブルの名前	いいえ (アクティビティソースの "query" が指定されている場合)

例:

{
    "name": "AzurePostgreSqlDataset",
    "properties": {
        "type": "AzurePostgreSqlTable",
        "linkedServiceName": {
            "referenceName": "<AzurePostgreSql linked service name>",
            "type": "LinkedServiceReference"
        },
        "typeProperties": {}
    }
}

コピーアクティビティのプロパティ

アクティビティの定義に利用できるセクションとプロパティの完全な一覧については、パイプラインおよびアクティビティに関するページを参照してください。このセクションでは、Azure Database for PostgreSQL ソースでサポートされるプロパティの一覧を示します。

ソースとしての Azure Database for PostgreSql

Azure Database for PostgreSQL からデータをコピーするには、コピーアクティビティのソースの種類を AzurePostgreSqlSource に設定します。コピーアクティビティの source セクションでは、次のプロパティがサポートされます。

プロパティ	Description	必須
type	コピーアクティビティのソースの type プロパティは AzurePostgreSqlSource に設定する必要があります	はい
query	カスタム SQL クエリを使用してデータを読み取ります。たとえば、`SELECT * FROM mytable` や `SELECT * FROM "MyTable"` などです。 PostgreSQL では、エンティティ名が引用符で囲まれていない場合、大文字と小文字が区別されません。	いいえ (データセットの tableName プロパティが指定されている場合)
partitionOptions	Azure SQL Database からのデータの読み込みに使用されるデータパーティション分割オプションを指定します。使用できる値は、以下のとおりです。None (既定値)、PhysicalPartitionsOfTable、および DynamicRange。パーティションオプションが有効になっている場合 (つまり、`None` ではない場合)、Azure SQL Database から同時にデータを読み込む並列処理の次数は、コピーアクティビティの `parallelCopies` の設定によって制御されます。	いいえ
partitionSettings	データパーティション分割の設定のグループを指定します。パーティションオプションが `None` でない場合に適用されます。	いいえ
`partitionSettings` の下:
partitionNames	コピーする必要がある物理パーティションのリスト。パーティションオプションが `PhysicalPartitionsOfTable` である場合に適用されます。クエリを使用してソースデータを取得する場合は、WHERE 句で `?AdfTabularPartitionName` をフックします。例については、「Azure Database for PostgreSQL からの並列コピー」セクションを参照してください。	いいえ
partitionColumnName	並列コピーの範囲パーティション分割で使用される整数型または日付/日時型 (`int`、`smallint`、`bigint`、`date`、`timestamp without time zone`、`timestamp with time zone` または `time without time zone`) のソース列の名前を指定します。指定されていない場合は、テーブルの主キーが自動検出され、パーティション列として使用されます。パーティションオプションが `DynamicRange` である場合に適用されます。クエリを使用してソースデータを取得する場合は、WHERE 句で `?AdfRangePartitionColumnName` をフックします。例については、「Azure Database for PostgreSQL からの並列コピー」セクションを参照してください。	いいえ
partitionUpperBound	データをコピーするパーティション列の最大値。パーティションオプションが `DynamicRange` である場合に適用されます。クエリを使用してソースデータを取得する場合は、WHERE 句で `?AdfRangePartitionUpbound` をフックします。例については、「Azure Database for PostgreSQL からの並列コピー」セクションを参照してください。	いいえ
partitionLowerBound	データをコピーするパーティション列の最小値。パーティションオプションが `DynamicRange` である場合に適用されます。クエリを使用してソースデータを取得する場合は、WHERE 句で `?AdfRangePartitionLowbound` をフックします。例については、「Azure Database for PostgreSQL からの並列コピー」セクションを参照してください。	いいえ

例:

"activities":[
    {
        "name": "CopyFromAzurePostgreSql",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<AzurePostgreSql input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "AzurePostgreSqlSource",
                "query": "<custom query e.g. SELECT * FROM mytable>"
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

シンクとしての Azure Database for PostgreSQL

データを Azure Database for PostgreSQL にコピーするために、コピーアクティビティの sink セクションでは以下のプロパティがサポートされています。

プロパティ	Description	必須
type	コピーアクティビティのシンクの type プロパティは AzurePostgreSQLSink に設定する必要があります	はい
preCopyScript	コピーアクティビティの毎回の実行で、データを Azure Database for PostgreSQL に書き込む前に実行する SQL クエリを指定します。このプロパティを使用して、事前に読み込まれたデータをクリーンアップできます。	いいえ
writeMethod	Azure Database for PostgreSQL にデータを書き込むために使用するメソッド。使用できる値は、CopyCommand (規定値で、パフォーマンスがより高い)、BulkInsert です。	いいえ
writeBatchSize	バッチごとに Azure Database for PostgreSQL に読み込まれる行の数。許可される値は行数を表す整数です。	いいえ (既定値は 1,000,000)
writeBatchTimeout	タイムアウトする前に一括挿入操作の完了を待つ時間です。 Timespan 文字列を値として使用できます。たとえば "00:30:00" (30 分) を指定できます。	いいえ (既定値は 00:30:00)

例:

"activities":[
    {
        "name": "CopyToAzureDatabaseForPostgreSQL",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Azure PostgreSQL output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "AzurePostgreSQLSink",
                "preCopyScript": "<custom SQL script>",
                "writeMethod": "CopyCommand",
                "writeBatchSize": 1000000
            }
        }
    }
]

Azure Database for PostgreSQL からの並列コピー

Azure Database for PostgreSQL コネクタでは、Copy アクティビティの際に、データを並列でコピーするための組み込みのデータパーティション分割が提供されます。データパーティション分割オプションは、コピーアクティビティの [ソース] タブにあります。

Screenshot of partition options

パーティション分割されるコピーを有効にすると、Copy アクティビティによって Azure Database for PostgreSQL ソースに対する並列クエリが実行され、パーティションごとにデータが読み込まれます。並列度は、コピーアクティビティの parallelCopies 設定によって制御されます。たとえば、parallelCopies を 4 に設定した場合、指定したパーティションオプションと設定に基づいて 4 つのクエリが同時に生成され、実行されます。各クエリでは、Azure Database for PostgreSQL からデータの一部を取得します。

Azure Database for PostgreSQL から大量のデータを読み込む場合は特に、データパーティション分割を行う並列コピーを有効にすることが推奨されます。さまざまなシナリオの推奨構成を以下に示します。ファイルベースのデータストアにデータをコピーする場合は、複数のファイルとしてフォルダーに書き込む (フォルダー名のみを指定する) ことをお勧めします。この場合、1 つのファイルに書き込むよりもパフォーマンスが優れています。

シナリオ	推奨設定
物理パーティションに分割された大きなテーブル全体から読み込む。	パーティションオプション: テーブルの物理パーティション。実行中に、サービスによって物理パーティションが自動的に検出され、パーティションごとにデータがコピーされます。
物理パーティションがなく、データパーティション分割用の整数列がある大きなテーブル全体から読み込む。	パーティションオプション: 動的範囲パーティション。パーティション列: データのパーティション分割に使用される列を指定します。指定されていない場合は、主キー列が使用されます。
カスタムクエリを使用して大量のデータを読み込む (物理パーティションがある場合)。	パーティションオプション: テーブルの物理パーティション。クエリ: `SELECT * FROM ?AdfTabularPartitionName WHERE <your_additional_where_clause>` パーティション名: データのコピー元のパーティション名を指定します。指定されていない場合は、PostgreSQL データセットで指定したテーブルの物理パーティションがサービスによって自動検出されます。実行中に、サービスによって `?AdfTabularPartitionName` が実際のパーティション名に置き換えられ、Azure Database for PostgreSQL に送信されます。
カスタムクエリを使用して大量のデータを読み込む (物理パーティションがなく、データパーティション分割用の整数列がある場合)。	パーティションオプション: 動的範囲パーティション。クエリ: `SELECT * FROM ?AdfTabularPartitionName WHERE ?AdfRangePartitionColumnName <= ?AdfRangePartitionUpbound AND ?AdfRangePartitionColumnName >= ?AdfRangePartitionLowbound AND <your_additional_where_clause>` パーティション列: データのパーティション分割に使用される列を指定します。整数データ型または date/datetime データ型の列に対してパーティション分割を実行できます。パーティションの上限とパーティションの下限: パーティション列に対してフィルター処理を実行して、下限から上限までの範囲内のデータのみを取得する場合に指定します。実行中に、`?AdfRangePartitionColumnName`、`?AdfRangePartitionUpbound`、`?AdfRangePartitionLowbound` が各パーティションの実際の列名と値の範囲に置き換えられ、Azure Database for PostgreSQL に送信されます。たとえば、パーティション列 "ID" で下限が 1、上限が 80 に設定され、並列コピーが 4 に設定されている場合、サービスは 4 つのパーティションでデータを取得します。これらの ID の範囲はそれぞれ [1, 20]、[21, 40]、[41, 60]、[61, 80] です。

パーティションオプションを使用してデータを読み込む場合のベストプラクティス:

データスキューを回避するため、パーティション列 (主キーや一意キーなど) には特徴のある列を選択します。
テーブルに組み込みパーティションがある場合は、パフォーマンスを向上させるためにパーティションオプションとして "テーブルの物理パーティション" を使用します。
Azure Integration Runtime を使用してデータをコピーする場合は、より大きな (4 より大きい) "データ統合単位 (DIU)" (>4) を設定すると、より多くのコンピューティングリソースを利用できます。そこで、該当するシナリオを確認してください。
パーティション数は、"コピーの並列処理の次数" によって制御されます。この数値を大きくしすぎるとパフォーマンスが低下するため、この数値は、(DIU またはセルフホステッド IR ノードの数) x (2 から 4) に設定することをお勧めします。

例: 複数の物理パーティションがある大きなテーブル全体から読み込む

"source": {
    "type": "AzurePostgreSqlSource",
    "partitionOption": "PhysicalPartitionsOfTable"
}

例: 動的範囲パーティションを使用してクエリを実行する

"source": {
    "type": "AzurePostgreSqlSource",
    "query": "SELECT * FROM <TableName> WHERE ?AdfDynamicRangePartitionCondition AND <your_additional_where_clause>",
    "partitionOption": "DynamicRange",
    "partitionSettings": {
        "partitionColumnName": "<partition_column_name>",
        "partitionUpperBound": "<upper_value_of_partition_column (optional) to decide the partition stride, not as data filter>",
        "partitionLowerBound": "<lower_value_of_partition_column (optional) to decide the partition stride, not as data filter>"
    }
}

Mapping Data Flow のプロパティ

マッピングデータフローでデータを変換する場合、Azure Database for PostgreSQL からテーブルの読み取りと書き込みを実行できます。詳細については、マッピングデータフローのソース変換とシンク変換に関する記事をご覧ください。ソースとシンクの種類として、Azure Database for PostgreSQL データセットまたはインラインデータセットを使用することができます。

ソース変換

次の表に、Azure Database for PostgreSQL ソースでサポートされるプロパティの一覧を示します。これらのプロパティは、 [ソースオプション] タブで編集できます。

名前	説明	必須	使用できる値	データフロースクリプトのプロパティ
テーブル	[テーブル] を入力として選択した場合、データセットで指定されたテーブルからすべてのデータがデータフローによってフェッチされます。	いいえ	-	(インラインデータセットのみ) tableName
クエリ	[クエリ] を入力として選択した場合は、ソースからデータをフェッチする SQL クエリを指定します。これにより、データセットで指定したテーブルがオーバーライドされます。テストまたはルックアップ対象の行を減らすうえで、クエリの使用は有効な手段です。 Order By 句はサポートされていませんが、完全な SELECT FROM ステートメントを設定することができます。ユーザー定義のテーブル関数を使用することもできます。 *select from udfGetData()** は、データフローで使用できるテーブルを返す SQL の UDF です。クエリ例: `select * from mytable where customerId > 1000 and customerId < 2000` または `select * from "MyTable"`。 PostgreSQL では、エンティティ名が引用符で囲まれていない場合、大文字と小文字が区別されません。	いいえ	String	query
スキーマ名	入力としてストアドプロシージャを選択する場合、ストアドプロシージャのスキーマを指定するか、[更新] を選択し、スキーマ名を検出するようにサービスに要求します。	いいえ	String	schemaName
ストアドプロシージャ	入力として [ストアドプロシージャ] を選択する場合、ソーステーブルからデータを読み込むストアドプロシージャの名前を指定するか、[更新] を選択し、プロシージャ名を検出するようにサービスに要求します。	はい ([ストアドプロシージャ] を入力として選択した場合)	String	procedureName
プロシージャのパラメーター	入力として [ストアドプロシージャ] を選択した場合、プロシージャで設定された順序でストアドプロシージャの入力パラメーターを指定するか、[インポート] を選択し、フォーム `@paraName` を使用してすべてのプロシージャパラメーターをインポートします。	いいえ	Array	inputs
バッチサイズ	大量データをバッチにまとめるバッチサイズを指定します。	いいえ	Integer	batchSize
Isolation Level	次のいずれかの分離レベルを選択します。 - コミットされたものを読み取り - コミットされていないものを読み取り (既定値) - 反復可能読み取り - シリアル化可能 - なし (分離レベルを無視)	いいえ	READ_COMMITTED READ_UNCOMMITTED REPEATABLE_READ SERIALIZABLE NONE	isolationLevel

Azure Database for PostgreSQL ソーススクリプトの例

ソースの種類として Azure Database for PostgreSQL を使用する場合、関連付けられているデータフロースクリプトは次のようになります。

source(allowSchemaDrift: true,
    validateSchema: false,
    isolationLevel: 'READ_UNCOMMITTED',
    query: 'select * from mytable',
    format: 'query') ~> AzurePostgreSQLSource

シンク変換

次の表に、Azure Database for PostgreSQL シンクでサポートされるプロパティの一覧を示します。これらのプロパティは、 [シンクオプション] タブで編集できます。

名前	説明	必須	使用できる値	データフロースクリプトのプロパティ
更新方法	対象となるデータベースに対して許可される操作を指定します。既定では、挿入のみが許可されます。行を更新、アップサート、または削除するには、それらのアクションに対して行をタグ付けするために行の変更変換が必要になります。	はい	`true` または `false`	deletable insertable updateable upsertable
[キー列]	更新、upsert、削除の場合、キー列 (複数可) を設定して、変更する行を決定する必要があります。キーとして選択する列の名前は、後続の更新、upsert、削除の一部として使用されます。そのため、シンクマッピングに存在する列を選択する必要があります。	いいえ	Array	キー
Skip writing key columns\(キー列の書き込みをスキップする)	キー列に値を書き込まない場合は、[Skip writing key columns](キー列の書き込みをスキップする) を選択します。	いいえ	`true` または `false`	skipKeyWrites
テーブルアクション	書き込み前に変換先テーブルのすべての行を再作成するか削除するかを指定します。 - なし: テーブルに対してアクションは実行されません。 - Recreate:テーブルが削除され、再作成されます。新しいテーブルを動的に作成する場合に必要です。 - Truncate:ターゲットテーブルのすべての行が削除されます。	いいえ	`true` または `false`	recreate truncate
バッチサイズ	各バッチで書き込まれる行の数を指定します。バッチサイズを大きくすると、圧縮とメモリの最適化が向上しますが、データをキャッシュする際にメモリ不足の例外が発生するリスクがあります。	いいえ	Integer	batchSize
ユーザー DB スキーマの選択	既定では、シンクスキーマの下にステージングとして一時テーブルが作成されます。または、[シンクスキーマを使用する] オプションをオフにして、Data Factory でステージングテーブルを作成してアップストリームデータを読み込み、完了時に自動的にクリーンアップするスキーマ名を指定することもできます。データベースにテーブルの作成権限があり、スキーマに対する変更権限があることを確認します。	いいえ	String	stagingSchemaName
事前および事後の SQL スクリプト	データがシンクデータベースに書き込まれる前 (前処理) と書き込まれた後 (後処理) に実行される複数行の SQL スクリプトを指定します。	いいえ	String	preSQLs postSQLs

ヒント

複数のコマンドを含む単一のバッチスクリプトを複数のバッチに分割することをお勧めします。
バッチの一部として実行できるのは、単純に更新数を返すデータ操作言語 (DML) ステートメントおよびデータ定義言語 (DDL) ステートメントだけです。詳細については、「バッチ操作の実行」を参照してください。

増分抽出を有効にする: このオプションを使用して、パイプラインが最後に実行されてから変更された行のみを処理するように ADF に指示します。
増分日付列: 増分抽出機能を使う場合は、ソーステーブルのウォーターマークとして使う日時列を選ぶ必要があります。
最初から読み取りを開始する: 増分抽出でこのオプションを設定すると、増分抽出が有効になっているパイプラインの最初の実行時にすべての行を読み取るよう ADF に指示します。

Azure Database for PostgreSQL シンクスクリプトの例

シンクの種類として Azure Database for PostgreSQL を使用する場合、関連付けられているデータフロースクリプトは次のようになります。

IncomingStream sink(allowSchemaDrift: true,
    validateSchema: false,
    deletable:false,
    insertable:true,
    updateable:true,
    upsertable:true,
    keys:['keyColumn'],
    format: 'table',
    skipDuplicateMapInputs: true,
    skipDuplicateMapOutputs: true) ~> AzurePostgreSQLSink

Lookup アクティビティのプロパティ

プロパティの詳細については、ルックアップアクティビティに関する記事を参照してください。

コピーアクティビティによってソース、シンクとしてサポートされるデータストアの一覧については、サポートされるデータストアに関するセクションを参照してください。