Amplitude connector
Product analytics for event streams, cohorts, user behavior, and experiments.
Use this as an implementation starting point: adapt the worked example to the requested dataset and the source documentation. See Custom API for the shared SDK pattern and document-upload workflow.
What this connector is for
Product analytics for event streams, cohorts, user behavior, and experiments.
A connector owns access to the source. Separate syncs shape selected resources into tables for analysis and downstream transformations using the Python SDK.
Data and tables you can pull
These are representative dataset families to consider, not a claim that the worked example implements every resource. Access, fields, and availability depend on the source account, product edition, and permissions. Use the vendor reference to select concrete resources.
| Suggested table | One row represents |
|---|---|
events | an event or configured event record |
cohorts | a cohort |
chart_query_results | a chart query result |
Relationships and extraction considerations
Choose stable object IDs and preserve references between related resources. Nested arrays may need child tables. The listed names are proposed output datasets; validate the actual endpoint, fields, permissions, and pagination before adding each sync.
Connecting to Amplitude
Uses the project API key and secret key via Basic authentication. Select US or EU data residency. Export ranges use server_upload_time and data can be delayed by two hours. The API returns a ZIP archive (up to 4 GB); downloads use temporary disk. Preview downloads the first requested hour before returning 100 events, so its source download can be large. A 404 export is surfaced during sync rather than silently treated as complete.
Additional resources may require additional scopes. Connection settings belong to the connector; dataset filters and resource selection belong to each sync. See the CLI workflow for registration and configuration.
Connection settings used by the example
| Setting | Requirement | Purpose |
|---|---|---|
api_key | Required; secret | API Key |
secret_key | Required; secret | Secret Key |
region | Required | Region |
requirements.in
Save these dependencies in requirements.in alongside connector.py; the example includes packages used by its sync. Install the published vortexify package for local development. Keep sync-specific dependencies in the sync's own requirements.in when splitting the implementation into separate resources.
vortexify==0.1.1
httpx>=0.27,<1
pandas>=2,<3No separately installed database driver is needed by this HTTP example.
Worked example: sync events
events — Event export by inclusive upload-hour window; ZIP members contain NDJSON records.
The complete files below implement only this example resource. They illustrate the current SDK signatures, a separate preview, bounded DataFrame output, and source-specific extraction. They do not establish complete source coverage. Live credentials and platform ingestion have not been used to validate this guide.
With default paths, place the connector files in connectors/amplitude/ and sync files in connectors/amplitude/syncs/events/. Use your project configuration if it specifies different paths.
Connection schema: config.json
This file contains a JSON Schema, not actual credential values. Provide those through platform configuration.
{
"type": "object",
"properties": {
"api_key": {
"type": "string",
"title": "API Key",
"minLength": 1,
"x-vortexify-secret": true,
"order": 0
},
"secret_key": {
"type": "string",
"minLength": 1,
"x-vortexify-secret": true,
"title": "Secret Key",
"order": 1
},
"region": {
"type": "string",
"minLength": 1,
"default": "us",
"enum": [
"us",
"eu"
],
"title": "Region",
"order": 2
}
},
"required": [
"api_key",
"secret_key",
"region"
]
}Connector: connector.py
import httpx
from datetime import datetime, timezone, timedelta
from vortexify import Column, Connector, DiscoveredTable
class AmplitudeConnector(Connector):
vendor_domain = "amplitude.com"
def client(self):
if self._client is None or self._client.is_closed:
self._client = httpx.Client(
base_url={
"us": "https://amplitude.com/",
"eu": "https://analytics.eu.amplitude.com/",
}[self.config.get("region", "us")],
auth=(self.config["api_key"], self.config["secret_key"]),
timeout=120,
)
return self._client
def check(self) -> bool:
hour = (datetime.now(timezone.utc) - timedelta(days=1)).strftime("%Y%m%dT%H")
with self.client().stream(
"GET", "api/2/export", params={"start": hour, "end": hour}
) as response:
if response.status_code != 404:
response.raise_for_status()
return True
def discover(self) -> list[DiscoveredTable]:
return [
DiscoveredTable(
name="events",
schema=[
Column(name="uuid", type="string"),
Column(name="event_type", type="string"),
Column(name="event_time", type="timestamp"),
Column(name="user_id", type="string"),
Column(name="document", type="json"),
],
)
]
def close(self) -> None:
if self._client is not None:
try:
self._client.close()
finally:
self._client = NoneSync schema: syncs/events/config.json
{
"type": "object",
"properties": {
"start_hour": {
"type": "string",
"title": "Start Hour",
"description": "YYYYMMDDTHH, inclusive UTC upload hour.",
"minLength": 1,
"order": 0
},
"end_hour": {
"type": "string",
"title": "End Hour",
"description": "YYYYMMDDTHH, inclusive UTC upload hour.",
"minLength": 1,
"order": 1
}
},
"required": [
"start_hour",
"end_hour"
]
}Sync: syncs/events/sync.py
from collections.abc import Iterator
from math import isfinite
import json
import pandas as pd
from vortexify import Sync
from datetime import datetime
from tempfile import TemporaryFile
from zipfile import ZipFile
class AmplitudeEventsSync(Sync):
def preview(self) -> pd.DataFrame:
start = datetime.strptime(self.config["start_hour"], "%Y%m%dT%H")
end = datetime.strptime(self.config["end_hour"], "%Y%m%dT%H")
if end < start or (end - start).total_seconds() >= 365 * 86400:
raise ValueError("Export window must be ordered and at most 365 days")
with TemporaryFile() as archive_file:
with self.client.stream(
"GET",
"api/2/export",
params={
"start": self.config["start_hour"],
"end": self.config["start_hour"],
},
) as response:
response.raise_for_status()
for chunk in response.iter_bytes():
archive_file.write(chunk)
archive_file.seek(0)
with ZipFile(archive_file) as archive:
rows = []
for info in archive.infolist():
if info.is_dir():
continue
with archive.open(info) as member:
for line in member:
if not line.strip():
continue
record = json.loads(line)
rows.append(
{
"uuid": record.get("uuid"),
"event_type": record.get("event_type"),
"event_time": record.get("event_time"),
"user_id": record.get("user_id"),
"document": json.dumps(record, ensure_ascii=False),
}
)
if len(rows) == 100:
return pd.DataFrame(
rows,
columns=[
"uuid",
"event_type",
"event_time",
"user_id",
"document",
],
dtype=object,
)
return pd.DataFrame(
rows,
columns=["uuid", "event_type", "event_time", "user_id", "document"],
dtype=object,
)
def extract(
self, *, avg_row_size_bytes: float | None = None
) -> Iterator[pd.DataFrame]:
memory_mib = self.compute.memory if self.compute else None
memory_mib = memory_mib if memory_mib is not None and memory_mib > 0 else 1024
target_bytes = min(16 * 1024 * 1024, max(1, memory_mib * 1024 * 1024 // 20))
batch_size = 100
if avg_row_size_bytes is not None:
if not isfinite(avg_row_size_bytes) or avg_row_size_bytes <= 0:
raise ValueError("avg_row_size_bytes must be finite and positive")
batch_size = max(1, int(target_bytes / max(1.0, avg_row_size_bytes) / 4))
batch_size = min(batch_size, 10000)
start = datetime.strptime(self.config["start_hour"], "%Y%m%dT%H")
end = datetime.strptime(self.config["end_hour"], "%Y%m%dT%H")
if end < start or (end - start).total_seconds() >= 365 * 86400:
raise ValueError("Export window must be ordered and at most 365 days")
with TemporaryFile() as archive_file:
with self.client.stream(
"GET",
"api/2/export",
params={
"start": self.config["start_hour"],
"end": self.config["end_hour"],
},
) as response:
response.raise_for_status()
for chunk in response.iter_bytes():
archive_file.write(chunk)
archive_file.seek(0)
with ZipFile(archive_file) as archive:
rows = []
for info in archive.infolist():
if info.is_dir():
continue
with archive.open(info) as member:
for line in member:
if not line.strip():
continue
record = json.loads(line)
rows.append(
{
"uuid": record.get("uuid"),
"event_type": record.get("event_type"),
"event_time": record.get("event_time"),
"user_id": record.get("user_id"),
"document": json.dumps(record, ensure_ascii=False),
}
)
if len(rows) == batch_size:
yield pd.DataFrame(
rows,
columns=[
"uuid",
"event_type",
"event_time",
"user_id",
"document",
],
dtype=object,
)
rows = []
if rows:
yield pd.DataFrame(
rows,
columns=[
"uuid",
"event_type",
"event_time",
"user_id",
"document",
],
dtype=object,
)Running and extending the example
Register the connector before the sync and commit each resource separately:
vortexify create connector amplitude
vortexify create sync amplitude/eventsEdit the files, inspect vortexify status, and use the CLI guide to commit and configure them. Registration alone does not run extraction or deploy the connector.
Preview must run with a fresh connector and disposable state. The extraction implementation determines full/incremental behavior; merely setting a mode does not implement change capture. API/driver failures propagate unless explicitly handled in the code. Check source limits and single-record/response sizes as well as output batch size.
Sources and further reading
- Vendor API and data reference — Resource definitions, authentication, fields, and source-specific limits. Check the selected resource's reference when extending the example.
- Vortexify connector and sync contracts — Client ownership, discovery, preview isolation, and DataFrame extraction.
- Vortexify CLI workflow — Project paths, resource registration, source commits, and credential configuration.