# Shopify Streaming Import連携

Shopifyは、あらゆる規模のビジネスのオンラインストアを支える主要なeコマースプラットフォームです。このStreaming Ingress Connectorを使用すると、ShopifyストアのWebhookイベントをリアルタイムでTreasure AIに取り込めます。顧客の購入、カート操作、商品の変更に基づいて、すぐに分析、トリガー、パーソナライゼーションを実行できます。

## サポートされるWebhookイベント 

この連携は、Shopifyから送信されるWebhookイベントを受信します。イベントは以下のカテゴリに分類されます。

| トピック | カテゴリ | 説明 |
|  --- | --- | --- |
| `checkouts/create` | 収益とリカバリー | 新しいチェックアウトが開始されたとき |
| `checkouts/update` | 収益とリカバリー | 既存のチェックアウトが変更されたとき |
| `orders/create` | 収益とリカバリー | 新しい注文が行われたとき |
| `orders/paid` | 収益とリカバリー | 注文の支払いが完了したとき |
| `orders/updated` | 収益とリカバリー | 既存の注文が変更されたとき |
| `orders/fulfilled` | リテンションとライフサイクル | 注文の発送が完了したとき |
| `orders/cancelled` | リテンションとライフサイクル | 注文がキャンセルされたとき |
| `refunds/create` | リテンションとライフサイクル | 返金が発行されたとき |
| `customers/create` | リテンションとライフサイクル | 新しい顧客アカウントが作成されたとき |
| `customers/update` | リテンションとライフサイクル | 顧客情報が更新されたとき |
| `customers/enable` | リテンションとライフサイクル | 顧客アカウントが有効化されたとき |
| `customers/disable` | リテンションとライフサイクル | 顧客アカウントが無効化されたとき |
| `carts/create` | エンゲージメントとオペレーション | 新しいカートが作成されたとき |
| `carts/update` | エンゲージメントとオペレーション | カートが変更されたとき |
| `inventory_levels/update` | エンゲージメントとオペレーション | 在庫レベルが変化したとき |
| `products/update` | エンゲージメントとオペレーション | 商品情報が変更されたとき |
| `collections/update` | エンゲージメントとオペレーション | コレクションが更新されたとき |
| `draft_orders/create` | エンゲージメントとオペレーション | 新しい下書き注文が作成されたとき |


Webhookイベントの完全なリストとデータ構造については、[Shopify Webhooksドキュメント](https://shopify.dev/docs/api/webhooks)を参照してください。

## 前提条件

- Treasure AIの基礎知識
- Shopifyの基礎知識
- Webhookを設定する権限を持つShopifyストア
- Treasure AIのAPIキー(書き込み専用)


## 要件と制限事項

- データを取り込むソースを作成する前に、Plazmaでデータベースとテーブルを作成する必要があります
- HMAC-SHA256署名の検証には、ShopifyのWebhookクライアントシークレットが必要です
- ペイロードの最大サイズは、ShopifyのWebhookペイロード制限に従います
- サポートされていないトピックのWebhookイベントは、通知なしで除外されます
- 重複したWebhook配信は、24時間の期間内で自動的に重複排除されます


## Treasure Data Integration の静的 IP アドレス

セキュリティポリシーで IP ホワイトリストが必要な場合は、接続を成功させるために Treasure Data の IP アドレスを許可リストに追加する必要があります。

リージョンごとに整理された静的 IP アドレスの完全なリストは、次のリンクにあります:
[IP Addresses for Integrations](/apis/endpoints/ip-addresses-integrations-result-workers)

## Treasure AI ConsoleからShopifyをインポート

### 認証の作成

1. Treasure AI Consoleを開きます。
2. **Integrations Hub**を選択します。
3. **Catalog**を選択します。
4. CatalogでShopify Streamingコネクタを検索し、アイコンにマウスカーソルを合わせて、**Create Authentication**を選択します。
5. **Credentials**タブが選択されていることを確認し、連携の認証情報を入力します。


![Integrations Hub CatalogのShopify streaming importコネクタのタイル](/assets/shopify-import-streaming-logo.eb780440a87030a28351404ab63fad180629f0a2697189896539a827d76d1553.cdfed7e3.png)

### 新しい認証フィールド

![Shopify streaming importの新しい認証の設定画面](/assets/shopify-import-streaming-new-authentication.5fe3dfdbd56557994584b146bd6833b66865e3435ae055eee0fa3e448b003a81.cdfed7e3.png)

| パラメータ | 説明 |
|  --- | --- |
| Treasure API Key | Treasure AIのAPIキー(書き込み専用) |
| Client Secret | ShopifyのWebhookクライアントシークレット。受信したWebhookの署名をHMAC-SHA256で検証するために使用します。 |


### ソースの作成

1. Treasure AI Consoleを開きます。
2. **Integrations Hub** > **Authentications**に移動します。
3. 新しく作成した認証を見つけて、**New Source**を選択します。


### 接続の作成

| パラメータ | 説明 |
|  --- | --- |
| Data Transfer Name | 転送の名前を定義できます。 |
| Authentication | 転送に使用される認証名。 |


1. **Data Transfer Name**フィールドにソース名を入力します。
2. **Next**を選択します。
3. Create Sourceページが、Source Tableタブが選択された状態で表示されます。


### ソーステーブルの指定

続行する前に、Plazmaでデータベースとテーブルを作成したことを確認してください。この連携では、Shopifyのデータを保存するための既存のデータベースとテーブルが必要です。

![New SourceボタンがあるAuthenticationsページ](/assets/shopify-import-streaming-authentications-new-source.1472929f7fe25ef09925fee9f39d40efe39346d7d2b4c4d8f9905a06bc5ecf6d.cdfed7e3.png)

1. Shopifyのデータを保存する宛先テーブルを設定し、**Next**を選択します。


| パラメータ | 説明 |
|  --- | --- |
| Datastore | Plazmaが使用可能なオプションです。 |
| Tags | オプション。タグを使用してこのソースを検索できます。 |
| Database | データをインポートするTreasure AI内のデータベースを指定します。 |
| Table | イベントデータを配置するデータベース内のテーブルを指定します。 |


### ソースIDのコピー

ソースID(UUID v4)は、ソース作成時に発行されます。

不正使用を防ぐため、**ソースID**は権限のない者に開示しないでください。

ソース作成後、自動的にソース一覧ページに移動します。

1. 作成したソースを検索します。
2. 同じ行の「...」をクリックし、**Copy Unique ID**を選択します。


このユニークIDが、ShopifyでWebhook URLを設定する際に必要なソースIDです。

### ShopifyでWebhook URLを設定

Treasure AIでShopifyのイベントを受信するには、Shopify管理画面でWebhookを設定する必要があります。

1. [Shopify Admin](https://admin.shopify.com/)にログインします。
2. **Settings** > **Notifications** > **Webhooks**に移動します。
3. **Create webhook**を選択します。
4. 購読するイベントトピックを選択します(例: `Order creation`)。
5. **Format**を**JSON**に設定します。
6. 使用しているTreasure AIリージョンのWebhook URLを入力します。`{source_id}`を前の手順でコピーしたソースIDに置き換えます。


US
```
https://shopify-in-streaming.treasuredata.com/v1/task/{source_id}
```

Tokyo
```
https://shopify-in-streaming.treasuredata.co.jp/v1/task/{source_id}
```

EU01
```
https://shopify-in-streaming.eu01.treasuredata.com/v1/task/{source_id}
```

AP02
```
https://shopify-in-streaming.ap02.treasuredata.com/v1/task/{source_id}
```

1. 最新の**Webhook API version**を選択します。
2. **Save**を選択します。


受信したいイベントトピックごとに、個別のWebhookサブスクリプションを作成する必要があります。[サポートされるWebhookイベント](#supported-webhook-events)セクションに記載されたトピックのうち、追跡したいトピックごとに上記の手順を繰り返してください。

ShopifyのWebhooks設定ページに表示される**Client Secret**は、Treasure AIで認証を設定したときに入力したClient Secretと一致する必要があります。このシークレットは、受信したWebhookが本当にShopifyから送信されたものであることを検証するために使用されます。

### Shopifyのリクエストヘッダー

Shopifyは、各Webhook配信で以下のHTTPヘッダーを送信します。コネクタは、これらのヘッダーを認証、重複排除、メタデータに使用します。

| ヘッダー | 用途 |
|  --- | --- |
| X-Shopify-Hmac-Sha256 | Webhookペイロードを検証するための、Base64エンコードされたHMAC-SHA256署名 |
| X-Shopify-Topic | Webhookイベントのトピック(例: `orders/create`) |
| X-Shopify-Shop-Domain | イベントを発生させたShopifyストアのドメイン |
| X-Shopify-Webhook-Id | 重複排除に使用される一意の配信ID |
| X-Shopify-API-Version | Webhookペイロードに使用されたShopify APIのバージョン |
| X-Shopify-Triggered-At | イベントが発生した日時のISO 8601タイムスタンプ |
| X-Shopify-Event-Id | イベントの一意の識別子 |


### データの取り込み

Webhookの設定とソースの作成が完了した後、最初のイベントが発生するとすぐに、コネクタはShopifyのイベントデータをTreasure AIに取り込み始めます。

### Webhookのレスポンスコード

コネクタは、Shopifyに以下のHTTPステータスコードを返します。

| ステータスコード | 意味 |
|  --- | --- |
| 200 OK | イベントを受信し、正常に処理しました。重複したWebhookや、サポートされていないトピック(通知なしで除外)の場合もこのコードを返します。 |
| 401 Unauthorized | HMAC-SHA256署名が無効か、存在しません。Client Secretが一致していることを確認してください。 |
| 403 Forbidden | ソースが見つからないか、ソースがアクティブな状態ではありません。 |
| 410 Gone | ソースが停止されています。 |
| 500 Internal Server Error | イベントの処理中に予期しないエラーが発生しました。 |


Shopifyは、2xx以外のステータスコードを返したWebhook配信を自動的に再試行します。401エラーが繰り返し発生する場合は、Treasure AIで設定したClient Secretが、ShopifyのWebhook設定に表示されているものと一致していることを確認してください。

## データスキーマ

### データのフラット化

コネクタは、ShopifyのWebhookから届くネストされたJSONペイロードを自動的にフラット化します。ネストされたオブジェクトはアンダースコアで区切ったキーでフラット化され、配列はJSON文字列としてシリアライズされます。

| JSONのデータ型 | 保存されるデータ型 | 処理 |
|  --- | --- | --- |
| string | string | そのまま保持 |
| number | long | そのまま保持 |
| boolean | string | そのまま保持 |
| object | フラット化されたフィールド | キーを`_`で連結(例: `customer.id`は`customer_id`になります) |
| array | string (JSON) | JSON文字列としてシリアライズ |
| null | — | 無視 |


### カラムの命名規則

- 小文字とアンダースコアのみ
- トップレベルのスカラーフィールド: `field_name`
- ネストされたフィールド: `parent_child`(例: `shipping_address_country`)


### イベントのメタデータフィールド

各イベントレコードには、フラット化されたWebhookペイロードに加えて、以下のメタデータフィールドが含まれます。

| カラム名 | データ型 | 説明 |
|  --- | --- | --- |
| raw_data | string | 元のJSON Webhookペイロード全体 |
| topic | string | Webhookイベントのトピック(例: `orders/create`、`customers/update`) |
| shop_domain | string | Webhookを送信したShopifyストアのドメイン |
| received_time | long | Treasure AIがイベントを受信した日時のUnixタイムスタンプ(秒) |
| webhook_id | string | Shopifyから送られる一意のWebhook配信ID。重複排除に使用されます |
| api_version | string | Webhookに使用されたShopify APIのバージョン(例: `2026-01`) |
| triggered_at | string | Shopifyでイベントが発生した日時のISO 8601タイムスタンプ |
| event_id | string | イベントの一意の識別子 |
| headers | string | リクエストヘッダー(文字列化したもの) |
| time | long | レコードがデータベースに書き込まれた日時のUnixタイムスタンプ。このフィールドはTreasure AIプラットフォームによって自動的に追加されます。 |


### イベントデータの例

**注文作成イベントの例:**

```json
{
  "topic": "orders/create",
  "shop_domain": "example-store.myshopify.com",
  "api_version": "2026-01",
  "webhook_id": "b1234567-89ab-cdef-0123-456789abcdef",
  "triggered_at": "2026-01-15T10:30:00.000Z",
  "received_time": 1768473000,
  "id": 5678901234,
  "email": "customer@example.com",
  "created_at": "2026-01-15T10:30:00-05:00",
  "updated_at": "2026-01-15T10:30:00-05:00",
  "total_price": "59.99",
  "currency": "USD",
  "financial_status": "paid",
  "fulfillment_status": null,
  "customer_id": 1234567890,
  "customer_email": "customer@example.com",
  "customer_first_name": "Jane",
  "customer_last_name": "Doe",
  "shipping_address_city": "New York",
  "shipping_address_country": "United States",
  "shipping_address_province": "New York",
  "line_items": "[{\"id\":9876543210,\"title\":\"Example Product\",\"quantity\":1,\"price\":\"59.99\"}]",
  "time": 1768473005
}
```

**顧客作成イベントの例:**

```json
{
  "topic": "customers/create",
  "shop_domain": "example-store.myshopify.com",
  "api_version": "2026-01",
  "webhook_id": "c2345678-90ab-cdef-1234-567890abcdef",
  "triggered_at": "2026-01-15T09:15:00.000Z",
  "received_time": 1768468500,
  "id": 1234567890,
  "email": "newcustomer@example.com",
  "first_name": "John",
  "last_name": "Smith",
  "phone": "+1-555-0100",
  "created_at": "2026-01-15T09:15:00-05:00",
  "updated_at": "2026-01-15T09:15:00-05:00",
  "orders_count": 0,
  "state": "enabled",
  "total_spent": "0.00",
  "default_address_city": "San Francisco",
  "default_address_province": "California",
  "default_address_country": "United States",
  "default_address_zip": "94105",
  "time": 1768468505
}
```

`line_items`、`discount_codes`、`tax_lines`などの配列カラムには、シリアライズされたJSONデータが含まれます。クエリでTreasure AIのJSON関数を使用してこれらを解析し、特定のフィールドを抽出できます。

## データのクエリ

取り込んだデータは、SQLクエリで参照できます。例:

```sql
SELECT
  id AS order_id,
  topic,
  email,
  total_price,
  currency,
  financial_status,
  customer_first_name,
  customer_last_name,
  shipping_address_city,
  shipping_address_country,
  line_items,
  time
FROM
  your_database.your_table
WHERE
  topic = 'orders/create'
  AND TD_INTERVAL(time, '-1d', 'JST')
```