---
title: MongoDB
slug: 6B8BYYUI7zQSJ7E1dggAM
docTags: 
createdAt: 2024-07-11T15:18:32.246Z
---

## Ingest Meters from MongoDB

Amberflo supports metering data ingestion from **MongoDB**, allowing you to set up metering **alongside your current system** without disruption.

If your system already uses MongoDB, you can configure Amberflo to ingest data without impacting existing functionality.

### Recommended Methods for MongoDB Integration

We recommend the following approaches to ingest data from MongoDB into Amberflo:

### 1.Using Logstash

Logstash is an open-source data collection engine with real-time data pipelining capabilities.

- It can **unify data from different sources**, normalize it, and forward it to your target systems.
- By using Logstash’s **input, output, filtering, and buffering** capabilities, you can reliably move data from MongoDB to Amberflo.

**Plugin Setup:**

- Use either the **MongoDB input plugin** or the **JDBC input plugin**
- Configure the **S3 output plugin** to write to the **Amberflo-provided S3 bucket**

Once data is in the S3 bucket, Amberflo will automatically pick it up for ingestion.
[Elastic Logstash](docId:8cHAoPY04KVMaRbaid5wV)
[https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html)
[https://github.com/phutchins/logstash-input-mongodb](https://github.com/phutchins/logstash-input-mongodb)

### 2.Query MongoDB using code

You can also query **MongoDB programmatically** using its SDK and then use the **Amberflo SDK** to ingest the retrieved data.

### Example (Python)

- Use **PyMongo** to query data from MongoDB.
- Use the **Amberflo Python SDK** to ingest the results as meter records.

This approach gives you full control over:

- **Filtering and transforming** data before ingestion
- **Handling new data only**, by persisting the **last processed timestamp or ID**

Amberflo automatically performs **deduplication**, so repeated ingestion of the same data will not create duplicates.

### 3.Use MongoDB's Change data stream

With the Amberflo SDK, you can use the MongoDB real-time data stream to act upon every insert:
[https://developer.mongodb.com/quickstart/python-change-streams/
](https://developer.mongodb.com/quickstart/python-change-streams/)
[Python SDK](https://docs.amberflo.io/docs/python-library)

::::CodeDrawer{title="1. Install the SDK" isResponseExpanded="true" autoGeneratedAnchorSlug="1-install-the-sdk" legacyHash="s1hK-Y_GvP9YKv-QZ1nAq"}
:::CodeblockTabsExamples
```none
Set up virtual environment and install SDK.
```
:::

:::CodeblockTabsResponses
```javascript
```
:::
::::

::::CodeDrawer{title="2. Get your API key - Go to Settings and copy your API key." isResponseExpanded="true" autoGeneratedAnchorSlug="2-get-your-api-key-go-to-settings-and-copy-your-ap" legacyHash="l5s02qu9QA9NInmUIPXNB"}
:::CodeblockTabsExamples
```python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]


```
:::

:::CodeblockTabsResponses
```javascript
```
:::
::::

::::CodeDrawer{title="3. Set up the Customer API client - You can use it to manage your customers. Meter records are associated with customers." isResponseExpanded="true" autoGeneratedAnchorSlug="3-set-up-the-customer-api-client-you-can-use-it-to" legacyHash="laWKN4mWUTTHE_OuWT3If"}
:::CodeblockTabsExamples
```python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))
```
:::

:::CodeblockTabsResponses
```javascript
```
:::
::::

::::CodeDrawer{title="4. Create a customer" isResponseExpanded="true" autoGeneratedAnchorSlug="4-create-a-customer" legacyHash="wrm9Z-RTopL95_ustST5R"}
:::CodeblockTabsExamples
```python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="customer-123@example.com",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)
```
:::

:::CodeblockTabsResponses
```javascript
```
:::
::::

::::CodeDrawer{title="5. Set up Ingest API client - By default, it will collect the meter records and send them in batches asynchronously." isResponseExpanded="true" autoGeneratedAnchorSlug="5-set-up-ingest-api-client-by-default-it-will-coll" legacyHash="ascP-vjDnyjQ-1S-LYZth"}
:::CodeblockTabsExamples
```python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="customer-123@example.com",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)

# Set up ingest API client and ingest a meter
from time import time
from metering.ingest import create_ingest_client

ingest_api = create_ingest_client(api_key=API_KEY)
```
:::

:::CodeblockTabsResponses
```javascript
```
:::
::::

::::CodeDrawer{title="6. Send a meter record - Meter records will be collected in batches and sent asynchronously." isResponseExpanded="true" autoGeneratedAnchorSlug="6-send-a-meter-record-meter-records-will-be-collec" legacyHash="8SerLfqWt9QZ0NPAOlvEv"}
:::CodeblockTabsExamples
```python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="customer-123@example.com",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)

# Set up ingest API client and ingest a meter
from time import time
from metering.ingest import create_ingest_client

ingest_api = create_ingest_client(api_key=API_KEY)

ingest_api.meter(
    meter_api_name="api-calls",
    meter_value=1,
    meter_time_in_millis=int(time() * 1000),
    customer_id=customer["customerId"],
)
```
:::

:::CodeblockTabsResponses
```javascript
```
:::
::::

::::CodeDrawer{title="7. Shut down Ingest API client - This is a blocking call that ensures all batches have been sent. Use it in your server shutdown procedure." isResponseExpanded="true" autoGeneratedAnchorSlug="7-shut-down-ingest-api-client-this-is-a-blocking-c" legacyHash="Upf5u6ZnDIh5_3dThkYA0"}
:::CodeblockTabsExamples
```python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="customer-123@example.com",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)

# Set up ingest API client and ingest a meter
from time import time
from metering.ingest import create_ingest_client

ingest_api = create_ingest_client(api_key=API_KEY)

ingest_api.meter(
    meter_api_name="api-calls",
    meter_value=1,
    meter_time_in_millis=int(time() * 1000),
    customer_id=customer["customerId"],
)

# Shut down ingest API client
ingest_api.shutdown()
```
:::

:::CodeblockTabsResponses
```javascript
```
:::
::::

::::CodeDrawer{title="8. Set up Usage API client" isResponseExpanded="true" autoGeneratedAnchorSlug="8-set-up-usage-api-client" legacyHash="eDWJRZOoD75CtMihdHnub"}
:::CodeblockTabsExamples
```python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="customer-123@example.com",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)

# Set up ingest API client and ingest a meter
from time import time
from metering.ingest import create_ingest_client

ingest_api = create_ingest_client(api_key=API_KEY)

ingest_api.meter(
    meter_api_name="api-calls",
    meter_value=1,
    meter_time_in_millis=int(time() * 1000),
    customer_id=customer["customerId"],
)

# Shut down ingest API client
ingest_api.shutdown()

# Setup usage API client and get a usage report
from metering.usage import (AggregationType, Take, TimeGroupingInterval,
                            TimeRange, UsageApiClient, create_usage_query)

usage_api = UsageApiClient(os.environ.get("API_KEY"))
```
:::

:::CodeblockTabsResponses
```javascript
```
:::
::::

::::CodeDrawer{title="9. Get a usage report" isResponseExpanded="true" autoGeneratedAnchorSlug="9-get-a-usage-report" legacyHash="u0aFC0fgnQyLUmE2THMdm"}
:::CodeblockTabsExamples
```python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="customer-123@example.com",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)

# Set up ingest API client and ingest a meter
from time import time
from metering.ingest import create_ingest_client

ingest_api = create_ingest_client(api_key=API_KEY)

ingest_api.meter(
    meter_api_name="api-calls",
    meter_value=1,
    meter_time_in_millis=int(time() * 1000),
    customer_id=customer["customerId"],
)

# Shut down ingest API client
ingest_api.shutdown()

# Setup usage API client and get a usage report
from metering.usage import (AggregationType, Take, TimeGroupingInterval,
                            TimeRange, UsageApiClient, create_usage_query)

usage_api = UsageApiClient(os.environ.get("API_KEY"))

since_two_days_ago = TimeRange(int(time()) - 60 * 60 * 24 * 2)

query = create_usage_query(
    meter_api_name="api-calls",
    aggregation=AggregationType.SUM,
    time_grouping_interval=TimeGroupingInterval.DAY,
    time_range=since_two_days_ago,
    group_by=["customerId"],
    usage_filter={"customerId": ["customer-123"]},
    take=Take(limit=10, is_ascending=False),
)
report = usage_api.get(query)
```
:::

:::CodeblockTabsResponses
```javascript
```
:::
::::