Add SensorsData Connector

Author: zhangqingsong

.github/workflows/backend.yml

@@ -1435,3 +1435,28 @@ jobs:
           ./mvnw -B -T 1 verify -DskipUT=true -DskipIT=false -D"license.skipAddThirdParty"=true -D"skip.ui"=true --no-snapshot-updates -pl :connector-redis-e2e -am -Pci
         env:
           MAVEN_OPTS: -Xmx4096m
+
+  connector-sensorsdata-it:
+    needs: [ changes, sanity-check ]
+    if: needs.changes.outputs.api == 'true' || contains(needs.changes.outputs.it-modules, 'connector-sensorsdata-e2e')
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        java: [ '8', '11' ]
+        os: [ 'ubuntu-latest' ]
+    timeout-minutes: 120
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up JDK ${{ matrix.java }}
+        uses: actions/setup-java@v3
+        with:
+          java-version: ${{ matrix.java }}
+          distribution: 'temurin'
+          cache: 'maven'
+      - name: free disk space
+        run: tools/github/free_disk_space.sh
+      - name: run sensorsdata connector integration test
+        run: |
+          ./mvnw -B -T 1 verify -DskipUT=true -DskipIT=false -D"license.skipAddThirdParty"=true -D"skip.ui"=true --no-snapshot-updates -pl :connector-sensorsdata-e2e -am -Pci
+        env:
+          MAVEN_OPTS: -Xmx4096m

.github/workflows/labeler/label-scope-conf.yml

@@ -315,4 +315,11 @@ aerospike:
   - all:
       - changed-files:
           - any-glob-to-any-file: seatunnel-connectors-v2/connector-aerospike/**
-          - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(aerospike)/**'
+          - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(aerospike)/**'
+
+sensorsdata:
+  - all:
+      - changed-files:
+          - any-glob-to-any-file: seatunnel-connectors-v2/connector-sensorsdata/**
+          - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(sensorsdata)/**'
+

config/plugin_config

@@ -94,3 +94,4 @@ connector-sls
 connector-qdrant
 connector-typesense
 connector-cdc-opengauss
+connector-sensorsdata

docs/en/connector-v2/changelog/connector-sensorsdata.md

@@ -0,0 +1,7 @@
+<details><summary> Change Log </summary>
+
+| Change | Commit | Version |
+| --- | --- | --- |
+
+
+</details>

docs/en/connector-v2/sink/SensorsData.md

@@ -1,3 +1,5 @@
+import ChangeLog from '../changelog/connector-sensorsdata.md';
+
 # SensorsData
 
 > SensorsData sink connector
@@ -18,35 +20,40 @@ A sink plugin which use SensorsData SDK send data records.
 
 ## Options
 
-|        name        |  type   | required | default value |
-|--------------------|---------|----------|---------------|
-| server_url         | string  | yes      | -             |
-| bulk_size          | int     | no       | 50            |
-| max_cache_row_size | int     | no       | 0             |
-| consumer           | string  | no       | -             |
-| entity_name        | string  | yes      | users         |
-| record_type        | string  | yes      | users         |
-| schema             | string  | yes      | users         |
-| distinct_id_column | string  | yes      | -             |
-| identity_fields    | array   | yes      | -             |
-| property_fields    | array   | yes      | -             |
-| event_name         | string  | yes      | -             |
-| time_column        | string  | yes      | -             |
-| time_free          | boolean | no       | false         |
-| detail_id_column   | string  | no       | -             |
-| item_id_column     | string  | no       | -             |
-| item_type_column   | string  | no       | -             |
-| skip_error_record  | boolean | no       | false         |
-| common-options     |         | no       | -             |
-
+| name                      | type    | required | default value |
+|---------------------------|---------|----------|---------------|
+| server_url                | string  | yes      | -             |
+| bulk_size                 | int     | no       | 50            |
+| max_cache_row_size        | int     | no       | 0             |
+| consumer                  | string  | no       | -             |
+| entity_name               | string  | yes      | users         |
+| record_type               | string  | yes      | users         |
+| schema                    | string  | yes      | users         |
+| distinct_id_column        | string  | yes      | -             |
+| identity_fields           | array   | yes      | -             |
+| property_fields           | array   | yes      | -             |
+| event_name                | string  | yes      | -             |
+| time_column               | string  | yes      | -             |
+| time_free                 | boolean | no       | false         |
+| detail_id_column          | string  | no       | -             |
+| item_id_column            | string  | no       | -             |
+| item_type_column          | string  | no       | -             |
+| skip_error_record         | boolean | no       | false         |
+| instant_events            | array   | no       | -             |
+| distinct_id_by_identities | boolean | no       | false         |
+| null_as_profile_unset     | boolean | no       | false         |
+| common-options            |         | no       | -             |
+
+
+## Parameter Interpretation
 ### server_url [string]
 
-SensorsData data sink address, the format is https://${host}:8106/sa?project=${project}
+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 sended. The default value is 50.
+data in the cache will be sent. The default value is 50.
 
 ### max_cache_row_size[int]
 
@@ -79,14 +86,27 @@ The identity fields of the user entity.
 
 ### property_fields[array]
 
-The property fields of the data record.
+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:
 
 1. Fill in the name of the event record.
-2. Use value of a field from upstream data as the event name, the format is ${your field name}, where event name is the
+2. 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:
@@ -96,8 +116,7 @@ For example, Upstream data is the following:
 | 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".
+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]
 
@@ -119,15 +138,29 @@ The item id column of the item entity.
 
 The item type column of the item entity.
 
-### skip_error_record[string]
+### 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](common-options.md) for details
 
-## Example
+## Examples
+
+### Basic Event Tracking
 
 ```hocon
 sink {
@@ -155,9 +188,106 @@ sink {
 }
 ```
 
-## Changelog
+### Dynamic Event Names
+
+```hocon
+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
+
+```hocon
+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
+
+```hocon
+sink {
+  SensorsData {
+    server_url = "http://10.1.136.63:8106/sa?project=default"
+    time_free = true
 
-### 2.3.8-beta 2024-11-13
+    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)
+
+```hocon
+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
 
-- Add SensorsData Sink Connector
+<ChangeLog />
 

plugin-mapping.properties

@@ -144,7 +144,7 @@ seatunnel.source.Opengauss-CDC = connector-cdc-opengauss
 seatunnel.source.GraphQL = connector-graphql
 seatunnel.sink.GraphQL = connector-graphql
 seatunnel.sink.Aerospike = connector-aerospike
-seatunnel.sink.SensorsData-SDK = connector-sensors-sdk
+seatunnel.sink.SensorsData = connector-sensorsdata
 
 seatunnel.transform.Sql = seatunnel-transforms-v2
 seatunnel.transform.FieldMapper = seatunnel-transforms-v2
@@ -163,4 +163,3 @@ seatunnel.transform.FieldRename = seatunnel-transforms-v2
 seatunnel.transform.TableRename = seatunnel-transforms-v2
 seatunnel.transform.TableMerge = seatunnel-transforms-v2
 seatunnel.transform.TableFilter = seatunnel-transforms-v2
-seatunnel.transform.SensorsDataJson = seatunnel-transforms-v2

seatunnel-connectors-v2/connector-sensorsdata/pom.xml

@@ -52,5 +52,10 @@
             <version>${project.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.sensorsdata.analytics.javasdk</groupId>
+            <artifactId>SensorsAnalyticsSDK</artifactId>
+            <version>3.6.9</version>
+        </dependency>
     </dependencies>
 </project>

seatunnel-connectors-v2/connector-sensorsdata/src/main/java/org/apache/seatunnel/connectors/seatunnel/sensorsdata/sdk/config/SensorsDataSDKSinkConfig.java

@@ -39,13 +39,13 @@ public class SensorsDataSDKSinkConfig extends SensorsDataConfigBase {
 
     public SensorsDataSDKSinkConfig(ReadonlyConfig config) {
         super(config);
-        // server 相关
-        this.serverUrl = config.get(SensorsDataSDKOptions.SERVER_URL);
-        this.bulkSize = config.get(SensorsDataSDKOptions.BULK_SIZE);
-        this.maxCacheRowSize = config.get(SensorsDataSDKOptions.MAX_CACHE_ROW_SIZE);
-        this.consumer = config.get(SensorsDataSDKOptions.CONSUMER);
+        // sensorsdata server
+        this.serverUrl = config.get(SensorsDataSDKSinkOptions.SERVER_URL);
+        this.bulkSize = config.get(SensorsDataSDKSinkOptions.BULK_SIZE);
+        this.maxCacheRowSize = config.get(SensorsDataSDKSinkOptions.MAX_CACHE_ROW_SIZE);
+        this.consumer = config.get(SensorsDataSDKSinkOptions.CONSUMER);
         this.instantEvents =
-                Optional.ofNullable(config.get(SensorsDataSDKOptions.INSTANT_EVENT_LIST))
+                Optional.ofNullable(config.get(SensorsDataSDKSinkOptions.INSTANT_EVENT_LIST))
                         .orElse(new ArrayList<>());
     }
 }

seatunnel-connectors-v2/connector-sensorsdata/src/main/java/org/apache/seatunnel/connectors/seatunnel/sensorsdata/sdk/config/SensorsDataSDKOptions.java renamed to seatunnel-connectors-v2/connector-sensorsdata/src/main/java/org/apache/seatunnel/connectors/seatunnel/sensorsdata/sdk/config/SensorsDataSDKSinkOptions.java

@@ -25,25 +25,29 @@
 import java.util.List;
 
 @SuppressWarnings("checkstyle:MagicNumber")
-public interface SensorsDataSDKOptions extends SensorsDataOptions {
+public interface SensorsDataSDKSinkOptions extends SensorsDataOptions {
 
     Option<String> SERVER_URL =
             Options.key("server_url")
                     .stringType()
                     .noDefaultValue()
-                    .withDescription("格式：https://{ip}:8106/sa?project={project}");
+                    .withDescription("Format：https://{ip}:8106/sa?project={project}");
 
     Option<Integer> BULK_SIZE =
             Options.key("bulk_size")
                     .intType()
                     .defaultValue(50)
-                    .withDescription("触发 flush 操作阈值，当内存缓存队列达到该值时，将缓存中的数据批量上报，默认 50");
+                    .withDescription(
+                            "Threshold for triggering flush operation. When the memory cache queue reaches this value, "
+                                    + "the data in the cache will be batch uploaded.");
 
     Option<Integer> MAX_CACHE_ROW_SIZE =
             Options.key("max_cache_row_size")
                     .intType()
                     .defaultValue(0)
-                    .withDescription("最大缓存刷新大小，若超过该值，立即触发 flush 操作，默认为 0 ，根据 bulkSize 来进行判断");
+                    .withDescription(
+                            "Maximum cache refresh size. If it exceeds this value, the flush operation will "
+                                    + "be triggered immediately. The default value is 0, which depends on bulkSize.");
 
     Option<String> CONSUMER =
             Options.key("consumer")
@@ -55,5 +59,6 @@ public interface SensorsDataSDKOptions extends SensorsDataOptions {
             Options.key("instant_events")
                     .listType()
                     .defaultValue(new ArrayList<>())
-                    .withDescription("即时事件的事件名列表，默认为空");
+                    .withDescription(
+                            "Given a list of event names, mark the event as an instant event.");
 }