Klaviyo

Klaviyo source connector

Description

Reads data from Klaviyo APIs. The connector builds Klaviyo request headers from private_key and revision, then uses the shared HTTP source runtime to parse the response.

Key features

• batch
• stream
• exactly-once
• column projection
• parallelism
• support user-defined split

tip

In streaming mode, the connector repeatedly calls the configured API. Set poll_interval_millis to control the request interval.

Options

name	type	required	default value	description
url	String	Yes	-	Klaviyo API URL.
private_key	String	Yes	-	Klaviyo private API key.
revision	String	Yes	-	Klaviyo API revision, usually in YYYY-MM-DD format.
method	String	No	GET	HTTP method. Supported values are GET and POST.
headers	Map	No	-	Extra HTTP request headers. The connector already sets Authorization, Accept, and revision.
params	Map	No	-	HTTP query parameters.
body	String	No	-	HTTP request body, usually used with POST.
format	String	No	TEXT	Response format. Use json when the response should be parsed by schema, json_field, or content_field.
schema	Config	No	-	Output schema. Required when format = "json".
json_field	Config	No	-	JSONPath mapping from response fields to output columns. Must be used together with schema.
content_field	String	No	-	JSONPath used to select the array or object that should be parsed as rows.
pageing	Config	No	-	Pagination settings. See Pagination.
poll_interval_millis	int	No	-	Request interval in milliseconds for streaming jobs.
retry	int	No	-	Maximum retry count when the request throws IOException.
retry_backoff_multiplier_ms	int	No	100	Retry backoff multiplier in milliseconds.
retry_backoff_max_ms	int	No	10000	Maximum retry backoff in milliseconds.
json_filed_missed_return_null	boolean	No	false	Return null when a field configured in json_field is missing.
common-options	config	No	-	Source common options. See Source Common Options.

Authentication

Set private_key to the Klaviyo private API key. The connector sends it as:

Authorization: Klaviyo-API-Key <private_key>
Accept: application/json
revision: <revision>

Response Parsing

The default format is TEXT, which returns the whole response as one content column.

Use format = "json" and schema when you want structured output:

format = "json"
schema = {
  fields {
    type = string
    id = string
    attributes = {
      name = string
      created = string
      updated = string
    }
    links = {
      self = string
    }
  }
}

Use content_field when the rows are inside a nested JSON node. Use json_field when columns need to be extracted from different JSONPath expressions.

Pagination

pageing can be used when the target API needs paging parameters.

name	type	required	default value	description
total_page_size	long	No	0	Total number of pages to request.
batch_size	int	No	100	Page size returned by each request.
start_page_number	long	No	1	First page number.
page_field	String	No	page	Request parameter name for page number pagination.
page_type	String	No	PageNumber	Pagination type. Supported values are PageNumber and Cursor.
cursor_field	String	No	-	Request parameter name for cursor pagination.
cursor_response_field	String	No	-	JSONPath field used to read the next cursor from the response.
use_placeholder_replacement	boolean	No	false	Use ${field} placeholder replacement in headers, parameters, and body.

Example

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  Klaviyo {
    plugin_output = "klaviyo"
    url = "https://a.klaviyo.com/api/lists"
    private_key = "replace-with-private-key"
    revision = "2020-10-17"
    method = "GET"
    format = "json"
    schema = {
      fields {
        type = string
        id = string
        attributes = {
          name = string
          created = string
          updated = string
        }
        links = {
          self = string
        }
      }
    }
  }
}

sink {
  Console {
    plugin_input = "klaviyo"
  }
}

Changelog

Change Log

Change	Commit	Version
[improve] http connector options (#8969)	https://github.com/apache/seatunnel/commit/63ff9f910a	2.3.10
[Feature][Connector-V2] Support TableSourceFactory/TableSinkFactory on http (#5816)	https://github.com/apache/seatunnel/commit/6f49ec6ead	2.3.4
[Improve][build] Give the maven module a human readable name (#4114)	https://github.com/apache/seatunnel/commit/d7cd601051	2.3.1
[Improve][Project] Code format with spotless plugin. (#4101)	https://github.com/apache/seatunnel/commit/a2ab166561	2.3.1
[Improve][Connector-V2][Http]Improve json parse option rule for all http connector (#3627)	https://github.com/apache/seatunnel/commit/589e4161ec	2.3.0
[Feature][Connector-V2][HTTP] Use json-path parsing (#3510)	https://github.com/apache/seatunnel/commit/1807eb6c95	2.3.0
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)	https://github.com/apache/seatunnel/commit/226dc6a119	2.3.0
[Improve][Connector-V2][Klaviyo]Unified exception for Klaviyo connector (#3555)	https://github.com/apache/seatunnel/commit/08f8615078	2.3.0
[Feature][Connector-V2][Klaviyo]Add Klaviyo source connector (#3443)	https://github.com/apache/seatunnel/commit/fc00a2866b	2.3.0