> This page is for Home.

> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.emnify.com/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.emnify.com/_mcp/server.

> Manage data streams with the emnify Portal or REST API

You can manage data streams with the emnify Portal or the Data Streamer API.

## Choose which interface to use

**[emnify Portal](#data-streamer-in-the-portal)**\
Managing data streams with the Portal is convenient when taking the first steps or if you don't expect configuration changes to be needed often.

**[Data Streamer API](#data-streamer-api)**\
Integrating the Data Streamer API becomes a faster and more secure approach when multiple data streams need to be managed as part of the daily business or in the case of advanced operations (for example, seamless migration from one data stream to another without lags or duplicates).

## Data Streamer in the Portal

To [manage your data streams](/multicloud-data-streamer/manage-data-streams), [log in to your emnify Portal account](https://portal.emnify.com/sign/).
Then, navigate to the [**Data Streams**](https://portal.emnify.com/integrations#data-streams) section of the [**Integrations** page](https://portal.emnify.com/integrations).

### View data streams

If there are no data streams configured, the [**Data Streams**](https://portal.emnify.com/integrations#data-streams) panel displays all available [connection types](/multicloud-data-streamer/connection-types) as tiles.

If you already have a data stream configured, the panel displays all existing streams, ordered by creation date (newest on the top).
This list view can be used to gain an overview of the current states of different data streams.

Data streams are marked as **Running** are properly operating.
Failed streams exhibit an **Error** status.
Paused streams are marked as **Paused**.

![Data streams list with Running, Paused, and Error status labels and a Details button on each row](/_fern-img/59b5a8672cfdaf7ca3280d50ae154ebd34105ae519abf9c7b3e4ec76e8285850.webp)

The **Details** button allows you to inspect the configuration of the specific stream.
The **Destination** configuration varies depending on the connection type.

![Expanded Datadog stream with Source stream type and Destination API Key and Region fields](/_fern-img/fdc6df816970ca4321632800ade4794a596d2ee4d6bab2825c48b27adce8b97a.webp)

### Create data streams \[#create-data-streams-portal]

When no configured data streams are available, click **Add** on the preferred connection type tile.
Otherwise, click **Add New Stream** at the top of the existing [**Data Streams**](https://portal.emnify.com/integrations#data-streams) list.
This shows the connection type tiles. You can choose your preferred connection type and click **Add**.

![Add new data stream button above the Data Streams list](/_fern-img/1df05fb823d08d69f9cf8ba218f442cb076add7fb89731ed4a372018f53fc3dd.webp)

Configure the source by selecting your desired [stream type](/multicloud-data-streamer/stream-types).
Optionally, you can add event options for [filtering on event types](/multicloud-data-streamer/manage-data-streams#filter-event-data-streams).

Provide the required settings to configure the destination.

> **Note**
>
> API keys and configuration parameters differ by connection type.
> Instructions for configuring each connection type are in the [Available integrations](/multicloud-data-streamer/integrations) section.

Click **Create** and verify the status of your new stream.

If everything goes well, the data stream is successfully created.
The configuration view closes, and you see the integration status shown as **Running**.

> **Warning**
>
> If the configuration is invalid, a warning displays along with an error message indicating the cause of the problem.
> If this happens, the stream isn't created and you must correct the configuration.

### Inspect and update data streams

With the **Details** button on the Data Stream tiles, you can inspect the configuration and update the [filters for event data streams](/multicloud-data-streamer/manage-data-streams#filter-event-data-streams).

![Event stream details with Stream specific events only selected and an Update location filter](/_fern-img/79294b45539c2b4393e0ee34254db6bed7d2c0e31b89a2b92ca3582af20044db.webp)

### Delete data streams \[#delete-data-streams-portal]

To permanently delete a data stream, click the trash bin icon near the **Details** button.

> **Info**
>
> You'll need to confirm this action.

## Data Streamer API

You can manage your data streams using the emnify Data Streamer API.
Available entry points are listed under the [Integrations section of the emnify REST API reference](/developers/api/integrations).

> **Note**
>
> You can find detailed information on the Data Streamer data properties and compatibility notes in the [Data Streamer API Reference](/developers/reference/data-streamer).

### Authentication

To use the emnify REST API, you need to authenticate with an authentication token.
emnify uses JWTs as the authentication token.

Each integration also requires credentials to verify that the Data Streamer has permission to write data to that service.

> **Tip**
>
> The [Getting Started guide in the Developers documentation](/developers) provides step-by-step instructions for retrieving this token.

### Specification and documentation

The emnify REST API is based on the OpenAPI Specification OAS3.
There's also [interactive API reference](/developers/api).

> **Tip**
>
> Prior OpenAPI knowledge isn't necessary for working with the emnify API.
> Nevertheless, [understanding how an OpenAPI Specification is structured](https://spec.openapis.org/oas/latest.html) could help you navigate the documentation.

The [Data Streamer reference page](/developers/reference/data-streamer) in the Developers documentation provides details about the data structure and common elements you'll encounter while working with the Data Streamer API.

### Work with the Data Streamer API

> **Note**
>
> API keys and configuration parameters differ by [connection type](/multicloud-data-streamer/connection-types).
> Details on setting up, updating, or deleting integrations are in the [emnify REST API reference](/developers/api/integrations).
> It lists all entry points for managing data streams programmatically and contains examples of the available connection types and filtering options.

The following entry points are available for managing data streams:

| Method   | Entrypoint                                     | Description                                           |
| -------- | ---------------------------------------------- | ----------------------------------------------------- |
| `GET`    | `/api/v2/data_stream`                          | List data streams.                                    |
| `GET`    | `/api/v2/data_stream/{data_stream_id}`         | Retrieve detailed information of a data stream by ID. |
| `POST`   | `/api/v2/data_stream`                          | Create a data stream.                                 |
| `PATCH`  | `/api/v2/data_stream/{data_stream_id}`         | Update a data stream by ID.                           |
| `POST`   | `/api/v2/data_stream/{data_stream_id}/restart` | Restart a data stream by ID.                          |
| `DELETE` | `/api/v2/data_stream/{data_stream_id}`         | Delete a data stream by ID.                           |

The following lookup entry points are available for retrieving possible configuration options:

| Method | Entrypoint                              | Description                                 |
| ------ | --------------------------------------- | ------------------------------------------- |
| `GET`  | `/api/v2/data_stream/status`            | List possible data stream statuses.         |
| `GET`  | `/api/v2/data_stream/connection_type`   | List possible data stream connection types. |
| `GET`  | `/api/v2/data_stream/type`              | List possible data stream types.            |
| `GET`  | `/api/v2/data_stream/filter_field_type` | List possible data stream filter fields.    |

### Create data streams \[#create-data-streams-api]

Once authenticated, users can create data streams by sending `POST` requests to `/api/v2/data_stream`.

An example curl to create a data stream would look like the following:

### Request

POST [https://cdn.emnify.net/api/v2/data\_stream](https://cdn.emnify.net/api/v2/data_stream)

**`Amazon S3 - Usage Data`**

```curl Amazon S3 - Usage Data
curl -X POST https://cdn.emnify.net/api/v2/data_stream \
     -H "Authorization: Bearer <token>" \
     -H "Content-Type: application/json" \
     -d '{
  "name": "My S3 Stream",
  "destination": {
    "connection_type": "S3",
    "format": "Csv",
    "credentials": {
      "bucket_name": "demo-bucket",
      "flush_size": 10000,
      "region": "af-south-1",
      "role_arn": "arn:aws:iam:1234567890:role/role_for_s3_data_stream",
      "rotate_interval": 10000
    }
  },
  "data_stream_type": {
    "id": 1
  }
}'
```

**`Amazon S3 - Usage Data`**

```python Amazon S3 - Usage Data
import requests

url = "https://cdn.emnify.net/api/v2/data_stream"

payload = {
    "name": "My S3 Stream",
    "destination": {
        "connection_type": "S3",
        "format": "Csv",
        "credentials": {
            "bucket_name": "demo-bucket",
            "flush_size": 10000,
            "region": "af-south-1",
            "role_arn": "arn:aws:iam:1234567890:role/role_for_s3_data_stream",
            "rotate_interval": 10000
        }
    },
    "data_stream_type": { "id": 1 }
}
headers = {
    "Authorization": "Bearer <token>",
    "Content-Type": "application/json"
}

response = requests.post(url, json=payload, headers=headers)

print(response.json())
```

**`Amazon S3 - Usage Data`**

```javascript Amazon S3 - Usage Data
const url = 'https://cdn.emnify.net/api/v2/data_stream';
const options = {
  method: 'POST',
  headers: {Authorization: 'Bearer <token>', 'Content-Type': 'application/json'},
  body: '{"name":"My S3 Stream","destination":{"connection_type":"S3","format":"Csv","credentials":{"bucket_name":"demo-bucket","flush_size":10000,"region":"af-south-1","role_arn":"arn:aws:iam:1234567890:role/role_for_s3_data_stream","rotate_interval":10000}},"data_stream_type":{"id":1}}'
};

try {
  const response = await fetch(url, options);
  const data = await response.json();
  console.log(data);
} catch (error) {
  console.error(error);
}
```

**`Amazon S3 - Usage Data`**

```go Amazon S3 - Usage Data
package main

import (
	"fmt"
	"strings"
	"net/http"
	"io"
)

func main() {

	url := "https://cdn.emnify.net/api/v2/data_stream"

	payload := strings.NewReader("{\n  \"name\": \"My S3 Stream\",\n  \"destination\": {\n    \"connection_type\": \"S3\",\n    \"format\": \"Csv\",\n    \"credentials\": {\n      \"bucket_name\": \"demo-bucket\",\n      \"flush_size\": 10000,\n      \"region\": \"af-south-1\",\n      \"role_arn\": \"arn:aws:iam:1234567890:role/role_for_s3_data_stream\",\n      \"rotate_interval\": 10000\n    }\n  },\n  \"data_stream_type\": {\n    \"id\": 1\n  }\n}")

	req, _ := http.NewRequest("POST", url, payload)

	req.Header.Add("Authorization", "Bearer <token>")
	req.Header.Add("Content-Type", "application/json")

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(res)
	fmt.Println(string(body))

}
```

**`Amazon S3 - Usage Data`**

```ruby Amazon S3 - Usage Data
require 'uri'
require 'net/http'

url = URI("https://cdn.emnify.net/api/v2/data_stream")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Post.new(url)
request["Authorization"] = 'Bearer <token>'
request["Content-Type"] = 'application/json'
request.body = "{\n  \"name\": \"My S3 Stream\",\n  \"destination\": {\n    \"connection_type\": \"S3\",\n    \"format\": \"Csv\",\n    \"credentials\": {\n      \"bucket_name\": \"demo-bucket\",\n      \"flush_size\": 10000,\n      \"region\": \"af-south-1\",\n      \"role_arn\": \"arn:aws:iam:1234567890:role/role_for_s3_data_stream\",\n      \"rotate_interval\": 10000\n    }\n  },\n  \"data_stream_type\": {\n    \"id\": 1\n  }\n}"

response = http.request(request)
puts response.read_body
```

**`Amazon S3 - Usage Data`**

```java Amazon S3 - Usage Data
import com.mashape.unirest.http.HttpResponse;
import com.mashape.unirest.http.Unirest;

HttpResponse<String> response = Unirest.post("https://cdn.emnify.net/api/v2/data_stream")
  .header("Authorization", "Bearer <token>")
  .header("Content-Type", "application/json")
  .body("{\n  \"name\": \"My S3 Stream\",\n  \"destination\": {\n    \"connection_type\": \"S3\",\n    \"format\": \"Csv\",\n    \"credentials\": {\n      \"bucket_name\": \"demo-bucket\",\n      \"flush_size\": 10000,\n      \"region\": \"af-south-1\",\n      \"role_arn\": \"arn:aws:iam:1234567890:role/role_for_s3_data_stream\",\n      \"rotate_interval\": 10000\n    }\n  },\n  \"data_stream_type\": {\n    \"id\": 1\n  }\n}")
  .asString();
```

**`Amazon S3 - Usage Data`**

```php Amazon S3 - Usage Data
<?php
require_once('vendor/autoload.php');

$client = new \GuzzleHttp\Client();

$response = $client->request('POST', 'https://cdn.emnify.net/api/v2/data_stream', [
  'body' => '{
  "name": "My S3 Stream",
  "destination": {
    "connection_type": "S3",
    "format": "Csv",
    "credentials": {
      "bucket_name": "demo-bucket",
      "flush_size": 10000,
      "region": "af-south-1",
      "role_arn": "arn:aws:iam:1234567890:role/role_for_s3_data_stream",
      "rotate_interval": 10000
    }
  },
  "data_stream_type": {
    "id": 1
  }
}',
  'headers' => [
    'Authorization' => 'Bearer <token>',
    'Content-Type' => 'application/json',
  ],
]);

echo $response->getBody();
```

**`Amazon S3 - Usage Data`**

```csharp Amazon S3 - Usage Data
using RestSharp;

var client = new RestClient("https://cdn.emnify.net/api/v2/data_stream");
var request = new RestRequest(Method.POST);
request.AddHeader("Authorization", "Bearer <token>");
request.AddHeader("Content-Type", "application/json");
request.AddParameter("application/json", "{\n  \"name\": \"My S3 Stream\",\n  \"destination\": {\n    \"connection_type\": \"S3\",\n    \"format\": \"Csv\",\n    \"credentials\": {\n      \"bucket_name\": \"demo-bucket\",\n      \"flush_size\": 10000,\n      \"region\": \"af-south-1\",\n      \"role_arn\": \"arn:aws:iam:1234567890:role/role_for_s3_data_stream\",\n      \"rotate_interval\": 10000\n    }\n  },\n  \"data_stream_type\": {\n    \"id\": 1\n  }\n}", ParameterType.RequestBody);
IRestResponse response = client.Execute(request);
```

**`Amazon S3 - Usage Data`**

```swift Amazon S3 - Usage Data
import Foundation

let headers = [
  "Authorization": "Bearer <token>",
  "Content-Type": "application/json"
]
let parameters = [
  "name": "My S3 Stream",
  "destination": [
    "connection_type": "S3",
    "format": "Csv",
    "credentials": [
      "bucket_name": "demo-bucket",
      "flush_size": 10000,
      "region": "af-south-1",
      "role_arn": "arn:aws:iam:1234567890:role/role_for_s3_data_stream",
      "rotate_interval": 10000
    ]
  ],
  "data_stream_type": ["id": 1]
] as [String : Any]

let postData = JSONSerialization.data(withJSONObject: parameters, options: [])

let request = NSMutableURLRequest(url: NSURL(string: "https://cdn.emnify.net/api/v2/data_stream")! as URL,
                                        cachePolicy: .useProtocolCachePolicy,
                                    timeoutInterval: 10.0)
request.httpMethod = "POST"
request.allHTTPHeaderFields = headers
request.httpBody = postData as Data

let session = URLSession.shared
let dataTask = session.dataTask(with: request as URLRequest, completionHandler: { (data, response, error) -> Void in
  if (error != nil) {
    print(error as Any)
  } else {
    let httpResponse = response as? HTTPURLResponse
    print(httpResponse)
  }
})

dataTask.resume()
```

The request body (set using the `-d` flag in curl) configures the data stream's parameters.

The following JSON request body example shows how to create a stream of event data records with the [Webhook integration](/multicloud-data-streamer/integrations#webhook):

```json
{
  "data_stream_type": {
    "id": 1 // usage data
  },
  "destination": {
    "connection_type": "RestAPI",
    "credentials": {
      "url": "https://my-application-server/servicebus.net",
      "method": "POST", // HTTP method expected by the application server for pushing data records
      "headers": [
        // can be used for authentication or
        // other mapping functionality on the application server side
        "my_custom_header_1:value1",
        "my_custom_header_2:value2"
      ]
    },
    "format": "Json" // further format options will be added in future
  }
}
```

You can also create an event stream with [AWS Kinesis integration](/multicloud-data-streamer/integrations#amazon-kinesis-data-streams) and enabled filtering, for example:

### Request

POST [https://cdn.emnify.net/api/v2/data\_stream](https://cdn.emnify.net/api/v2/data_stream)

**`Amazon Kinesis Data Streams - Event Data with Filters`**

```curl Amazon Kinesis Data Streams - Event Data with Filters
curl -X POST https://cdn.emnify.net/api/v2/data_stream \
     -H "Authorization: Bearer <token>" \
     -H "Content-Type: application/json" \
     -d '{
  "name": "Filtered Kinesis stream",
  "destination": {
    "connection_type": "AwsKinesis",
    "format": "Json",
    "credentials": {
      "region": "af-south-1",
      "role_arn": "arn:aws:iam:1234567890:role/role_for_kinesis_data_stream",
      "stream_name": "demo-stream"
    }
  },
  "filters": [
    {
      "field": "event_type_id",
      "value": [
        "1",
        "2",
        "3"
      ]
    }
  ],
  "data_stream_type": {
    "id": 2
  }
}'
```

**`Amazon Kinesis Data Streams - Event Data with Filters`**

```python Amazon Kinesis Data Streams - Event Data with Filters
import requests

url = "https://cdn.emnify.net/api/v2/data_stream"

payload = {
    "name": "Filtered Kinesis stream",
    "destination": {
        "connection_type": "AwsKinesis",
        "format": "Json",
        "credentials": {
            "region": "af-south-1",
            "role_arn": "arn:aws:iam:1234567890:role/role_for_kinesis_data_stream",
            "stream_name": "demo-stream"
        }
    },
    "filters": [
        {
            "field": "event_type_id",
            "value": ["1", "2", "3"]
        }
    ],
    "data_stream_type": { "id": 2 }
}
headers = {
    "Authorization": "Bearer <token>",
    "Content-Type": "application/json"
}

response = requests.post(url, json=payload, headers=headers)

print(response.json())
```

**`Amazon Kinesis Data Streams - Event Data with Filters`**

```javascript Amazon Kinesis Data Streams - Event Data with Filters
const url = 'https://cdn.emnify.net/api/v2/data_stream';
const options = {
  method: 'POST',
  headers: {Authorization: 'Bearer <token>', 'Content-Type': 'application/json'},
  body: '{"name":"Filtered Kinesis stream","destination":{"connection_type":"AwsKinesis","format":"Json","credentials":{"region":"af-south-1","role_arn":"arn:aws:iam:1234567890:role/role_for_kinesis_data_stream","stream_name":"demo-stream"}},"filters":[{"field":"event_type_id","value":["1","2","3"]}],"data_stream_type":{"id":2}}'
};

try {
  const response = await fetch(url, options);
  const data = await response.json();
  console.log(data);
} catch (error) {
  console.error(error);
}
```

**`Amazon Kinesis Data Streams - Event Data with Filters`**

```go Amazon Kinesis Data Streams - Event Data with Filters
package main

import (
	"fmt"
	"strings"
	"net/http"
	"io"
)

func main() {

	url := "https://cdn.emnify.net/api/v2/data_stream"

	payload := strings.NewReader("{\n  \"name\": \"Filtered Kinesis stream\",\n  \"destination\": {\n    \"connection_type\": \"AwsKinesis\",\n    \"format\": \"Json\",\n    \"credentials\": {\n      \"region\": \"af-south-1\",\n      \"role_arn\": \"arn:aws:iam:1234567890:role/role_for_kinesis_data_stream\",\n      \"stream_name\": \"demo-stream\"\n    }\n  },\n  \"filters\": [\n    {\n      \"field\": \"event_type_id\",\n      \"value\": [\n        \"1\",\n        \"2\",\n        \"3\"\n      ]\n    }\n  ],\n  \"data_stream_type\": {\n    \"id\": 2\n  }\n}")

	req, _ := http.NewRequest("POST", url, payload)

	req.Header.Add("Authorization", "Bearer <token>")
	req.Header.Add("Content-Type", "application/json")

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(res)
	fmt.Println(string(body))

}
```

**`Amazon Kinesis Data Streams - Event Data with Filters`**

```ruby Amazon Kinesis Data Streams - Event Data with Filters
require 'uri'
require 'net/http'

url = URI("https://cdn.emnify.net/api/v2/data_stream")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Post.new(url)
request["Authorization"] = 'Bearer <token>'
request["Content-Type"] = 'application/json'
request.body = "{\n  \"name\": \"Filtered Kinesis stream\",\n  \"destination\": {\n    \"connection_type\": \"AwsKinesis\",\n    \"format\": \"Json\",\n    \"credentials\": {\n      \"region\": \"af-south-1\",\n      \"role_arn\": \"arn:aws:iam:1234567890:role/role_for_kinesis_data_stream\",\n      \"stream_name\": \"demo-stream\"\n    }\n  },\n  \"filters\": [\n    {\n      \"field\": \"event_type_id\",\n      \"value\": [\n        \"1\",\n        \"2\",\n        \"3\"\n      ]\n    }\n  ],\n  \"data_stream_type\": {\n    \"id\": 2\n  }\n}"

response = http.request(request)
puts response.read_body
```

**`Amazon Kinesis Data Streams - Event Data with Filters`**

```java Amazon Kinesis Data Streams - Event Data with Filters
import com.mashape.unirest.http.HttpResponse;
import com.mashape.unirest.http.Unirest;

HttpResponse<String> response = Unirest.post("https://cdn.emnify.net/api/v2/data_stream")
  .header("Authorization", "Bearer <token>")
  .header("Content-Type", "application/json")
  .body("{\n  \"name\": \"Filtered Kinesis stream\",\n  \"destination\": {\n    \"connection_type\": \"AwsKinesis\",\n    \"format\": \"Json\",\n    \"credentials\": {\n      \"region\": \"af-south-1\",\n      \"role_arn\": \"arn:aws:iam:1234567890:role/role_for_kinesis_data_stream\",\n      \"stream_name\": \"demo-stream\"\n    }\n  },\n  \"filters\": [\n    {\n      \"field\": \"event_type_id\",\n      \"value\": [\n        \"1\",\n        \"2\",\n        \"3\"\n      ]\n    }\n  ],\n  \"data_stream_type\": {\n    \"id\": 2\n  }\n}")
  .asString();
```

**`Amazon Kinesis Data Streams - Event Data with Filters`**

```php Amazon Kinesis Data Streams - Event Data with Filters
<?php
require_once('vendor/autoload.php');

$client = new \GuzzleHttp\Client();

$response = $client->request('POST', 'https://cdn.emnify.net/api/v2/data_stream', [
  'body' => '{
  "name": "Filtered Kinesis stream",
  "destination": {
    "connection_type": "AwsKinesis",
    "format": "Json",
    "credentials": {
      "region": "af-south-1",
      "role_arn": "arn:aws:iam:1234567890:role/role_for_kinesis_data_stream",
      "stream_name": "demo-stream"
    }
  },
  "filters": [
    {
      "field": "event_type_id",
      "value": [
        "1",
        "2",
        "3"
      ]
    }
  ],
  "data_stream_type": {
    "id": 2
  }
}',
  'headers' => [
    'Authorization' => 'Bearer <token>',
    'Content-Type' => 'application/json',
  ],
]);

echo $response->getBody();
```

**`Amazon Kinesis Data Streams - Event Data with Filters`**

```csharp Amazon Kinesis Data Streams - Event Data with Filters
using RestSharp;

var client = new RestClient("https://cdn.emnify.net/api/v2/data_stream");
var request = new RestRequest(Method.POST);
request.AddHeader("Authorization", "Bearer <token>");
request.AddHeader("Content-Type", "application/json");
request.AddParameter("application/json", "{\n  \"name\": \"Filtered Kinesis stream\",\n  \"destination\": {\n    \"connection_type\": \"AwsKinesis\",\n    \"format\": \"Json\",\n    \"credentials\": {\n      \"region\": \"af-south-1\",\n      \"role_arn\": \"arn:aws:iam:1234567890:role/role_for_kinesis_data_stream\",\n      \"stream_name\": \"demo-stream\"\n    }\n  },\n  \"filters\": [\n    {\n      \"field\": \"event_type_id\",\n      \"value\": [\n        \"1\",\n        \"2\",\n        \"3\"\n      ]\n    }\n  ],\n  \"data_stream_type\": {\n    \"id\": 2\n  }\n}", ParameterType.RequestBody);
IRestResponse response = client.Execute(request);
```

**`Amazon Kinesis Data Streams - Event Data with Filters`**

```swift Amazon Kinesis Data Streams - Event Data with Filters
import Foundation

let headers = [
  "Authorization": "Bearer <token>",
  "Content-Type": "application/json"
]
let parameters = [
  "name": "Filtered Kinesis stream",
  "destination": [
    "connection_type": "AwsKinesis",
    "format": "Json",
    "credentials": [
      "region": "af-south-1",
      "role_arn": "arn:aws:iam:1234567890:role/role_for_kinesis_data_stream",
      "stream_name": "demo-stream"
    ]
  ],
  "filters": [
    [
      "field": "event_type_id",
      "value": ["1", "2", "3"]
    ]
  ],
  "data_stream_type": ["id": 2]
] as [String : Any]

let postData = JSONSerialization.data(withJSONObject: parameters, options: [])

let request = NSMutableURLRequest(url: NSURL(string: "https://cdn.emnify.net/api/v2/data_stream")! as URL,
                                        cachePolicy: .useProtocolCachePolicy,
                                    timeoutInterval: 10.0)
request.httpMethod = "POST"
request.allHTTPHeaderFields = headers
request.httpBody = postData as Data

let session = URLSession.shared
let dataTask = session.dataTask(with: request as URLRequest, completionHandler: { (data, response, error) -> Void in
  if (error != nil) {
    print(error as Any)
  } else {
    let httpResponse = response as? HTTPURLResponse
    print(httpResponse)
  }
})

dataTask.resume()
```

> **API reference**
>
> [Create data stream](/developers/api/integrations/create-data-streamer)

### List data streams

You can list all data streams belonging to your organization by sending `GET` requests to `/api/v2/data_stream`.

An example curl request to list data streams would look like the following:

### Request

GET [https://cdn.emnify.net/api/v2/data\_stream](https://cdn.emnify.net/api/v2/data_stream)

**`List Data Streams`**

```curl List Data Streams
curl https://cdn.emnify.net/api/v2/data_stream \
     -H "Authorization: Bearer <token>"
```

**`List Data Streams`**

```python List Data Streams
import requests

url = "https://cdn.emnify.net/api/v2/data_stream"

headers = {"Authorization": "Bearer <token>"}

response = requests.get(url, headers=headers)

print(response.json())
```

**`List Data Streams`**

```javascript List Data Streams
const url = 'https://cdn.emnify.net/api/v2/data_stream';
const options = {method: 'GET', headers: {Authorization: 'Bearer <token>'}};

try {
  const response = await fetch(url, options);
  const data = await response.json();
  console.log(data);
} catch (error) {
  console.error(error);
}
```

**`List Data Streams`**

```go List Data Streams
package main

import (
	"fmt"
	"net/http"
	"io"
)

func main() {

	url := "https://cdn.emnify.net/api/v2/data_stream"

	req, _ := http.NewRequest("GET", url, nil)

	req.Header.Add("Authorization", "Bearer <token>")

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(res)
	fmt.Println(string(body))

}
```

**`List Data Streams`**

```ruby List Data Streams
require 'uri'
require 'net/http'

url = URI("https://cdn.emnify.net/api/v2/data_stream")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Get.new(url)
request["Authorization"] = 'Bearer <token>'

response = http.request(request)
puts response.read_body
```

**`List Data Streams`**

```java List Data Streams
import com.mashape.unirest.http.HttpResponse;
import com.mashape.unirest.http.Unirest;

HttpResponse<String> response = Unirest.get("https://cdn.emnify.net/api/v2/data_stream")
  .header("Authorization", "Bearer <token>")
  .asString();
```

**`List Data Streams`**

```php List Data Streams
<?php
require_once('vendor/autoload.php');

$client = new \GuzzleHttp\Client();

$response = $client->request('GET', 'https://cdn.emnify.net/api/v2/data_stream', [
  'headers' => [
    'Authorization' => 'Bearer <token>',
  ],
]);

echo $response->getBody();
```

**`List Data Streams`**

```csharp List Data Streams
using RestSharp;

var client = new RestClient("https://cdn.emnify.net/api/v2/data_stream");
var request = new RestRequest(Method.GET);
request.AddHeader("Authorization", "Bearer <token>");
IRestResponse response = client.Execute(request);
```

**`List Data Streams`**

```swift List Data Streams
import Foundation

let headers = ["Authorization": "Bearer <token>"]

let request = NSMutableURLRequest(url: NSURL(string: "https://cdn.emnify.net/api/v2/data_stream")! as URL,
                                        cachePolicy: .useProtocolCachePolicy,
                                    timeoutInterval: 10.0)
request.httpMethod = "GET"
request.allHTTPHeaderFields = headers

let session = URLSession.shared
let dataTask = session.dataTask(with: request as URLRequest, completionHandler: { (data, response, error) -> Void in
  if (error != nil) {
    print(error as Any)
  } else {
    let httpResponse = response as? HTTPURLResponse
    print(httpResponse)
  }
})

dataTask.resume()
```

> **API reference**
>
> [List data stream configurations of your organization](/developers/api/integrations/list-data-streamer-v-2-s)

### Retrieve details

Retrieving details of an existing data stream is possible by making a `GET` request to `/api/v2/data_stream/{data_stream_id}`.

The `data_stream_id` path parameter is the top-level `id` property in each object returned by the [list data streams](#list-data-streams) request and is an ID that's unique for each data stream.

An example curl request to get detailed information for a data stream would look like the following:

### Request

GET [https://cdn.emnify.net/api/v2/data\_stream/\{data\_stream\_id}](https://cdn.emnify.net/api/v2/data_stream/\{data_stream_id})

**`Active Data Stream Configuration`**

```curl Active Data Stream Configuration
curl https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4 \
     -H "Authorization: Bearer <token>"
```

**`Active Data Stream Configuration`**

```python Active Data Stream Configuration
import requests

url = "https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4"

headers = {"Authorization": "Bearer <token>"}

response = requests.get(url, headers=headers)

print(response.json())
```

**`Active Data Stream Configuration`**

```javascript Active Data Stream Configuration
const url = 'https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4';
const options = {method: 'GET', headers: {Authorization: 'Bearer <token>'}};

try {
  const response = await fetch(url, options);
  const data = await response.json();
  console.log(data);
} catch (error) {
  console.error(error);
}
```

**`Active Data Stream Configuration`**

```go Active Data Stream Configuration
package main

import (
	"fmt"
	"net/http"
	"io"
)

func main() {

	url := "https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4"

	req, _ := http.NewRequest("GET", url, nil)

	req.Header.Add("Authorization", "Bearer <token>")

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(res)
	fmt.Println(string(body))

}
```

**`Active Data Stream Configuration`**

```ruby Active Data Stream Configuration
require 'uri'
require 'net/http'

url = URI("https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Get.new(url)
request["Authorization"] = 'Bearer <token>'

response = http.request(request)
puts response.read_body
```

**`Active Data Stream Configuration`**

```java Active Data Stream Configuration
import com.mashape.unirest.http.HttpResponse;
import com.mashape.unirest.http.Unirest;

HttpResponse<String> response = Unirest.get("https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4")
  .header("Authorization", "Bearer <token>")
  .asString();
```

**`Active Data Stream Configuration`**

```php Active Data Stream Configuration
<?php
require_once('vendor/autoload.php');

$client = new \GuzzleHttp\Client();

$response = $client->request('GET', 'https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4', [
  'headers' => [
    'Authorization' => 'Bearer <token>',
  ],
]);

echo $response->getBody();
```

**`Active Data Stream Configuration`**

```csharp Active Data Stream Configuration
using RestSharp;

var client = new RestClient("https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4");
var request = new RestRequest(Method.GET);
request.AddHeader("Authorization", "Bearer <token>");
IRestResponse response = client.Execute(request);
```

**`Active Data Stream Configuration`**

```swift Active Data Stream Configuration
import Foundation

let headers = ["Authorization": "Bearer <token>"]

let request = NSMutableURLRequest(url: NSURL(string: "https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4")! as URL,
                                        cachePolicy: .useProtocolCachePolicy,
                                    timeoutInterval: 10.0)
request.httpMethod = "GET"
request.allHTTPHeaderFields = headers

let session = URLSession.shared
let dataTask = session.dataTask(with: request as URLRequest, completionHandler: { (data, response, error) -> Void in
  if (error != nil) {
    print(error as Any)
  } else {
    let httpResponse = response as? HTTPURLResponse
    print(httpResponse)
  }
})

dataTask.resume()
```

> **API reference**
>
> [Get details on existing data stream](/developers/api/integrations/get-data-streamer-by-id-v-2)

### Update data streams

Authenticated users may update data streams belonging to their organization by making a `PATCH` request to `/api/v2/data_stream/{data_stream_id}`.

The `data_stream_id` path parameter is the top-level `id` property in each object returned by the [list data streams](#list-data-streams) request and is an ID that's unique for each data stream.

An example curl request to update a data stream:

### Request

PATCH [https://cdn.emnify.net/api/v2/data\_stream/\{data\_stream\_id}](https://cdn.emnify.net/api/v2/data_stream/\{data_stream_id})

**`Change Data Stream configuration`**

```curl Change Data Stream configuration
curl -X PATCH https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4 \
     -H "Authorization: Bearer <token>" \
     -H "Content-Type: application/json" \
     -d '{
  "destination": {
    "connection_type": "RestAPI",
    "credentials": {
      "url": "https://example.com"
    }
  },
  "filters": [],
  "status": {
    "id": 2
  }
}'
```

**`Change Data Stream configuration`**

```python Change Data Stream configuration
import requests

url = "https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4"

payload = {
    "destination": {
        "connection_type": "RestAPI",
        "credentials": { "url": "https://example.com" }
    },
    "filters": [],
    "status": { "id": 2 }
}
headers = {
    "Authorization": "Bearer <token>",
    "Content-Type": "application/json"
}

response = requests.patch(url, json=payload, headers=headers)

print(response.json())
```

**`Change Data Stream configuration`**

```javascript Change Data Stream configuration
const url = 'https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4';
const options = {
  method: 'PATCH',
  headers: {Authorization: 'Bearer <token>', 'Content-Type': 'application/json'},
  body: '{"destination":{"connection_type":"RestAPI","credentials":{"url":"https://example.com"}},"filters":[],"status":{"id":2}}'
};

try {
  const response = await fetch(url, options);
  const data = await response.json();
  console.log(data);
} catch (error) {
  console.error(error);
}
```

**`Change Data Stream configuration`**

```go Change Data Stream configuration
package main

import (
	"fmt"
	"strings"
	"net/http"
	"io"
)

func main() {

	url := "https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4"

	payload := strings.NewReader("{\n  \"destination\": {\n    \"connection_type\": \"RestAPI\",\n    \"credentials\": {\n      \"url\": \"https://example.com\"\n    }\n  },\n  \"filters\": [],\n  \"status\": {\n    \"id\": 2\n  }\n}")

	req, _ := http.NewRequest("PATCH", url, payload)

	req.Header.Add("Authorization", "Bearer <token>")
	req.Header.Add("Content-Type", "application/json")

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(res)
	fmt.Println(string(body))

}
```

**`Change Data Stream configuration`**

```ruby Change Data Stream configuration
require 'uri'
require 'net/http'

url = URI("https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Patch.new(url)
request["Authorization"] = 'Bearer <token>'
request["Content-Type"] = 'application/json'
request.body = "{\n  \"destination\": {\n    \"connection_type\": \"RestAPI\",\n    \"credentials\": {\n      \"url\": \"https://example.com\"\n    }\n  },\n  \"filters\": [],\n  \"status\": {\n    \"id\": 2\n  }\n}"

response = http.request(request)
puts response.read_body
```

**`Change Data Stream configuration`**

```java Change Data Stream configuration
import com.mashape.unirest.http.HttpResponse;
import com.mashape.unirest.http.Unirest;

HttpResponse<String> response = Unirest.patch("https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4")
  .header("Authorization", "Bearer <token>")
  .header("Content-Type", "application/json")
  .body("{\n  \"destination\": {\n    \"connection_type\": \"RestAPI\",\n    \"credentials\": {\n      \"url\": \"https://example.com\"\n    }\n  },\n  \"filters\": [],\n  \"status\": {\n    \"id\": 2\n  }\n}")
  .asString();
```

**`Change Data Stream configuration`**

```php Change Data Stream configuration
<?php
require_once('vendor/autoload.php');

$client = new \GuzzleHttp\Client();

$response = $client->request('PATCH', 'https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4', [
  'body' => '{
  "destination": {
    "connection_type": "RestAPI",
    "credentials": {
      "url": "https://example.com"
    }
  },
  "filters": [],
  "status": {
    "id": 2
  }
}',
  'headers' => [
    'Authorization' => 'Bearer <token>',
    'Content-Type' => 'application/json',
  ],
]);

echo $response->getBody();
```

**`Change Data Stream configuration`**

```csharp Change Data Stream configuration
using RestSharp;

var client = new RestClient("https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4");
var request = new RestRequest(Method.PATCH);
request.AddHeader("Authorization", "Bearer <token>");
request.AddHeader("Content-Type", "application/json");
request.AddParameter("application/json", "{\n  \"destination\": {\n    \"connection_type\": \"RestAPI\",\n    \"credentials\": {\n      \"url\": \"https://example.com\"\n    }\n  },\n  \"filters\": [],\n  \"status\": {\n    \"id\": 2\n  }\n}", ParameterType.RequestBody);
IRestResponse response = client.Execute(request);
```

**`Change Data Stream configuration`**

```swift Change Data Stream configuration
import Foundation

let headers = [
  "Authorization": "Bearer <token>",
  "Content-Type": "application/json"
]
let parameters = [
  "destination": [
    "connection_type": "RestAPI",
    "credentials": ["url": "https://example.com"]
  ],
  "filters": [],
  "status": ["id": 2]
] as [String : Any]

let postData = JSONSerialization.data(withJSONObject: parameters, options: [])

let request = NSMutableURLRequest(url: NSURL(string: "https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4")! as URL,
                                        cachePolicy: .useProtocolCachePolicy,
                                    timeoutInterval: 10.0)
request.httpMethod = "PATCH"
request.allHTTPHeaderFields = headers
request.httpBody = postData as Data

let session = URLSession.shared
let dataTask = session.dataTask(with: request as URLRequest, completionHandler: { (data, response, error) -> Void in
  if (error != nil) {
    print(error as Any)
  } else {
    let httpResponse = response as? HTTPURLResponse
    print(httpResponse)
  }
})

dataTask.resume()
```

The request body (set using the `-d` flag in curl) configures the parameters of the data stream itself.

The following shows where this example paused the stream and erased any filters:

```json
{
  "status": {
    "id": 2 // 1 = Running, 2 = Paused, 3 = Error, 4 = Pending
  },
  "filters": []
}
```

> **API reference**
>
> [Modify existing data stream](/developers/api/integrations/patch-v-2-data-stream)

### Delete data streams \[#delete-data-streams-api]

Sending a `DELETE` request to `/api/v2/data_stream/{data_stream_id}` deletes the data stream with that ID.

The `data_stream_id` path parameter is the top-level `id` property in each object returned by the [list data streams](#list-data-streams) request and is an ID that's unique for each data stream.

The following curl request would then delete a data stream:

### Request

DELETE [https://cdn.emnify.net/api/v2/data\_stream/\{data\_stream\_id}](https://cdn.emnify.net/api/v2/data_stream/\{data_stream_id})

```curl
curl -X DELETE https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4 \
     -H "Authorization: Bearer <token>"
```

```python
import requests

url = "https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4"

headers = {"Authorization": "Bearer <token>"}

response = requests.delete(url, headers=headers)

print(response.json())
```

```javascript
const url = 'https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4';
const options = {method: 'DELETE', headers: {Authorization: 'Bearer <token>'}};

try {
  const response = await fetch(url, options);
  const data = await response.json();
  console.log(data);
} catch (error) {
  console.error(error);
}
```

```go
package main

import (
	"fmt"
	"net/http"
	"io"
)

func main() {

	url := "https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4"

	req, _ := http.NewRequest("DELETE", url, nil)

	req.Header.Add("Authorization", "Bearer <token>")

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(res)
	fmt.Println(string(body))

}
```

```ruby
require 'uri'
require 'net/http'

url = URI("https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Delete.new(url)
request["Authorization"] = 'Bearer <token>'

response = http.request(request)
puts response.read_body
```

```java
import com.mashape.unirest.http.HttpResponse;
import com.mashape.unirest.http.Unirest;

HttpResponse<String> response = Unirest.delete("https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4")
  .header("Authorization", "Bearer <token>")
  .asString();
```

```php
<?php
require_once('vendor/autoload.php');

$client = new \GuzzleHttp\Client();

$response = $client->request('DELETE', 'https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4', [
  'headers' => [
    'Authorization' => 'Bearer <token>',
  ],
]);

echo $response->getBody();
```

```csharp
using RestSharp;

var client = new RestClient("https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4");
var request = new RestRequest(Method.DELETE);
request.AddHeader("Authorization", "Bearer <token>");
IRestResponse response = client.Execute(request);
```

```swift
import Foundation

let headers = ["Authorization": "Bearer <token>"]

let request = NSMutableURLRequest(url: NSURL(string: "https://cdn.emnify.net/api/v2/data_stream/a1b2c3d4")! as URL,
                                        cachePolicy: .useProtocolCachePolicy,
                                    timeoutInterval: 10.0)
request.httpMethod = "DELETE"
request.allHTTPHeaderFields = headers

let session = URLSession.shared
let dataTask = session.dataTask(with: request as URLRequest, completionHandler: { (data, response, error) -> Void in
  if (error != nil) {
    print(error as Any)
  } else {
    let httpResponse = response as? HTTPURLResponse
    print(httpResponse)
  }
})

dataTask.resume()
```

> **API reference**
>
> [Delete existing data stream](/developers/api/integrations/delete-data-streamer-v-2)