---
schema_version: "1"
id: ad8a473063724635541e02ed72408a0c9e1c000e299c4d39df5995eeaaf0dff0
title: Amazon Kinesis Data Streams Connector 1.0 の例
collection: mulesoft
version: current
locale: ja-jp
portal_url: https://docs.salesforce.com/mulesoft/current/ja-jp/amazon-kinesis-connector/1.0/amazon-kinesis-connector-examples.md
canonical_url: https://docs.mulesoft.com/jp/amazon-kinesis-connector/1.0/amazon-kinesis-connector-examples
source_site: docs.mulesoft.com
content_type: documentation
content_hash: dfbaa03f222671fc1b76ddb049b8834d54893cdaafccafc550f0f35e0e555d36
description: MuleSoft Documentation Site
release: "1.0"
latest: true
---
# Amazon Kinesis Data Streams Connector 1.0 の例

次の例は、Amazon Kinesis Data Streams 用 Anypoint Connector (Amazon Kinesis Data Streams コネクタ) を使用して、データレコードを配置し、Amazon Kinesis データストリームからリスンする方法を示しています。

<table>
<tbody><tr>
<td class="icon">
<i class="fa icon-note" title="Note"></i>
</td>
<td class="content">
<div class="paragraph">
<p>Amazon Kinesis Data Streams Connector は、デフォルトで <a href="https://docs.aws.amazon.com/streams/latest/dev/enhanced-consumers.html">拡張ファンアウト (EFO)</a>​ をサポートしています。これは、コネクタが EFO を使用して設定済みストリームのすべてのシャードに自動的にサブスクライブすることを意味します。</p>
</div>
<div class="paragraph">
<p>コネクタの動作は、AWS Kinesis Client Library (KCL) バージョン 2.0 以降に準拠しています。コネクタは、 <a href="https://docs.aws.amazon.com/streams/latest/dev/building-enhanced-consumers-kcl-java.html">「Java で KCL 2.x を使用して拡張ファンアウトコンシューマーを開発する」</a>​に示されている実装を使用します。</p>
</div>
</td>
</tr>
</tbody></table>

この例を試す前に、Anypoint Studio (Studio) にアクセスし、​**\[Mule Palette (Mule パレット)\]**​ ビューに Kinesis のエントリが表示されていることを確認します。表示されていない場合、​[「プロジェクトへのコネクタの追加」](https://docs.mulesoft.com/jp/amazon-kinesis-connector/1.0/amazon-kinesis-connector-studio#add-connector-to-project)​の指示に従います。

## [](#この例のフロー)この例のフロー

次のスクリーンショットは、この例の Anypoint Studio アプリケーションフローを示しています。

- このフローは、​**Put Records**​ 操作を使用して、Amazon Kinesis データストリームにデータを配置します。このフローでは、​**\[HTTP\] > \[Listener\]**​ は、​**Put Record**​ 操作のペイロードを設定する ​**data**​ という名前のクエリパラメーターを受け取ります。​**partitionKey**​ という名前のクエリパラメーターを使用して、​**Put Record**​ 操作のパーティションキー値を設定します。
    
    ![Put Record 操作フロー](/mulesoft/current/ja-jp/amazon-kinesis-connector/1.0/_images/amazon-kinesis-example-put-record.png)
- このフローは、​**Listener**​ ソースを使用して新しいデータレコードをリスンします。
    
    ![Listener ソースフロー](/mulesoft/current/ja-jp/amazon-kinesis-connector/1.0/_images/amazon-kinesis-example-listener.png)
- このフローは、​**Listener**​ ソースを使用して新しいデータレコードをリスンします。新しいレコードを受信したら、​**Checkpoint**​ 操作をコールして手動チェックポイントを設定します。
    
    ![Checkpoint 操作フロー](/mulesoft/current/ja-jp/amazon-kinesis-connector/1.0/_images/amazon-kinesis-example-checkpoint.png)

## [](#この例の-xml-コード)この例の XML コード

```xml
<?xml version="1.0" encoding="UTF-8"?>

<mule xmlns:os="http://www.mulesoft.org/schema/mule/os" xmlns:ee="http://www.mulesoft.org/schema/mule/ee/core"
      xmlns:http="http://www.mulesoft.org/schema/mule/http"
      xmlns:kinesis="http://www.mulesoft.org/schema/mule/kinesis" xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:doc="http://www.mulesoft.org/schema/mule/documentation" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
http://www.mulesoft.org/schema/mule/kinesis http://www.mulesoft.org/schema/mule/kinesis/current/mule-kinesis.xsd
http://www.mulesoft.org/schema/mule/http http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd
http://www.mulesoft.org/schema/mule/ee/core http://www.mulesoft.org/schema/mule/ee/core/current/mule-ee.xsd
http://www.mulesoft.org/schema/mule/os http://www.mulesoft.org/schema/mule/os/current/mule-os.xsd">

    <configuration-properties doc:name="Configuration properties" file="mule-artifact.properties" />

    <kinesis:config name="Kinesis_config" doc:name="Kinesis config">
        <kinesis:connection accessKey="${config.accessKey}" secretKey="${config.secretKey}" />
    </kinesis:config>
    <http:listener-config name="HTTP_Listener_config" doc:name="HTTP Listener config" >
        <http:listener-connection host="0.0.0.0" port="8081" />
    </http:listener-config>

    <flow name="kinesis-demo-put-record" >
        <http:listener doc:name="Listener" config-ref="HTTP_Listener_config" path="/putrecord"/>
        <kinesis:put-record doc:name="Put Record"  config-ref="Kinesis_config" streamName="${streamName}"
                            partitionKey="#[attributes.queryParams.partitionKey]">
            <kinesis:data ><![CDATA[#[attributes.queryParams.data]]]></kinesis:data>
        </kinesis:put-record>
        <ee:transform doc:name="Transform Message" >
            <ee:message >
                <ee:set-payload ><![CDATA[%dw 2.0
output application/json
---
payload]]></ee:set-payload>
            </ee:message>
        </ee:transform>
        <logger level="INFO" doc:name="Logger" message="Sent message: #[payload]"/>
    </flow>

    <!-- Stop the following flow and enable/start the other one in order to run it -->
    <flow name="kinesis-demo-listener-latest">
        <kinesis:listener doc:name="Listener" config-ref="Kinesis_config"
                          streamName="${streamName}" applicationName="${applicationName}"
                          absolutePosition="LATEST"/>
        <ee:transform doc:name="Transform Message"  >
            <ee:message >
                <ee:set-payload ><![CDATA[%dw 2.0
output application/json
---
payload]]></ee:set-payload>
            </ee:message>
        </ee:transform>
        <logger level="INFO" doc:name="Logger" message="Received message:  #[payload]" />
    </flow>

    <flow name="kinesis-demo-checkpoint" initialState="stopped">
        <kinesis:listener doc:name="Listener" config-ref="Kinesis_config"
                          streamName="${streamName}" applicationName="${applicationName}"
                          absolutePosition="LATEST" checkpointOnComplete="false"/>
        <ee:transform doc:name="Transform Message" >
            <ee:message >
                <ee:set-payload ><![CDATA[%dw 2.0
output application/json
---
payload]]></ee:set-payload>
            </ee:message>
        </ee:transform>
        <logger level="INFO" doc:name="Logger" message="Received message:  #[payload]" />
        <kinesis:checkpoint doc:name="Checkpoint" config-ref="Kinesis_config" applicationName="${applicationName}" streamName="${streamName}"/>
    </flow>

</mule>
```

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

- [Anypoint Connector の概要](https://docs.mulesoft.com/connectors/introduction/introduction-to-anypoint-connectors)
- [MuleSoft ヘルプセンター](https://help.mulesoft.com)