---
schema_version: "1"
id: 0a240b8ef490aa274af559b4d3f076bcc96696f9742b6fde5f88fc9ddc555930
title: DataWeave でのストリーミング
collection: mulesoft
version: current
locale: ja-jp
portal_url: https://docs.salesforce.com/mulesoft/current/ja-jp/dataweave/2.4/dataweave-streaming.md
canonical_url: https://docs.mulesoft.com/jp/dataweave/2.4/dataweave-streaming
source_site: docs.mulesoft.com
content_type: documentation
content_hash: 3ea937a8abef6952aff0f4cb4ff6421beed12ed18695fc1e99534d723331c14c
description: MuleSoft Documentation Site
release: "2.4"
latest: false
---
# DataWeave でのストリーミング

DataWeave は、Mule アプリケーション内のフローを通してエンドツーエンドのストリーミングをサポートします。ストリーミングにより、大きなドキュメントでもメモリをオーバーフローさせることなく処理をスピードアップできます。

DataWeave は、ドキュメント全体をスキャンしてインデックス化するのではなく、ストリーミングされたデータのバイトが到着するたびに処理します。遅延モードでは、DataWeave はストリーミングされた出力データをディスクに保存せずに、メッセージプロセッサーに直接渡すこともできます。この動作によって DataWeave と Mule は、デフォルトのプロセスでデータを読み書きするよりも速くデータを処理し、リソースコンシューム量を少なく抑えることができます。

データを正常にストリーミングするためには、以下を理解することが重要です。

- ストリームの基本単位はデータ形式ごとに異なります。  
    CSV ドキュメントであればレコード、JSON ドキュメントであれば配列の要素、XML ドキュメントであればコレクションが基本単位となります。
- ストリーミングでは、各ストリーム単位に順次アクセスします。  
    ストリーミングでは、ドキュメントへのランダムアクセスはサポートされません。

## [](#ストリーミングの有効化)ストリーミングの有効化

ストリーミングはデフォルトでは有効化されていません。2 つの設定プロパティを使用して、サポートされているデータ形式でデータをストリーミングできます。

- ソースデータをストリームとして読み取るための ​`streaming`​ プロパティ
- 出力ストリームをフロー内の次のメッセージプロセッサーに直接渡すための ​`deferred`​ ライタープロパティ

DataWeave がソースデータをストリームとして読み取るには、データソースで ​`streaming`​ リーダープロパティを ​`true`​ に設定する必要があります。Mule アプリケーション内で、この設定を ​**MIME Type (MIME タイプ)**​ プロパティ ​`outputMimeType`​ または ​`mimeType`​ の値に追加します。HTTP Listener 操作、HTTP Request 操作、On New or Updated File 操作、Set Payload コンポーネントなど、データを生成する任意のコネクタ操作または Mule コンポーネントでプロパティを設定できます。DataWeave は、ストリーミングプロパティを設定したアプリケーション内のポイントから、DataWeave 式とスクリプトを含むすべてのダウンストリームコンポーネントとコネクタ操作を介してデータをストリーミングデータとして読み取ります。

```XML
<flow name="dw-streaming-example" >
  <http:listener doc:name="Listener"
      outputMimeType="application/json; streaming=true"
      config-ref="HTTP_Listener_config" path="/input"/>
</flow>
```

`streaming=true`​ は ​`outputMimeType`​ 値の一部ですので注意してください。File コンポーネントや FTP コンポーネントなどの他の多くの Mule コンポーネントも MIME タイプ設定をサポートします。

ストリーミングされた出力を次のメッセージプロセッサーに渡すには、DataWeave スクリプトで ​`output`​ ディレクティブを次のように使用します。

```XML
output application/json deferred=true
```

`deferred`​ プロパティの使い方の詳細は、​[出力のストリーミング](#stream_output)​ を参照してください。

## [](#stream_csv)CSV のストリーミング

CSV は、その構造により最もシンプルなストリーミング形式となります。CSV ヘッダーの下にある各行は、ストリーム可能なレコードです。次の CSV の例は、​`name`​、​`lastName`​、​`age`​ の各値を含むレコードから構成されています。 ​

```CSV
name,lastName,age
mariano,achaval,37
leandro,shokida,30
pedro,achaval,4
christian,chibana,25
sara,achaval,2
matias,achaval,8
```

この CSV の例をストリーミングするため、次のスクリプトは各レコードから値を選択しています。このスクリプトは、​`map`​ 関数を使用して、ドキュメント内の各レコードを反復処理しています。 ​

```DataWeave
payload map (record) ->
    {
        FullName: record.lastName ++ "," ++ record.name,
        Age: record.age
    }
```

ストリーミングでは、ドキュメント全体へのランダムアクセスはサポートされませんが、各レコードはメモリに読み込まれるため、DataWeave スクリプトは​_各レコード内の_​データにはランダムにアクセスできます。たとえば、​`record.lastName "," record.name,`​ という式により、入力での値の順序が逆であっても、​`name`​ 値にアクセスする前に ​`lastName`​ 値にアクセスできます。

ただし、ストリーミングは次のスクリプトでは機能しません。入力とは異なる順序で要素を返すためには、スクリプトでドキュメント全体へのランダムアクセスが必要になるためです。 ​

```DataWeave
[payload[-2], payload[-1], payload[3]]
```

​

## [](#stream_json)JSON のストリーミング

​JSON ストリーミングの単位は配列の各要素です。

DataWeave 2.2.0 を使用して Mule 4.2 で JSON をストリーミングする場合には、配列が入力のルートに存在する必要があります。Mule 4.3 で JSON をストリーミングする場合には、配列が入力のルートに存在しなくても構いません。

```JSON
{
    "name" : "Mariano",
    "lastName": "Achaval",
    "family": [
        {"name": "Sara", "age": 2},
        {"name": "Pedro", "age": 4},
        {"name": "Matias", "age": 8}
    ],
    "age": 37
}
```

この例では、DataWeave は ​`payload.family`​ をストリーミングして、その配列の各要素にランダムアクセスします。ただし、DataWeave はコンテナオブジェクトにはランダムアクセスすることはできません。たとえば、​`age`​ は ​`family`​ の後にあり、DataWeave は後戻りすることはできないため、​`{ a: payload.age , b: payload.family}`​ をストリーミングすることはできません。 ​

## [](#stream_xml)XML のストリーミング

​ XML ではドキュメントに配列が含まれないため、JSON より複雑になります。

XML をストリーミングするため、DataWeave は次のリーダープロパティを使用して、ストリーミングするドキュメント内の場所を定義します。

- ​`collectionPath`

たとえば、次の XML 入力があるとします。 ​

```XML
<order>
    <header>
        <date>Wed Nov 15 13:45:28 EST 2006</date>
        <customer number="123123">Joe</customer>
    </header>
    <order-items>
        <order-item id="31">
            <product>111</product>
            <quantity>2</quantity>
            <price>8.90</price>
        </order-item>
        <order-item id="31">
            <product>222</product>
            <quantity>7</quantity>
            <price>5.20</price>
        </order-item>
        <order-item id="31">
            <product>111</product>
            <quantity>2</quantity>
            <price>8.90</price>
        </order-item>
        <order-item id="31">
            <product>222</product>
            <quantity>7</quantity>
            <price>5.20</price>
        </order-item>
        <order-item id="31">
            <product>222</product>
            <quantity>7</quantity>
            <price>5.20</price>
        </order-item>
    </order-items>
</order>
```

​ この XML ソースデータに対して、​`outputMimeType`​ 値で ​`collectionPath=order.order-items`​ を設定することにより、ストリーミング単位を ​`<order-item/>`​ に設定できます。

```XML
<flow name="dw-streaming-example" >
  <http:listener doc:name="Listener"
      outputMimeType="application/xml; collectionpath=order.order-items; streaming=true"
      config-ref="HTTP_Listener_config" path="/input"/>
</flow>
```

`streaming=true`​ と ​`collectionPath`​ 値の両方を設定する必要があります。いずれかがない場合、コンテンツはストリーミングされません。 次の DataWeave スクリプトは、各 ​`<order-items/>`​ 要素をストリーム可能単位として使用することで、XML 入力をストリーミングします。

```DataWeave
%dw 2.0
output application/xml
---
{
  salesorder: {
    itemList: payload.order."order-items".*"order-item" map {
      ("i_" ++ $$) : {
        id: $.@id,
        productId: $.product,
        quantity: $.quantity,
        price: $.price
      }
    }
  }
}
```

このスクリプトの XML 出力は次のようになります。

```XML
<?xml version='1.0' encoding='UTF-8'?>
<salesorder>
  <itemList>
    <i_0>
      <id>31</id>
      <quantity>2</quantity>
      <productId>111</productId>
      <price>8.90</price>
    </i_0>
  </itemList>
  <itemList>
    <i_1>
      <id>31</id>
      <quantity>7</quantity>
      <productId>222</productId>
      <price>5.20</price>
    </i_1>
  </itemList>
  <itemList>
    <i_2>
      <id>31</id>
      <quantity>2</quantity>
      <productId>111</productId>
      <price>8.90</price>
    </i_2>
  </itemList>
  <itemList>
    <i_3>
      <id>31</id>
      <quantity>7</quantity>
      <productId>222</productId>
      <price>5.20</price>
    </i_3>
  </itemList>
  <itemList>
    <i_4>
      <id>31</id>
      <quantity>7</quantity>
      <productId>222</productId>
      <price>5.20</price>
    </i_4>
  </itemList>
</salesorder>
```

## [](#ストリーミングされたデータのスクリプトの検証-実験的機能)ストリーミングされたデータのスクリプトの検証 (実験的機能)

コードが入力ストリームを正常に処理できることを確認するため、DataWeave は次の​_高度で実験的な_​アノテーションと関連ディレクティブを提供しています。

- `@StreamCapable()`  
    このアノテーションを使用して、スクリプトが変数 (特に ​`payload`​ 変数) に順次アクセスするかどうかを確認してください。
- `input`​ ディレクティブ:  
    `@StreamCapable()`​ アノテーションでは、データソースの MIME タイプを識別するための入力ディレクティブを DataWeave スクリプトで使用する必要があります (例: `input payload application/xml`​)。

DataWeave バリデーター (スクリプト内の ​`@StreamCapable`​ アノテーションでトリガーされます) は、スクリプトが以下の条件を満たしているかどうかを確認します。

- 変数が 1 回だけ参照される。
- インデックスセレクターが負の値にアクセスしていない (例: `[-1]`​)。
- ネストされたラムダで変数への参照がない。

すべての条件が満たされていれば、選択されたデータはストリーム可能です。

次の例は、正常に検証されます。このスクリプトは、​[JSON ストリーム](#stream_json)​セクションの​[JSON 入力](#example_json)​を処理します。

```DataWeave
%dw 2.0

@StreamCapable()
input payload application/json
output application/json
---
payload.family filter (member) -> member.age > 3
```

スクリプトは検証を正しく実行し、次の出力を返します。

```JSON
[
  {
    "name": "Pedro",
    "age": 4
  },
  {
    "name": "Matias",
    "age": 8
  }
]
```

### [](#検証の失敗)検証の失敗

いずれかの条件が満足していないとバリデーターは検証に失敗します。

ストリーミングが機能しても検証に失敗する場合がありますので注意してください。指定したデータソースの入力変数に順次アクセスするようにスクリプトを作成しても、スクリプトが機能しない場合があります。たとえば、JSON では、オブジェクト内のキーの順序に制約を設けていません。JSON ドキュメント内のキーがスクリプトの想定とは異なる順序で入力されていると、ストリーミングは失敗します。アノテーションプロセッサーは、書式設定のルールに従い、キーが常に同じ順序で入力されるとは想定できません。

#### [](#error-variable-is-referenced-more-than-once-エラー-変数が複数回参照されています)Error: Variable Is Referenced More Than Once (エラー: 変数が複数回参照されています)

スクリプトが同じ変数を複数回参照しようとすると、検証は失敗します。

次のスクリプトは、​[JSON ストリーム](#stream_json)​セクションの​[JSON 入力](#example_json)​を処理します。スクリプトが ​`payload`​ 変数を複数回参照しようとしているため、検証は失敗します:

```DataWeave
%dw 2.0

@StreamCapable()
input payload application/json
output application/json
---
 {
     family: payload.family filter (member) -> member.age > 3,
     name: payload.name
 }
```

スクリプトは次のエラーを返して失敗します。

```error
4| input payload application/json streaming=true
         ^^^^^^^
Parameter `payload` is not stream capable.
Reasons:
 - Variable payload is referenced more than once. Locations:
 ---------------------------

8|      family: payload.family filter (member) -> member.age > 3,
                ^^^^^^^
 ---------------------------

9|      name: payload.name
              ^^^^^^^ at
4| input payload application/json streaming=true
```

#### [](#error-wrong-scope-reference-エラー-スコープ参照が正しくありません)Error: Wrong Scope Reference (エラー: スコープ参照が正しくありません)

スクリプトが、変数が定義されているスコープとは異なるスコープの変数を参照しようとすると、検証は失敗します。

次のスクリプトは、​`payload`​ 変数がラムダ式 ​`[1,2,3] map ((item, index) → payload)`​ 内から参照されているために失敗します。式は ​`[1] map ((item, index) → payload`​ ですが、​`payload`​ が間違ったスコープにあるため、ストリーミングは失敗します。

```DataWeave
%dw 2.0

@StreamCapable()
input payload application/json
output application/json
---
[1,2,3] map ((item, index) -> payload)
```

この例は次のエラーを返して失敗します。

```error
4| input payload application/json
         ^^^^^^^
Parameter `payload` is not stream capable.
Reasons:
 - Variable payload is referenced in a different scope from where it was defined. Locations:
 ---------------------------

9| [1,2,3] map ((item, index) -> payload)
                ^^^^^^^^^^^^^^^^^^^^^^^^ at
4| input payload application/json
```

## [](#stream_output)出力のストリーミング

ストリーミングデータの処理後は、別のメッセージプロセッサーに出力を直接ストリーミングすることができます。この動作を簡単に実装するには、​`deferred`​ ライタープロパティを DataWeave スクリプトの output ディレクティブで使用します (例: `output application/json deferred=true`​)。

NOTE

`deferred = true`​ を設定すると、例外は処理されません。たとえば、フローで例外が発生したときに、Studio でこの動作を確認できます。Studio デバッグモードでアプリケーションを実行中に、​`deferred = true`​ に設定されていて Transform Message コンポーネントで例外が発生した場合、コンソールにエラーは記録されますが、フローは Tranform Message コンポーネントで停止しません。

[「JSON のストリーミング」](#stream_json)​の例に基づいた次のフローは、DataWeave スクリプトを使用して、ストリーミングされた入力を絞り込み、出力を Write 操作に直接ストリーミングします。

```XML
<flow name="dw-streamingexample">
  <file:listener doc:name="On New or Updated File"
    config-ref="File_Config" directory="/Users/me/testing/json" recursive="false" outputMimeType="application/json;
    streaming=true">
    <scheduling-strategy>
      <fixed-frequency timeUnit="SECONDS" />
    </scheduling-strategy>
    <file:matcher />
  </file:listener>
  <ee:transform doc:name="Transform Message">
    <ee:message>
      <ee:set-payload><![CDATA[%dw 2.0

@StreamCapable()
input payload application/json
output application/json deferred = true
---
{
   family: payload.family filter (member) -> member.age > 1
}]]></ee:set-payload>
    </ee:message>
  </ee:transform>
  <file:write doc:name="Write"
      config-ref="File_Config2"
      path="/Users/me/testing/output.json"/>
</flow>
```

フローは次の設定を使用します。

1. リスナー (​`<file:listener>`​) は ​`streaming=true`​ を使用して、受信した JSON データをストリームします。
2. `<ee:transform/>`​ の DataWeave スクリプトは、ストリーミングデータのレコードを絞り込み、​`deferred = true`​ プロパティを使用して、結果レコードをフロー内の次のプロセッサーに直接ストリーミングします。
3. フロー内の次のコンポーネントである ​`<file:write/>`​ は、絞り込まれたストリームを直接受け取り、レコードをファイルに書き込みます。

## [](#関連情報)関連情報

- [CSV 形式](https://docs.mulesoft.com/jp/dataweave/2.4/dataweave-formats-csv)
- [JSON 形式](https://docs.mulesoft.com/jp/dataweave/2.4/dataweave-formats-json)
- [XML 形式](https://docs.mulesoft.com/jp/dataweave/2.4/dataweave-formats-xml)
- [サポートされる DataWeave 形式](https://docs.mulesoft.com/jp/dataweave/2.4/dataweave-formats)