Azure Data Factory を使用して Azure Cognitive Search インデックスにデータをプッシュする

Note

この記事は、Data Factory のバージョン 1 に適用されます。 現在のバージョンの Data Factory サービスを使用している場合は、V2 の Azure Cognitive Search コネクタに関するページを参照してください。

この記事では、コピー アクティビティを使用して、サポートされているソース データ ストアから Azure Cognitive Search インデックスにデータをプッシュする方法について説明します。 サポートされているソース データ ストアについては、サポートされているソースとシンクの表のソースの列を参照してください。 この記事は、「 データ移動アクティビティ 」という記事に基づき、コピー アクティビティによるデータ移動の一般概要とサポートされるデータ ストアの組み合わせについて紹介しています。

接続を有効にする

Data Factory サービスからオンプレミスのデータ ストアに接続できるようにするには、オンプレミスの環境に Data Management Gateway をインストールします。 データ ストアをホストするコンピューターと同じコンピューター、またはデータ ストアとのリソースの競合を避けるために別のコンピューター上にゲートウェイをインストールできます。

Data Management Gateway では、安全かつ管理された方法でオンプレミスのデータがクラウド サービスに接続されます。 Data Management Gateway の詳細については、 オンプレミスとクラウド間でのデータ移動 に関する記事を参照してください。

作業の開始

さまざまなツールや API を使用して、ソース データ ストアから検索インデックスにデータをプッシュするコピー アクティビティを含むパイプラインを作成できます。

パイプラインを作成する最も簡単な方法は、コピー ウィザードを使うことです。 「チュートリアル:コピー ウィザードを使用してパイプラインを作成する」を参照してください。データのコピー ウィザードを使用してパイプラインを作成する簡単なチュートリアルです。

また、次のツールを使用してパイプラインを作成することもできます。Visual StudioAzure PowerShellAzure Resource Manager テンプレート.NET APIREST API。 コピー アクティビティを含むパイプラインを作成するための詳細な手順については、コピー アクティビティのチュートリアルをご覧ください。

ツールと API のいずれを使用する場合も、次の手順を実行して、ソース データ ストアからシンク データ ストアにデータを移動するパイプラインを作成します。

  1. リンクされたサービスを作成し、入力データ ストアと出力データ ストアをデータ ファクトリにリンクします。
  2. コピー操作用の入力データと出力データを表すデータセットを作成します。
  3. 入力としてのデータセットと出力としてのデータセットを受け取るコピー アクティビティを含むパイプラインを作成します。

ウィザードを使用すると、Data Factory エンティティ (リンクされたサービス、データセット、パイプライン) に関する JSON の定義が自動的に作成されます。 (.NET API を除く) ツールまたは API を使う場合は、JSON 形式でこれらの Data Factory エンティティを定義します。 検索インデックスにデータをコピーするために使用される Data Factory エンティティに関する JSON 定義のサンプルについては、この記事の「JSON 例: SQL Server から Azure Cognitive Search インデックスにデータをコピーする」のセクションを参照してください。

以降のセクションでは、検索インデックスに固有の Data Factory エンティティを定義するために使用される JSON プロパティに関する詳細について説明します。

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

次の表では、Azure Cognitive Search のリンクされたサービスに固有の JSON 要素について説明しています。

プロパティ 説明 必須
type type プロパティは、次のように設定する必要があります:AzureSearch はい
url 検索サービスの URL。 はい
key 検索サービスの管理者キー。 はい

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

データセットの定義に使用できるセクションとプロパティの完全な一覧については、 データセットの作成 に関する記事をご覧ください。 データセット JSON の構造、可用性、ポリシーなどのセクションは、データセットのすべての型でほぼ同じです。 typeProperties セクションは、データセットの型ごとに異なります。 AzureSearchIndex 型のデータセットの typeProperties セクションには次のプロパティがあります。

プロパティ 説明 必須
type type プロパティを AzureSearchIndex に設定する必要があります。 はい
indexName 検索インデックスの名前。 Data Factory では、インデックスは作成されません。 このインデックスは Azure Cognitive Search に存在する必要があります。 はい

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

アクティビティの定義に使用できるセクションとプロパティの完全な一覧については、 パイプラインの作成 に関する記事をご覧ください。 名前、説明、入力テーブル、出力テーブル、さまざまなポリシーなどのプロパティは、あらゆる種類のアクティビティで使用できます。 一方、typeProperties セクションで使用できるプロパティは各アクティビティの種類によって異なります。 コピー アクティビティの場合、ソースとシンクの種類によって異なります。

コピー アクティビティで、シンクの種類が AzureSearchIndexSink である場合は、typeProperties セクションで次のプロパティを使用できます。

プロパティ 説明 使用できる値 必須
WriteBehavior ドキュメントがそのインデックスに既に存在する場合に、マージするか置換するかを指定します。 詳細については、「WriteBehavior プロパティ」を参照してください。 マージ (既定値)
アップロード
いいえ
WriteBatchSize バッファー サイズが writeBatchSize に達すると、検索インデックスにデータをアップロードします。 詳細については、「WriteBatchSize プロパティ」を参照してください。 1 ~ 1,000。 既定値は 1,000 です。 いいえ

WriteBehavior プロパティ

データを書き込むときに AzureSearchSink で upsert されます。 つまり、ドキュメントを書き込むとき、そのドキュメントのキーが検索インデックスに既に存在する場合、Azure Cognitive Search は競合の例外をスローするのではなく、既存のドキュメントを更新します。

AzureSearchSink で提供される upsert 動作 (AzureSearch SDK の使用による) は、次の 2 とおりあります。

  • マージ: 新しいドキュメントのすべての列を既存の列と結合します。 新しいドキュメント内に null 値を持つ列がある場合は、既存の列の値が保持されます。
  • アップロード: 既存のドキュメントが新しいドキュメントで置き換えられます。 新しいドキュメントで指定されていない列の場合は、既存のドキュメントに null 以外の値があるかどうかに関係なく、値は null に設定されます。

既定の動作はマージです。

WriteBatchSize プロパティ

Azure Cognitive Search サービスは、バッチとしてのドキュメントの書き込みをサポートしています。 バッチには、1 ~ 1,000 のアクションを含めることができます。 1 つのアクションで、1 つのドキュメントのアップロード/マージ操作の実行を処理します。

データ型のサポート

次の表は、Azure Cognitive Search のデータ型がサポートされているかどうかを示しています。

Azure Cognitive Search のデータ型 Azure Cognitive Search のシンクでのサポート
String Y
Int32 Y
Int64 Y
Double Y
Boolean Y
DataTimeOffset Y
String Array N
GeographyPoint N

JSON の使用例:SQL Server から Azure Cognitive Search インデックスにデータをコピーする

次のサンプルは以下を示しています。

  1. AzureSearch 型のリンクされたサービス。
  2. OnPremisesSqlServer型のリンクされたサービス。
  3. SqlServerTable 型の入力データセット
  4. AzureSearchIndex 型の出力データセット
  5. SqlSourceAzureSearchIndexSink を使用するコピー アクティビティを含むパイプライン

このサンプルは、SQL Server データベースから検索インデックスに時系列データを 1 時間ごとにコピーします。 このサンプルで使用される JSON プロパティの説明は、サンプルに続くセクションにあります。

最初の手順として、オンプレミスのコンピューターでデータ管理ゲートウェイを設定します。 設定手順は、 オンプレミスの場所とクラウドの間でのデータ移動 に関する記事に記載されています。

Azure Cognitive Search のリンクされたサービス:

{
    "name": "AzureSearchLinkedService",
    "properties": {
        "type": "AzureSearch",
        "typeProperties": {
            "url": "https://<service>.search.windows.net",
            "key": "<AdminKey>"
        }
    }
}

SQL Server のリンクされているサービス

{
  "Name": "SqlServerLinkedService",
  "properties": {
    "type": "OnPremisesSqlServer",
    "typeProperties": {
      "connectionString": "Data Source=<servername>;Initial Catalog=<databasename>;Integrated Security=False;User ID=<username>;Password=<password>;",
      "gatewayName": "<gatewayname>"
    }
  }
}

SQL Server 入力データセット

このサンプルでは、SQL Server で「MyTable」という名前のテーブルを作成し、時系列データ用に「timestampcolumn」という名前の列が含まれているものと想定しています。 1 つのデータセットを使用して同じデータベース内の複数のテーブルに対してクエリを実行することはできますが、データセットの tableName typeProperty には 1 つのテーブルを使用する必要があります。

"external": "true" の設定により、このデータセットが Data Factory の外部にあり、Data Factory のアクティビティによって生成されたものではないことが Data Factory サービスに通知されます。

{
  "name": "SqlServerDataset",
  "properties": {
    "type": "SqlServerTable",
    "linkedServiceName": "SqlServerLinkedService",
    "typeProperties": {
      "tableName": "MyTable"
    },
    "external": true,
    "availability": {
      "frequency": "Hour",
      "interval": 1
    },
    "policy": {
      "externalData": {
        "retryInterval": "00:01:00",
        "retryTimeout": "00:10:00",
        "maximumRetry": 3
      }
    }
  }
}

Azure Cognitive Search の出力データセット:

このサンプルは、products という名前の Azure Cognitive Search インデックスにデータをコピーします。 Data Factory では、インデックスは作成されません。 サンプルをテストするには、この名前を持つインデックスを作成します。 入力データセットと同じ列数を持つ検索インデックスを作成します。 新しいエントリが検索インデックスに 1 時間ごとに追加されます。

{
    "name": "AzureSearchIndexDataset",
    "properties": {
        "type": "AzureSearchIndex",
        "linkedServiceName": "AzureSearchLinkedService",
        "typeProperties" : {
            "indexName": "products",
        },
        "availability": {
            "frequency": "Minute",
            "interval": 15
        }
    }
}

SQL ソースと Azure Cognitive Search インデックス シンクを使用したパイプラインでのコピー アクティビティ:

パイプラインには、入力データセットと出力データセットを使用するように構成され、1 時間おきに実行するようにスケジュールされているコピー アクティビティが含まれています。 パイプライン JSON 定義で、source 型が SqlSource に設定され、sink 型が AzureSearchIndexSink に設定されています。 SqlReaderQuery プロパティに指定されている SQL クエリは過去のデータを選択してコピーします。

{
  "name":"SamplePipeline",
  "properties":{
    "start":"2014-06-01T18:00:00",
    "end":"2014-06-01T19:00:00",
    "description":"pipeline for copy activity",
    "activities":[
      {
        "name": "SqlServertoAzureSearchIndex",
        "description": "copy activity",
        "type": "Copy",
        "inputs": [
          {
            "name": " SqlServerInput"
          }
        ],
        "outputs": [
          {
            "name": "AzureSearchIndexDataset"
          }
        ],
        "typeProperties": {
          "source": {
            "type": "SqlSource",
            "SqlReaderQuery": "$$Text.Format('select * from MyTable where timestampcolumn >= \\'{0:yyyy-MM-dd HH:mm}\\' AND timestampcolumn < \\'{1:yyyy-MM-dd HH:mm}\\'', WindowStart, WindowEnd)"
          },
          "sink": {
            "type": "AzureSearchIndexSink"
          }
        },
        "scheduler": {
          "frequency": "Hour",
          "interval": 1
        },
        "policy": {
          "concurrency": 1,
          "executionPriorityOrder": "OldestFirst",
          "retry": 0,
          "timeout": "01:00:00"
        }
      }
    ]
  }
}

クラウド データ ストアから Azure Cognitive Search にデータをコピーしている場合は、executionLocation プロパティが必要です。 たとえば、コピー アクティビティ typeProperties で必要となる変更は次の JSON スニペットのようになります。 「クラウド データ ストア間でのデータのコピー」セクションで、サポートされている値および詳細を確認してください。

"typeProperties": {
  "source": {
    "type": "BlobSource"
  },
  "sink": {
    "type": "AzureSearchIndexSink"
  },
  "executionLocation": "West US"
}

クラウド ソースからコピー

クラウド データ ストアから Azure Cognitive Search にデータをコピーしている場合は、executionLocation プロパティが必要です。 たとえば、コピー アクティビティ typeProperties で必要となる変更は次の JSON スニペットのようになります。 「クラウド データ ストア間でのデータのコピー」セクションで、サポートされている値および詳細を確認してください。

"typeProperties": {
  "source": {
    "type": "BlobSource"
  },
  "sink": {
    "type": "AzureSearchIndexSink"
  },
  "executionLocation": "West US"
}

コピー アクティビティ定義で、ソース データセットの列をシンク データセットの列にマップすることもできます。 詳細については、Azure Data Factory のデータセット列のマッピングに関するページを参照してください。

パフォーマンスとチューニング

データ移動 (コピー アクティビティ) のパフォーマンスに影響する主な要因と、パフォーマンスを最適化するための各種方法については、「コピー アクティビティのパフォーマンスとチューニングに関するガイド」をご覧ください。

次のステップ

次の記事をご覧ください。