Azure Data Factory および Azure Synapse Analytics でのルックアップ アクティビティ

適用対象: Azure Data Factory Azure Synapse Analytics

ルックアップ アクティビティは、Azure Data Factory および Azure Synapse Analytics パイプラインでサポートされている任意のデータ ソースからデータセットを取得できます。 それを使用して、オブジェクト名をハード コーディングする代わりに、後続のアクティビティで操作するオブジェクトを動的に決定できます。 オブジェクトの例として、ファイルとテーブルがあります。

ルックアップ アクティビティは、構成ファイルやテーブルの内容を読み取って返します。 また、クエリまたはストアド プロシージャの実行結果を返します。 出力には、シングルトン値または属性の配列を指定できます。これは、ForEach アクティビティなどの後続のコピー、変換、または制御フロー アクティビティで使用できます。

UI を使用してルックアップ アクティビティを作成する

パイプライン内でルックアップ アクティビティを使用するには、次の手順を実行します。

  1. パイプラインの [アクティビティ] ペイン内でルックアップを検索し、ルックアップ アクティビティをパイプライン キャンバスにドラッグします。

  2. キャンバス上で新しいルックアップ アクティビティ (まだ選択されていない場合)、その [設定] タブの順に選択して、その詳細を編集します。

    ルックアップ アクティビティの UI を示しています。

  3. 既存のソース データセットを選択するか、[新規] ボタンを選択して新しいデータセットを作成します。

  4. ソース データセットから含める行を識別するためのオプションは、データセットの種類によって異なります。 上の例は、区切りテキスト データセットの構成オプションを示しています。 Azure SQL テーブル データセットと OData データセットの構成オプションの例を次に示します。

    Azure SQL テーブル データセットのルックアップ アクティビティの構成オプションを示しています。

    OData データセットのルックアップ アクティビティの構成オプションを示しています。

サポートされる機能

次のことを考慮してください。

  • ルックアップ アクティビティでは、最大 5,000 行を返すことができます。結果セットにそれを超えるレコードが含まれている場合は、最初の 5,000 行が返されます。
  • ルックアップ アクティビティの出力でサポートされる最大サイズは 4 MB で、サイズがこの制限を超えるとアクティビティは失敗します。
  • タイムアウト前のルックアップ アクティビティの最長期間は 24 時間です。

注意

クエリまたはストアド プロシージャを使用してデータを検索する場合は、1 つの結果セットのみが返されるようにしてください。 そうしないと、ルックアップ アクティビティは失敗します。

次のデータ ソースがルックアップ アクティビティでサポートされています。

カテゴリ データ ストア
Azure Azure BLOB Storage
  NoSQL 用 Azure Cosmos DB
  Azure Data Explorer (Kusto)
  Azure Data Lake Storage Gen1
  Azure Data Lake Storage Gen2
  Azure Database for MariaDB
  Azure Database for MySQL
  Azure Database for PostgreSQL
  Azure Databricks Delta Lake
  Azure Files
  Azure SQL Database
  Azure SQL Database マネージド インスタンス
  Azure Synapse Analytics
  Azure Table Storage
[データベース] Amazon RDS for Oracle
  Amazon RDS for SQL Server
  Amazon Redshift
  Apache Impala
  Azure SQL Database マネージド インスタンス
  DB2
  Drill
  Google AdWords
  Google BigQuery
  Greenplum
  HBase
  Hive
  Informix
  MariaDB
  Microsoft Access
  MySQL
  Netezza
  Oracle
  Phoenix
  PostgreSQL
  Presto
  SAP Business Warehouse Open Hub
  MDX を介した SAP Business Warehouse
  SAP HANA
  SAP テーブル
  SQL Server
  Spark
  Sybase
  Teradata
  Vertica
NoSQL Cassandra
  Couchbase (プレビュー)
[最近使ったファイル] Amazon S3
  Amazon S3 互換ストレージ
  ファイル システム
  FTP
  Google Cloud Storage
  HDFS
  汎用 HTTP
  Oracle Cloud Storage
  SFTP
汎用プロトコル 汎用 OData
  汎用 ODBC
  SharePoint Online リスト
サービスとアプリ Amazon Marketplace Web Service
  Concur (プレビュー)
  Dataverse
  Dynamics 365
  Dynamics AX
  Dynamics CRM
  HubSpot
  Jira
  Azure Data Explorer (Kusto)
  Magento (プレビュー)
  Marketo (プレビュー)
  Oracle Eloqua (プレビュー)
  Oracle Responsys (プレビュー)
  Oracle Service Cloud (プレビュー)
  Paypal (プレビュー)
  QuickBooks (プレビュー)
  SAP Cloud for Customer (C4C)
  SAP ECC
  Salesforce
  Salesforce Marketing Cloud
  Salesforce Service Cloud
  ServiceNow
  Shopify (プレビュー)
  Snowflake
  Square (プレビュー)
  Web テーブル (HTML テーブル)
  Xero
  Zoho (プレビュー)

Note

"プレビュー" と記載されたコネクタは試用版です。フィードバックをお寄せください。 ソリューションでプレビュー版コネクタの依存関係を取得したい場合、Azure サポートにお問い合わせください。

構文

{
    "name":"LookupActivity",
    "type":"Lookup",
    "typeProperties":{
        "source":{
            "type":"<source type>"
        },
        "dataset":{
            "referenceName":"<source dataset name>",
            "type":"DatasetReference"
        },
        "firstRowOnly":<true or false>
    }
}

型のプロパティ

名前 説明 Type 必須
dataset ルックアップ用のデータセット参照を提供します。 対応する各コネクタの記事の 「データセットのプロパティ」セクションから詳細を取得します。 キーと値のペア はい
source データセット固有のソース プロパティを含みます (コピー アクティビティ ソースと同じ)。 対応する各コネクタの記事の「コピー アクティビティのプロパティ」セクションから詳細を取得します。 キーと値のペア はい
firstRowOnly 最初の行のみまたはすべての行のどちらを返すかを示します。 Boolean いいえ。 既定では、 trueです。

Note

  • ByteArray 型のソース列はサポートされていません。
  • データセット定義内の構造体はサポートされていません。 テキスト形式のファイルの場合は、ヘッダー行を使用して列名を指定できます。
  • ルックアップ ソースが JSON ファイルの場合、JSON オブジェクトを整形するための jsonPathDefinition 設定はサポートされていません。 オブジェクト全体が取得されます。

ルックアップ アクティビティの結果の使用

ルックアップ結果は、アクティビティ実行結果の output セクションに返されます。

  • firstRowOnlytrue (既定値) に設定されているときは、出力形式は次のコードに示すとおりです。 ルックアップ結果は固定の firstRow キーの下にあります。 後続のアクティビティで結果を使用するには、パターン @{activity('LookupActivity').output.firstRow.table} を使用します。

    {
        "firstRow":
        {
            "Id": "1",
            "schema":"dbo",
            "table":"Table1"
        }
    }
    
  • firstRowOnlyfalse に設定されているときは、出力形式は次のコードに示すとおりです。 count フィールドは、返されたレコードの数を示します。 固定された value 配列の下に詳細な値が表示されます。 このような場合は、ルックアップ アクティビティの後ろに Foreach アクティビティが続きます。 value 配列は、パターン @activity('MyLookupActivity').output.value を使用して ForEach アクティビティの items フィールドに渡します。 value 配列の要素にアクセスするには、構文 @{activity('lookupActivity').output.value[zero based index].propertyname} を使用します。 たとえば @{activity('lookupActivity').output.value[0].schema} です。

    {
        "count": "2",
        "value": [
            {
                "Id": "1",
                "schema":"dbo",
                "table":"Table1"
            },
            {
                "Id": "2",
                "schema":"dbo",
                "table":"Table2"
            }
        ]
    } 
    

この例では、パイプラインに 2 つのアクティビティ、ルックアップコピーが含まれています。 コピー アクティビティは、お使いの Azure SQL Database インスタンスの SQL テーブルから Azure Blob Storage にデータをコピーします。 SQL テーブルの名前は、Blob Storage 内の JSON ファイルに格納されます。 ルックアップ アクティビティは、実行時にテーブル名を検索します。 JSON は、この方法を使用して動的に変更されます。 パイプラインやデータセットを再デプロイする必要はありません。

この例では、最初の行のみのルックアップを示します。 すべての行のルックアップについて、および ForEach アクティビティで結果をチェーンするには、複数のテーブルの一括コピーに関する記事のサンプルを参照してください。

パイプライン

  • ルックアップ アクティビティは、Azure Blob Storage 内の場所を表す LookupDataset を使用するように設定されています。 ルックアップ アクティビティは、この場所にある JSON ファイルから SQL テーブルの名前を読み取ります。
  • コピー アクティビティは、SQL テーブルの名前であるルックアップ アクティビティの出力を使用します。 SourceDataset 内の tableName プロパティは、ルックアップ アクティビティからの出力を使用するように設定されています。 コピー アクティビティは、SQL テーブルから Azure Blob Storage 内の場所にデータをコピーします。 場所は SinkDataset プロパティによって指定されます。
{
    "name": "LookupPipelineDemo",
    "properties": {
        "activities": [
            {
                "name": "LookupActivity",
                "type": "Lookup",
                "dependsOn": [],
                "policy": {
                    "timeout": "7.00:00:00",
                    "retry": 0,
                    "retryIntervalInSeconds": 30,
                    "secureOutput": false,
                    "secureInput": false
                },
                "userProperties": [],
                "typeProperties": {
                    "source": {
                        "type": "JsonSource",
                        "storeSettings": {
                            "type": "AzureBlobStorageReadSettings",
                            "recursive": true
                        },
                        "formatSettings": {
                            "type": "JsonReadSettings"
                        }
                    },
                    "dataset": {
                        "referenceName": "LookupDataset",
                        "type": "DatasetReference"
                    },
                    "firstRowOnly": true
                }
            },
            {
                "name": "CopyActivity",
                "type": "Copy",
                "dependsOn": [
                    {
                        "activity": "LookupActivity",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    }
                ],
                "policy": {
                    "timeout": "7.00:00:00",
                    "retry": 0,
                    "retryIntervalInSeconds": 30,
                    "secureOutput": false,
                    "secureInput": false
                },
                "userProperties": [],
                "typeProperties": {
                    "source": {
                        "type": "AzureSqlSource",
                        "sqlReaderQuery": {
                            "value": "select * from [@{activity('LookupActivity').output.firstRow.schema}].[@{activity('LookupActivity').output.firstRow.table}]",
                            "type": "Expression"
                        },
                        "queryTimeout": "02:00:00",
                        "partitionOption": "None"
                    },
                    "sink": {
                        "type": "DelimitedTextSink",
                        "storeSettings": {
                            "type": "AzureBlobStorageWriteSettings"
                        },
                        "formatSettings": {
                            "type": "DelimitedTextWriteSettings",
                            "quoteAllText": true,
                            "fileExtension": ".txt"
                        }
                    },
                    "enableStaging": false,
                    "translator": {
                        "type": "TabularTranslator",
                        "typeConversion": true,
                        "typeConversionSettings": {
                            "allowDataTruncation": true,
                            "treatBooleanAsNumber": false
                        }
                    }
                },
                "inputs": [
                    {
                        "referenceName": "SourceDataset",
                        "type": "DatasetReference",
                        "parameters": {
                            "schemaName": {
                                "value": "@activity('LookupActivity').output.firstRow.schema",
                                "type": "Expression"
                            },
                            "tableName": {
                                "value": "@activity('LookupActivity').output.firstRow.table",
                                "type": "Expression"
                            }
                        }
                    }
                ],
                "outputs": [
                    {
                        "referenceName": "SinkDataset",
                        "type": "DatasetReference",
                        "parameters": {
                            "schema": {
                                "value": "@activity('LookupActivity').output.firstRow.schema",
                                "type": "Expression"
                            },
                            "table": {
                                "value": "@activity('LookupActivity').output.firstRow.table",
                                "type": "Expression"
                            }
                        }
                    }
                ]
            }
        ],
        "annotations": [],
        "lastPublishTime": "2020-08-17T10:48:25Z"
    }
}

ルックアップ データセット

ルックアップ データセットは、AzureBlobStorageLinkedService 型で指定された Azure Storage ルックアップ フォルダー内の sourcetable.json ファイルです。

{
    "name": "LookupDataset",
    "properties": {
        "linkedServiceName": {
            "referenceName": "AzureBlobStorageLinkedService",
            "type": "LinkedServiceReference"
        },
        "annotations": [],
        "type": "Json",
        "typeProperties": {
            "location": {
                "type": "AzureBlobStorageLocation",
                "fileName": "sourcetable.json",
                "container": "lookup"
            }
        }
    }
}

コピー アクティビティのソース データセット

ソース データセットは、SQL テーブルの名前であるルックアップ アクティビティの出力を使用します。 コピー アクティビティは、SQL テーブルから Azure Blob Storage 内の場所にデータをコピーします。 場所は sink データセットによって指定されます。

{
    "name": "SourceDataset",
    "properties": {
        "linkedServiceName": {
            "referenceName": "AzureSqlDatabase",
            "type": "LinkedServiceReference"
        },
        "parameters": {
            "schemaName": {
                "type": "string"
            },
            "tableName": {
                "type": "string"
            }
        },
        "annotations": [],
        "type": "AzureSqlTable",
        "schema": [],
        "typeProperties": {
            "schema": {
                "value": "@dataset().schemaName",
                "type": "Expression"
            },
            "table": {
                "value": "@dataset().tableName",
                "type": "Expression"
            }
        }
    }
}

コピー アクティビティのシンク データセット

コピー アクティビティは、SQL テーブルから、Azure Storage の csv フォルダー内の filebylookup.csv ファイルにデータをコピーします。 このファイルは、AzureBlobStorageLinkedService プロパティで指定されています。

{
    "name": "SinkDataset",
    "properties": {
        "linkedServiceName": {
            "referenceName": "AzureBlobStorageLinkedService",
            "type": "LinkedServiceReference"
        },
        "parameters": {
            "schema": {
                "type": "string"
            },
            "table": {
                "type": "string"
            }
        },
        "annotations": [],
        "type": "DelimitedText",
        "typeProperties": {
            "location": {
                "type": "AzureBlobStorageLocation",
                "fileName": {
                    "value": "@{dataset().schema}_@{dataset().table}.csv",
                    "type": "Expression"
                },
                "container": "csv"
            },
            "columnDelimiter": ",",
            "escapeChar": "\\",
            "quoteChar": "\""
        },
        "schema": []
    }
}

sourcetable.json

sourcetable.json ファイルには、次の 2 種類の形式を使用できます。

オブジェクトのセット

{
   "Id":"1",
   "schema":"dbo",
   "table":"Table1"
}
{
   "Id":"2",
   "schema":"dbo",
   "table":"Table2"
}

オブジェクトの配列

[ 
    {
        "Id": "1",
        "schema":"dbo",
        "table":"Table1"
    },
    {
        "Id": "2",
        "schema":"dbo",
        "table":"Table2"
    }
]

制限事項と回避策

Lookup アクティビティと提案される回避策のいくつかの制限を次に示します。

制限事項 回避策
ルックアップ アクティビティの最大行数は 5,000 行で、最大サイズは 4 MB です。 最大行数またはサイズを超えないデータを取得する、内側パイプライン上で外側パイプラインが反復される 2 段のパイプラインを設計します。

次のステップ

Azure Data Factory パイプラインおよび Azure Synapse Analytics パイプラインでサポートされているその他の制御フロー アクティビティを参照してください。