Version: Next

SensorsData

SensorsData sink connector

Support Those Engines

Spark
Flink
SeaTunnel Zeta

Key Features

Description

The SensorsData sink uses the SensorsData SDK to send SeaTunnel rows as SensorsData records. It supports user events, user detail records, and item records. Use consumer = "console" when you want to print the converted records locally instead of sending them to the SensorsData server.

Sink Options

Name	Type	Required	Default	Description
server_url	string	Yes	-	SensorsData data receiving address, for example `https://host:8106/sa?project=default`.
bulk_size	int	No	50	Flush threshold used by the SensorsData SDK cache.
max_cache_row_size	int	No	0	Maximum cache size before an immediate flush. `0` means it follows `bulk_size`.
consumer	string	No	batch	Consumer type. Use `batch` to send to SensorsData, or `console` to print records for local checks.
entity_name	string	Yes	users	Entity name. Supported values are `users` and `items`.
record_type	string	Yes	users	Record type. Common values are `events`, `details`, and `items`.
schema	string	Conditional	users	Schema name. Required for user records (`entity_name = "users"`).
distinct_id_column	string	Conditional	-	Input column used as the SensorsData distinct ID. Required for user records.
identity_fields	array	Conditional	-	Identity mappings. Required for user records.
property_fields	array	Conditional	-	Property mappings and target data types. Required for user records.
event_name	string	Conditional	-	Event name or `${field_name}` expression. Required when `record_type = "events"`.
time_column	string	Conditional	-	Event time column. Required when `record_type = "events"`. Use `current_time()` to use processing time.
detail_id_column	string	Conditional	-	Detail ID column. Required when `record_type = "details"`.
item_id_column	string	Conditional	-	Item ID column. Required when `record_type = "items"`.
item_type_column	string	Conditional	-	Item type column. Required when `record_type = "items"`.
time_free	boolean	No	false	Whether to enable SensorsData historical data mode.
skip_error_record	boolean	No	false	Whether to skip records that fail conversion.
instant_events	array	No	[]	Event names that should be marked as instant events.
distinct_id_by_identities	boolean	No	false	Fill distinct ID from `identity_fields` when `distinct_id_column` is null.
null_as_profile_unset	boolean	No	false	Convert null profile properties to profile-unset operations.
common-options	config	No	-	Sink common options. See Sink Common Options.

Parameter Interpretation

Record type requirements

For user events, set entity_name = "users" and record_type = "events", then configure event_name, time_column, distinct_id_column, identity_fields, and property_fields.
For user details, set entity_name = "users" and record_type = "details", then configure detail_id_column, distinct_id_column, identity_fields, and property_fields.
For item records, set entity_name = "items" and record_type = "items", then configure item_id_column, item_type_column, and property_fields.

server_url [string]

SensorsData data sink address, the format is https://${host}:8106/sa?project=${project}

bulk_size [int]

Threshold for the triggering flush operation in SensorsData SDK. When the memory cache queue reaches this value, the data in the cache will be sent. The default value is 50.

max_cache_row_size [int]

Maximum cache refresh size for SensorsData SDK. If it exceeds this value, the flush operation will be triggered immediately. The default value is 0, which depends on bulkSize.

consumer [string]

When consumer is set to "console", the data will be output to console instead of send to the server.

entity_name [string]

The entity name of the SensorsData entity data model to receive the data records.

record_type [string]

The record type of the SensorsData entity data model.

schema [string]

The schema name of the SensorsData entity data model.

distinct_id_column [string]

The distinct id column of the user entity.

identity_fields [array]

The identity fields of the user entity.

property_fields [array]

The property fields of the data record. Supported types:

BOOLEAN
DECIMAL
INT
BIGINT
FLOAT
DOUBLE
NUMBER
STRING
DATE
TIMESTAMP
LIST
LIST_COMMA
LIST_SEMICOLON

event_name [string]

Currently, two formats are supported:

Fill in the name of the event record.
Use value of a field from upstream data as the event name, the format is ${your field name}, where event name is the value of the columns of the upstream data.

For example, Upstream data is the following:

name	prop1	prop2
Purchase	16	data-example1
Order	23	data-example2

If ${name} is set as the event name, the event name of the first row is "Purchase", and the event name of the second row is "Order".

time_column [string]

The time column of the event record.

time_free [boolean]

Enable historical data mode.

detail_id_column [string]

The detail id column of the user entity.

item_id_column [string]

The item id column of the item entity.

item_type_column [string]

The item type column of the item entity.

skip_error_record [boolean]

Whether ignore the error in translating the data record.

instant_events [array]

Given a list of event names, mark the event as an instant event.

distinct_id_by_identities [boolean]

When enabled, this option automatically fills the distinct_id using the values from identity_fields columns when the distinct_id_column value is null. This ensures that SensorsData receives a non-null distinct_id value as required.

null_as_profile_unset [boolean]

When enabled, null values in profile properties will be converted to profile unset operations, effectively removing the existing value from the profile.

common options

Sink plugin common parameters, please refer to Sink Common Options for details

Examples

Basic Event Tracking

sink {
  SensorsData {
    server_url = "http://10.1.136.63:8106/sa?project=default"
    time_free = true

    record_type = events
    schema = events
    event_name = "$AppStart"
    time_column = col_date
    distinct_id_column = col_id
    identity_fields = [
      { source = col_id, target = "$identity_login_id" }
      { source = col_id, target = "$identity_distinct_id" }
    ]
    property_fields = [
      { target = prop1, source = col1, type = INT }
      { target = prop2, source = col2, type = BIGINT }
      { target = prop3, source = col3, type = STRING }
      { target = prop4, source = col4, type = BOOLEAN }
    ]
    skip_error_record = true
  }
}

Dynamic Event Names

sink {
  SensorsData {
    server_url = "http://10.1.136.63:8106/sa?project=default"
    time_free = true

    record_type = events
    schema = events
    event_name = "${event_type}"  # Use dynamic event name from data
    time_column = event_timestamp
    distinct_id_column = user_id
    identity_fields = [
      { source = user_id, target = "$identity_login_id" }
      { source = user_id, target = "$identity_distinct_id" }
    ]
    property_fields = [
      { target = "price", source = amount, type = DECIMAL }
      { target = "category", source = product_category, type = STRING }
      { target = "device", source = device_type, type = STRING }
    ]
    instant_events = ["$AppStart", "$AppEnd"]  # Mark specific events as instant
  }
}

Profile Property Updates

sink {
  SensorsData {
    server_url = "http://10.1.136.63:8106/sa?project=default"
    time_free = true

    entity_name = users
    record_type = profile
    schema = users
    distinct_id_column = user_id
    identity_fields = [
      { source = email, target = "$identity_email" }
      { source = phone, target = "$identity_phone" }
    ]
    property_fields = [
      { target = "name", source = full_name, type = STRING }
      { target = "age", source = user_age, type = INT }
      { target = "gender", source = user_gender, type = STRING }
      { target = "location", source = user_location, type = STRING }
    ]
    null_as_profile_unset = true  # Remove properties when null
  }
}

Item Tracking

sink {
  SensorsData {
    server_url = "http://10.1.136.63:8106/sa?project=default"
    time_free = true

    record_type = items
    schema = items
    event_name = "$ItemViewed"
    time_column = view_time
    distinct_id_column = user_id
    identity_fields = [
      { source = user_id, target = "$identity_login_id" }
    ]
    property_fields = [
      { target = "view_duration", source = duration, type = INT }
      { target = "referrer", source = referrer_url, type = STRING }
    ]
    item_id_column = product_id
    item_type_column = product_type
  }
}

Console Output (for Testing)

sink {
  SensorsData {
    server_url = "http://10.1.136.63:8106/sa?project=default"
    consumer = "console"  # Output to console instead of sending to server
    record_type = events
    schema = events
    event_name = "$TestEvent"
    time_column = timestamp
    distinct_id_column = test_id
    property_fields = [
      { target = "test", source = test_field, type = STRING }
    ]
  }
}

Changelog

Change Log

Change	Commit	Version
[Chore] fix typos filed -> field (#9757)	https://github.com/apache/seatunnel/commit/e3e1c67d29	2.3.12
[Feature][connector-v2] Add Sensorsdata Connector Support #9323 (#9432)	https://github.com/apache/seatunnel/commit/bb53f77264	2.3.12

SensorsData

Support Those Engines​

Key Features​

Description​

Sink Options​

Parameter Interpretation​

Record type requirements​

server_url [string]​

bulk_size [int]​

max_cache_row_size [int]​

consumer [string]​

entity_name [string]​

record_type [string]​

schema [string]​

distinct_id_column [string]​

identity_fields [array]​

property_fields [array]​

event_name [string]​

time_column [string]​

time_free [boolean]​

detail_id_column [string]​

item_id_column [string]​

item_type_column [string]​

skip_error_record [boolean]​

instant_events [array]​

distinct_id_by_identities [boolean]​

null_as_profile_unset [boolean]​

common options​

Examples​

Basic Event Tracking​

Dynamic Event Names​

Profile Property Updates​

Item Tracking​

Console Output (for Testing)​

Changelog​