Repository: lf-edge/ekuiper
Pull request #2990 (2024-07-05 06:13:34)

docs(source): add down sampling doc


Patch:
diff --git a/docs/directory.json b/docs/directory.json
index adf399d7..41c22268 100644
--- a/docs/directory.json
+++ b/docs/directory.json
@@ -173,6 +173,10 @@
           "title": "数据源",
           "path": "guide/sources/overview",
           "children": [
+            {
+              "title": "降采样",
+              "path": "guide/sources/down_sample"
+            },
             {
               "title": "内置源",
               "collapsed": true,
@@ -925,6 +929,10 @@
           "title": "Source Connector",
           "path": "guide/sources/overview",
           "children": [
+            {
+              "title": "Down sampling",
+              "path": "guide/sources/down_sample"
+            },
             {
               "title": "Built-in Sources",
               "collapsed": true,
diff --git a/docs/en_US/guide/sources/down_sample.md b/docs/en_US/guide/sources/down_sample.md
new file mode 100644
index 00000000..abb6afaf
--- /dev/null
+++ b/docs/en_US/guide/sources/down_sample.md
@@ -0,0 +1,285 @@
+# Down Sampling
+
+The process of down sampling aggregates the input data at the source, thereby reducing the computation required for data
+decoding and transformation. The functionality of down sampling itself can be achieved through the aggregation
+calculation using a time window. However, the time window uses the full sampled data source, and the discarded data has
+actually undergone unnecessary computations such as decoding. Compared to the time window scheme, down sampling at the
+data source mainly enhances the performance of data processing. The greater the difference between the sampling
+frequency and the input frequency, the more significant the performance and resource utilization improvement. Data
+source down sampling is suitable for scenarios with higher performance requirements.
+
+## Applicable Types
+
+Data streams can be categorized into PullSource and PushSource based on the method of data acquisition. PullSource is a
+data stream that samples data periodically, and down sampling can be achieved by setting the sampling rate. The data
+inflow frequency of PushSource is determined by the push speed of the data source. In eKuiper, common configurations
+such as sampling frequency are provided, and various strategies for down sampling can also be implemented.
+
+## Configuration
+
+eKuiper data sources provide the following common properties for configuring down sampling characteristics.
+
+### interval
+
+The sampling interval, in the format of a Go duration string, such as "10s", "500ms", etc.
+For PullSource, this property determines the interval at which data is pulled.
+For PushSource, the data pushed by the data source will accumulate until the sampling interval, and then it will be
+emitted according to the configured sampling strategy.
+
+### mergeField
+
+The default down sampling strategy takes the last message within the sampling period. If mergeField is configured, it
+will take the last message corresponding to each mergeField and then merge them into a single message for output.
+
+The mergeField is configured as the column name in the message, such as "id". Data with different ids will be aggregated
+into a single downs ample data.
+
+Notes:
+
+- Only formats that support columnar decoding support this property. Currently, only the JSON format is supported. Users
+  can implement `message.PartialDecoder` to support columnar decoding for custom formats.
+- The mergeField currently only supports top-level non-composite fields. Nested fields or composite types such as arrays
+  or structures will be aggregated as the same key.
+
+## Down sampling Strategy Examples
+
+The input for Source down sampling is N messages within the sampling period, and the output is a single message. How to
+convert from N messages to a single message is determined by the down sampling strategy. Currently, we support two down
+sampling strategies: taking the latest value and aggregating by column.
+
+Taking the MQTT data source as an example, by subscribing to the MQTT topic, the data source will receive pushed data,
+such as 1Hz or higher frequency data. Suppose the calculation does not require such high-frequency data; we can
+configure a 1-second down sampling to reduce the consumption of computing resources. The following are examples of
+configurations for the two down sampling strategies.
+
+### Taking the Latest Value
+
+This strategy will output the last data received within the sampling period when it receives N data.
+
+1. Create down sampling configuration: Through the following REST API, we create the MQTT configuration `onesec`. It
+   includes the configuration item `interval`, setting the sampling period to 1 second. Note: The MQTT server address
+   and other configurations will follow the default settings.
+
+    ```http request
+    PUT http://{{host}}/metadata/sources/mqtt/confKeys/onesec
+    
+    {
+      "interval": "1s"
+    }
+    ```
+
+2. Create data stream: The following API creates a data stream named `mqttOneSec`, which uses the down sampling
+   configuration created in step 1 through `CONF_KEY="onesec"`.
+
+    ```http request
+    POST http://{{host}}/streams
+    Content-Type: application/json
+    
+    {
+      "sql": "CREATE STREAM mqttOneSec() WITH (TYPE=\"mqtt\",FORMAT=\"json\",DATASOURCE=\"demo\",CONF_KEY=\"onesec\");"
+    }
+    ```
+
+3. Create rules based on the down sampling data stream: Next, users can create rules based on this data stream. The
+   following is the simplest rule, which sends all data to MQTT. This rule will receive data at a 1-second sampling
+   rate, that is, the last data of each second.
+
+    ```http request
+    POST http://{{host}}/rules
+    Content-Type: application/json
+    
+    {
+      "id": "ruleOneSecLatest",
+      "sql": "SELECT * FROM mqttOneSec",
+      "actions": [
+        {
+          "mqtt": {
+            "server": "tcp://127.0.0.1:1883",
+            "topic": "result/onesec",
+            "sendSingle": true
+          }
+        }
+      ]
+    }
+    ```
+
+### Column Aggregation
+
+This strategy, when it receives N data entries within the sampling period, will distinguish these entries by the values
+of the specified columns and merge the last entry for each column into a complete data entry. This strategy is suitable
+for situations where the data stream inherently contains data from different sources with different schemas. For
+example, the input might be:
+
+```json lines
+{
+  "id": 1,
+  "temperature": 20
+}
+{
+  "id": 2,
+  "humidity": 80
+}
+{
+  "id": 1,
+  "temperature": 30
+}
+```
+
+The aggregated result is a single data entry:
+
+```json
+{
+  "id": 1,
+  "temperature": 30,
+  "humidity": 80
+}
+```
+
+1. Create downsampling configuration: Through the following REST API, we create the MQTT configuration `onesec_merge`.
+   It includes the `interval` configuration item, setting the sampling period to 1 second. It also configures
+   the `mergeField` for aggregation. Note: The MQTT server address and other configurations will follow the default
+   settings.
+
+    ```http request
+    PUT http://{{host}}/metadata/sources/mqtt/confKeys/onesec_merge
+    
+    {
+      "interval": "1s",
+      "mergeField": "id"
+    }
+    ```
+
+2. Create data stream: The following API creates a data stream named `mqttOneSecM`, which adopts the downsampling
+   configuration created in the first step through `CONF_KEY="onesec_merge"`.
+
+    ```http request
+    POST http://{{host}}/streams
+    Content-Type: application/json
+    
+    {
+      "sql": "CREATE STREAM mqttOneSecM() WITH (TYPE=\"mqtt\",FORMAT=\"json\",DATASOURCE=\"demo\",CONF_KEY=\"onesec_merge\");"
+    }
+    ```
+
+3. Create rules based on the downsampling data stream: Next, users can create rules based on this data stream. The
+   following is the simplest rule, which takes all data and sends it to MQTT. This rule will receive data at a 1-second
+   sampling rate, aggregate it into a single data entry by column, and then send it.
+
+    ```http request
+    POST http://{{host}}/rules
+    Content-Type: application/json
+    
+    {
+      "id": "RuleOneSecM",
+      "sql": "SELECT * FROM mqttOneSecM",
+      "actions": [
+        {
+          "mqtt": {
+            "server": "tcp://127.0.0.1:1883",
+            "topic": "result/onesecm",
+            "sendSingle": true
+          }
+        }
+      ]
+    }
+    ```
+
+### Full Aggregation?
+
+The column-based aggregation rule requires a specified column for aggregation. What if you want to aggregate without
+specifying a column? It is recommended to use a time window approach and perform aggregation through the `merge_agg`
+function.
+
+```SQL
+SELECT merge_agg(*)
+FROM normalStream
+GROUP BY TumblingWindow(ss, 1)
+```
+
+Why this design: The original intention of the down sampling data stream is to reduce unnecessary computations, thereby
+decreasing CPU and resource usage. The current two down sampling strategies only require minimal decoding computations
+to determine whether the data needs to be sampled and computed. Full aggregation, however, requires a resource-intensive
+decoding operation for each data entry. Essentially, there is no down sampling at the data source level; instead, data
+transformation is performed after full sampling. Therefore, users can directly use a fully sampled data source and
+perform aggregation operations through window functions.
+
+## Down sampling Metrics Observation
+
+Data source downsampling is implemented in the `ratelimit` operator. To obtain rule execution metrics, you can observe
+the operation of downsampling through the metrics of the `ratelimit` operator. For example, in the following case:
+`source_mqttOneMiMerge_0_records_out_total` indicates that the MQTT data source has read in 25 data entries.
+The metrics of the `ratelimit` operator include `op_2_ratelimit_0_records_in_total`: 25
+and `op_2_ratelimit_0_records_out_total`: 1, which means that 25 data entries were read in and downsample to 1 entry.
+The subsequent `decode` operator only needs to parse the downsampled data.
+
+```json
+{
+  "status": "running",
+  "lastStartTimestamp": "1720151899579",
+  "lastStopTimestamp": "0",
+  "nextStopTimestamp": "0",
+  "source_mqttOneMiMerge_0_records_in_total": 25,
+  "source_mqttOneMiMerge_0_records_out_total": 25,
+  "source_mqttOneMiMerge_0_messages_processed_total": 25,
+  "source_mqttOneMiMerge_0_process_latency_us": 0,
+  "source_mqttOneMiMerge_0_buffer_length": 0,
+  "source_mqttOneMiMerge_0_last_invocation": "2024-07-05T11:58:40.733398",
+  "source_mqttOneMiMerge_0_exceptions_total": 0,
+  "source_mqttOneMiMerge_0_last_exception": "",
+  "source_mqttOneMiMerge_0_last_exception_time": 0,
+  "op_2_ratelimit_0_records_in_total": 25,
+  "op_2_ratelimit_0_records_out_total": 1,
+  "op_2_ratelimit_0_messages_processed_total": 25,
+  "op_2_ratelimit_0_process_latency_us": 0,
+  "op_2_ratelimit_0_buffer_length": 0,
+  "op_2_ratelimit_0_last_invocation": "2024-07-05T11:58:40.733398",
+  "op_2_ratelimit_0_exceptions_total": 0,
+  "op_2_ratelimit_0_last_exception": "",
+  "op_2_ratelimit_0_last_exception_time": 0,
+  "op_3_payload_decoder_0_records_in_total": 1,
+  "op_3_payload_decoder_0_records_out_total": 1,
+  "op_3_payload_decoder_0_messages_processed_total": 1,
+  "op_3_payload_decoder_0_process_latency_us": 0,
+  "op_3_payload_decoder_0_buffer_length": 0,
+  "op_3_payload_decoder_0_last_invocation": "2024-07-05T11:59:19.59698",
+  "op_3_payload_decoder_0_exceptions_total": 0,
+  "op_3_payload_decoder_0_last_exception": "",
+  "op_3_payload_decoder_0_last_exception_time": 0,
+  "op_4_project_0_records_in_total": 1,
+  "op_4_project_0_records_out_total": 1,
+  "op_4_project_0_messages_processed_total": 1,
+  "op_4_project_0_process_latency_us": 0,
+  "op_4_project_0_buffer_length": 0,
+  "op_4_project_0_last_invocation": "2024-07-05T11:59:19.59698",
+  "op_4_project_0_exceptions_total": 0,
+  "op_4_project_0_last_exception": "",
+  "op_4_project_0_last_exception_time": 0,
+  "op_mqtt_0_0_transform_0_records_in_total": 1,
+  "op_mqtt_0_0_transform_0_records_out_total": 1,
+  "op_mqtt_0_0_transform_0_messages_processed_total": 1,
+  "op_mqtt_0_0_transform_0_process_latency_us": 0,
+  "op_mqtt_0_0_transform_0_buffer_length": 0,
+  "op_mqtt_0_0_transform_0_last_invocation": "2024-07-05T11:59:19.59698",
+  "op_mqtt_0_0_transform_0_exceptions_total": 0,
+  "op_mqtt_0_0_transform_0_last_exception": "",
+  "op_mqtt_0_0_transform_0_last_exception_time": 0,
+  "op_mqtt_0_1_encode_0_records_in_total": 1,
+  "op_mqtt_0_1_encode_0_records_out_total": 1,
+  "op_mqtt_0_1_encode_0_messages_processed_total": 1,
+  "op_mqtt_0_1_encode_0_process_latency_us": 0,
+  "op_mqtt_0_1_encode_0_buffer_length": 0,
+  "op_mqtt_0_1_encode_0_last_invocation": "2024-07-05T11:59:19.59698",
+  "op_mqtt_0_1_encode_0_exceptions_total": 0,
+  "op_mqtt_0_1_encode_0_last_exception": "",
+  "op_mqtt_0_1_encode_0_last_exception_time": 0,
+  "sink_mqtt_0_0_records_in_total": 1,
+  "sink_mqtt_0_0_records_out_total": 1,
+  "sink_mqtt_0_0_messages_processed_total": 1,
+  "sink_mqtt_0_0_process_latency_us": 0,
+  "sink_mqtt_0_0_buffer_length": 0,
+  "sink_mqtt_0_0_last_invocation": "2024-07-05T11:59:19.59698",
+  "sink_mqtt_0_0_exceptions_total": 0,
+  "sink_mqtt_0_0_last_exception": "",
+  "sink_mqtt_0_0_last_exception_time": 0
+}
+```
diff --git a/docs/zh_CN/guide/sources/down_sample.md b/docs/zh_CN/guide/sources/down_sample.md
new file mode 100644
index 00000000..30f0fa66
--- /dev/null
+++ b/docs/zh_CN/guide/sources/down_sample.md
@@ -0,0 +1,243 @@
+# 降采样
+
+降采样在数据源处对输入数据进行聚合操作，从而减少数据解码，变换等计算。降采样的功能本身可以通过时间窗口进行聚合计算的方式实现。然而时间窗口使用的是全采样数据源，丢弃的数据实际上也进行了解码等不必要的计算。相比时间窗口方案，数据源降采样主要是提升了数据处理的性能。采样频率与输入频率差别越大，性能和资源占用提升也越明显。数据源降采样适用于对性能要求更高的场景。
+
+## 适用类型
+
+数据流按照获取数据的方式可分为 PullSource 和 PushSource。其中，PullSource 为定期采样数据流，通过设置采样率可实现降采样的目的。而
+PushSource 的数据流入频率由数据源的推送速度决定。在 eKuiper 中，提供通用的采样频率等配置，也可以实现各种策略的降采样。
+
+## 配置
+
+eKuiper 数据源提供了以下通用属性用于配置降采样特性。
+
+### interval
+
+采样间隔，格式为 go 的 duration 字符串，例如 "10s", "500ms" 等。
+对于 PullSource，该属性决定了拉数据的间隔。
+对于 PushSource，数据源推送的数据会累计，直到采样间隔再通过配置的采样策略发出。
+
+### mergeField
+
+默认的降采样策略取采样周期内最后的消息。若配置 mergeField，则会取每个 mergeField 对应的最后一条消息，然后合并成一条消息输出。
+
+mergeField 配置为消息中的列名，例如 id。则不同 id 的数据会聚合为一条降采样数据。
+注意：
+
+- 只有支持按列解码的格式支持配置此属性。当前，仅有 JSON 格式支持。用户自定义的格式可通过实现 `message.PartialDecoder`
+  实现按列解码。
+- mergeField 当前仅支持顶层非复合类型的字段。嵌套字段或者数组/结构体等复合类型将作为相同 key 聚合。
+
+## 降采样策略案例
+
+Source 降采样输入都是采样周期内的 N 条消息，输出为单条消息。如何从 N 条消息转为单条消息是由降采样策略决定的。目前我们支持取最新值以及按列聚合两种降采样策略。
+
+以 MQTT 数据源为例，通过订阅 MQTT 主题，数据源会收到推送的数据，例如 1Hz 或者更高频率的数据。假设计算中不需要如此高频的数据，我们可以配置
+1s 降采样，减少计算的资源消耗。以下为两种降采样策略的配置案例。
+
+### 取最新值
+
+该策略在采样周期内收到 N 条数据，会输出最后一条数据。
+
+1. 创建降采样配置：通过以下 REST API, 我们创建了 MQTT 配置 `onesec`。其中包含了配置项 `interval`, 配置采样周期为 1
+   秒。注意：MQTT 服务器地址等配置，将会沿用默认配置。
+
+    ```http request
+    ###
+    PUT http://{{host}}/metadata/sources/mqtt/confKeys/onesec
+    
+    {
+      "interval": "1s"
+    }
+    
+    ```
+
+2. 创建数据流：以下 API 创建了名为 mqttOneSec 数据流，通过 `CONF_KEY="onesec"` 采用了第一步创建的降采样配置。
+
+    ```http request
+    ###
+    POST http://{{host}}/streams
+    Content-Type: application/json
+    
+    {
+      "sql": "CREATE STREAM mqttOneSec() WITH (TYPE=\"mqtt\",FORMAT=\"json\",DATASOURCE=\"demo\",CONF_KEY=\"onesec\");"
+    }
+    ```
+
+3. 基于降采样数据流创建规则：接下来用户可基于该数据流创建规则。以下为最简单的规则，取出所有数据发送到 MQTT 。该规则将收到 1s
+   采样率的数据，即每一秒的最后一条数据。
+
+    ```http request
+    ###
+    POST http://{{host}}/rules
+    Content-Type: application/json
+    
+    {
+      "id": "ruleOneSecLatest",
+      "sql": "SELECT * FROM mqttOneSec",
+      "actions": [
+        {
+          "mqtt": {
+            "server": "tcp://127.0.0.1:1883",
+            "topic": "result/onesec",
+            "sendSingle": true
+          }
+        }
+      ]
+    }
+    ```
+
+### 按列聚合
+
+该策略在采样周期内收到 N 条数据，会将这些数据按指定的列的值区分，合并每种列的最后一条数据为一条完整的数据。该策略适合数据流本身包含不同来源不同
+schema 数据的情况。例如，输入可能是
+
+```json lines
+{"id":1, "temperature":20}
+{"id":2, "humidity":80}
+{"id":1, "temperature":30}
+```
+
+聚合后结果为单条数据：
+
+```json
+{"id":1, "temperature":30, "humidity":80}
+```
+
+1. 创建降采样配置：通过以下 REST API, 我们创建了 MQTT 配置 `onesec_merge`。其中包含了配置项 `interval`, 配置采样周期为 1
+   秒。也配置了聚合列 `mergeField`。注意：MQTT 服务器地址等配置，将会沿用默认配置。
+
+    ```http request
+    ###
+    PUT http://{{host}}/metadata/sources/mqtt/confKeys/onesec_merge
+    
+    {
+      "interval": "1s",
+      "mergeField": "id"
+    }
+    ```
+
+2. 创建数据流：以下 API 创建了名为 mqttOneSecM 数据流，通过 `CONF_KEY="onesec_merge"` 采用了第一步创建的降采样配置。
+
+   ```http request
+    ###
+    POST http://{{host}}/streams
+    Content-Type: application/json
+    
+    {
+      "sql": "CREATE STREAM mqttOneSecM() WITH (TYPE=\"mqtt\",FORMAT=\"json\",DATASOURCE=\"demo\",CONF_KEY=\"onesec_merge\");"
+    }
+    ```
+
+3. 基于降采样数据流创建规则：接下来用户可基于该数据流创建规则。以下为最简单的规则，取出所有数据发送到 MQTT 。该规则将收到 1s
+   采样率的数据，按列聚合为单条数据后发送。
+
+   ```http request
+    ###
+    POST http://{{host}}/rules
+    Content-Type: application/json
+    
+    {
+      "id": "RuleOneSecM",
+      "sql": "SELECT * FROM mqttOneSecM",
+      "actions": [
+        {
+          "mqtt": {
+            "server": "tcp://127.0.0.1:1883",
+            "topic": "result/onesecm",
+            "sendSingle": true
+          }
+        }
+      ]
+    }
+    ```
+
+### 全聚合？
+
+按列聚合规则需要指定聚合列，如果不指定列做聚合应该如何做呢？建议采用时间窗口的方式，通过 merge_agg 函数进行聚合。
+
+```SQL
+SELECT merge_agg(*) FROM normalStream GROUP BY TumblingWindow(ss, 1)
+```
+
+为何这样设计：降采样数据流的初衷是减少不必要的计算，从而降低 CPU
+和资源占用。目前两种降采样策略都只需要很少的解码计算量即可判断数据是否需要采样计算。而全聚合需要对每条数据进行消耗大量资源的解码操作，本质上数据源测并没有降采样，而是在全采样之后再做数据变换。因此，用户可直接采用全采样数据源，通过窗口函数进行聚合运算。
+
+## 降采样指标观测
+
+数据源降采样在 ratelimit 算子中实现。获取规则运行指标，通过其中 ratelimit 算子的运行指标可观测降采样的运行情况。如下例子中：
+`source_mqttOneMiMerge_0_records_out_total` 可知 MQTT 数据源读入 25 条数据。
+ratelimit 算子的指标包括 `op_2_ratelimit_0_records_in_total`: 25 和 `op_2_ratelimit_0_records_out_total`: 1 表示读入 25
+条数据，并降采样为 1 条。之后的 decode 算子仅需解析降采样后的数据。
+
+```json
+{
+  "status": "running",
+  "lastStartTimestamp": "1720151899579",
+  "lastStopTimestamp": "0",
+  "nextStopTimestamp": "0",
+  "source_mqttOneMiMerge_0_records_in_total": 25,
+  "source_mqttOneMiMerge_0_records_out_total": 25,
+  "source_mqttOneMiMerge_0_messages_processed_total": 25,
+  "source_mqttOneMiMerge_0_process_latency_us": 0,
+  "source_mqttOneMiMerge_0_buffer_length": 0,
+  "source_mqttOneMiMerge_0_last_invocation": "2024-07-05T11:58:40.733398",
+  "source_mqttOneMiMerge_0_exceptions_total": 0,
+  "source_mqttOneMiMerge_0_last_exception": "",
+  "source_mqttOneMiMerge_0_last_exception_time": 0,
+  "op_2_ratelimit_0_records_in_total": 25,
+  "op_2_ratelimit_0_records_out_total": 1,
+  "op_2_ratelimit_0_messages_processed_total": 25,
+  "op_2_ratelimit_0_process_latency_us": 0,
+  "op_2_ratelimit_0_buffer_length": 0,
+  "op_2_ratelimit_0_last_invocation": "2024-07-05T11:58:40.733398",
+  "op_2_ratelimit_0_exceptions_total": 0,
+  "op_2_ratelimit_0_last_exception": "",
+  "op_2_ratelimit_0_last_exception_time": 0,
+  "op_3_payload_decoder_0_records_in_total": 1,
+  "op_3_payload_decoder_0_records_out_total": 1,
+  "op_3_payload_decoder_0_messages_processed_total": 1,
+  "op_3_payload_decoder_0_process_latency_us": 0,
+  "op_3_payload_decoder_0_buffer_length": 0,
+  "op_3_payload_decoder_0_last_invocation": "2024-07-05T11:59:19.59698",
+  "op_3_payload_decoder_0_exceptions_total": 0,
+  "op_3_payload_decoder_0_last_exception": "",
+  "op_3_payload_decoder_0_last_exception_time": 0,
+  "op_4_project_0_records_in_total": 1,
+  "op_4_project_0_records_out_total": 1,
+  "op_4_project_0_messages_processed_total": 1,
+  "op_4_project_0_process_latency_us": 0,
+  "op_4_project_0_buffer_length": 0,
+  "op_4_project_0_last_invocation": "2024-07-05T11:59:19.59698",
+  "op_4_project_0_exceptions_total": 0,
+  "op_4_project_0_last_exception": "",
+  "op_4_project_0_last_exception_time": 0,
+  "op_mqtt_0_0_transform_0_records_in_total": 1,
+  "op_mqtt_0_0_transform_0_records_out_total": 1,
+  "op_mqtt_0_0_transform_0_messages_processed_total": 1,
+  "op_mqtt_0_0_transform_0_process_latency_us": 0,
+  "op_mqtt_0_0_transform_0_buffer_length": 0,
+  "op_mqtt_0_0_transform_0_last_invocation": "2024-07-05T11:59:19.59698",
+  "op_mqtt_0_0_transform_0_exceptions_total": 0,
+  "op_mqtt_0_0_transform_0_last_exception": "",
+  "op_mqtt_0_0_transform_0_last_exception_time": 0,
+  "op_mqtt_0_1_encode_0_records_in_total": 1,
+  "op_mqtt_0_1_encode_0_records_out_total": 1,
+  "op_mqtt_0_1_encode_0_messages_processed_total": 1,
+  "op_mqtt_0_1_encode_0_process_latency_us": 0,
+  "op_mqtt_0_1_encode_0_buffer_length": 0,
+  "op_mqtt_0_1_encode_0_last_invocation": "2024-07-05T11:59:19.59698",
+  "op_mqtt_0_1_encode_0_exceptions_total": 0,
+  "op_mqtt_0_1_encode_0_last_exception": "",
+  "op_mqtt_0_1_encode_0_last_exception_time": 0,
+  "sink_mqtt_0_0_records_in_total": 1,
+  "sink_mqtt_0_0_records_out_total": 1,
+  "sink_mqtt_0_0_messages_processed_total": 1,
+  "sink_mqtt_0_0_process_latency_us": 0,
+  "sink_mqtt_0_0_buffer_length": 0,
+  "sink_mqtt_0_0_last_invocation": "2024-07-05T11:59:19.59698",
+  "sink_mqtt_0_0_exceptions_total": 0,
+  "sink_mqtt_0_0_last_exception": "",
+  "sink_mqtt_0_0_last_exception_time": 0
+}
+```
diff --git a/internal/binder/io/builtin.go b/internal/binder/io/builtin.go
index e38ed8c7..eb7cc113 100644
--- a/internal/binder/io/builtin.go
+++ b/internal/binder/io/builtin.go
@@ -47,7 +47,7 @@ func init() {
 	modules.RegisterSink("memory", func() api.Sink { return memory.GetSink() })
 	modules.RegisterSink("neuron", neuron.GetSink)
 	modules.RegisterSink("file", file.GetSink)
-	// modules.RegisterSink("websocket", func() api.Sink { return &websocket.WebSocketSink{} })
+	modules.RegisterSink("websocket", func() api.Sink { return websocket.GetSink() })
 
 	modules.RegisterLookupSource("memory", memory.GetLookupSource)
 	// modules.RegisterLookupSource("httppull", func() api.LookupSource { return http.GetLookUpSource() })
diff --git a/internal/io/http/httpserver/data_server.go b/internal/io/http/httpserver/data_server.go
index 8db68fc8..d2b22b6a 100644
--- a/internal/io/http/httpserver/data_server.go
+++ b/internal/io/http/httpserver/data_server.go
@@ -33,6 +33,7 @@ import (
 
 type GlobalServerManager struct {
 	sync.RWMutex
+	instanceID        int
 	endpoint          map[string]string
 	server            *http.Server
 	router            *mux.Router
diff --git a/internal/io/http/httpserver/websocketConn.go b/internal/io/http/httpserver/websocketConn.go
index 22c98c51..c106b974 100644
--- a/internal/io/http/httpserver/websocketConn.go
+++ b/internal/io/http/httpserver/websocketConn.go
@@ -22,6 +22,7 @@ import (
 
 type WebsocketConnection struct {
 	RecvTopic string
+	SendTopic string
 	cfg       *connectionCfg
 }
 
@@ -46,12 +47,13 @@ func createWebsocketServerConnection(ctx api.StreamContext, props map[string]any
 	if err := cast.MapToStruct(props, cfg); err != nil {
 		return nil, err
 	}
-	recvTopic, err := RegisterWebSocketEndpoint(ctx, cfg.Datasource)
+	rTopic, sTopic, err := RegisterWebSocketEndpoint(ctx, cfg.Datasource)
 	if err != nil {
 		return nil, err
 	}
 	return &WebsocketConnection{
-		RecvTopic: recvTopic,
+		RecvTopic: rTopic,
+		SendTopic: sTopic,
 		cfg:       cfg,
 	}, nil
 }
diff --git a/internal/io/http/httpserver/websocket_server.go b/internal/io/http/httpserver/websocket_server.go
index d2137bdf..fbef1d61 100644
--- a/internal/io/http/httpserver/websocket_server.go
+++ b/internal/io/http/httpserver/websocket_server.go
@@ -32,12 +32,20 @@ const (
 	WebsocketTopicPrefix = "$$websocket/"
 )
 
+func recvTopic(endpoint string) string {
+	return fmt.Sprintf("recv/%s/%s", WebsocketTopicPrefix, endpoint)
+}
+
+func sendTopic(endpoint string) string {
+	return fmt.Sprintf("send/%s/%s", WebsocketTopicPrefix, endpoint)
+}
+
 type websocketEndpointContext struct {
 	wg    *sync.WaitGroup
 	conns map[*websocket.Conn]context.CancelFunc
 }
 
-func RegisterWebSocketEndpoint(ctx api.StreamContext, endpoint string) (string, error) {
+func RegisterWebSocketEndpoint(ctx api.StreamContext, endpoint string) (string, string, error) {
 	return manager.RegisterWebSocketEndpoint(ctx, endpoint)
 }
 
@@ -49,14 +57,56 @@ func UnRegisterWebSocketEndpoint(endpoint string) {
 	}
 }
 
-func (m *GlobalServerManager) recvProcess(ctx api.StreamContext, endpoint string, c *websocket.Conn, wg *sync.WaitGroup) {
+func (m *GlobalServerManager) handleProcess(ctx api.StreamContext, endpoint string, instanceID int, c *websocket.Conn, cancel context.CancelFunc, parWg *sync.WaitGroup) {
 	defer func() {
 		m.CloseEndpointConnection(endpoint, c)
-		conf.Log.Infof("websocket endpoint %v stop recvProcess", endpoint)
+		parWg.Done()
+	}()
+	subWg := &sync.WaitGroup{}
+	subWg.Add(2)
+	go m.recvProcess(ctx, endpoint, c, cancel, subWg)
+	go m.sendProcess(ctx, endpoint, instanceID, c, cancel, subWg)
+	subWg.Wait()
+}
+
+func (m *GlobalServerManager) sendProcess(ctx api.StreamContext, endpoint string, instanceID int, c *websocket.Conn, cancel context.CancelFunc, wg *sync.WaitGroup) {
+	conf.Log.Infof("websocket endpoint %v start sendProcess", endpoint)
+	topic := sendTopic(endpoint)
+	sourceID := fmt.Sprintf("ws/send/%v", instanceID)
+	defer func() {
+		pubsub.CloseSourceConsumerChannel(topic, sourceID)
+		cancel()
+		c.Close()
+		wg.Done()
+		conf.Log.Infof("websocket send endpoint %v stop sendProcess", endpoint)
+	}()
+	ch := pubsub.CreateSub(topic, nil, sourceID, 1024)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case d := <-ch:
+			data := d.([]byte)
+			if err := c.WriteMessage(websocket.TextMessage, data); err != nil {
+				if websocket.IsCloseError(err) || strings.Contains(err.Error(), "close") {
+					conf.Log.Infof("websocket endpoint %s connection get closed: %v", endpoint, err)
+					return
+				}
+				conf.Log.Warnf("websocket endpoint %v send data meet error: %v", endpoint, err)
+			}
+		}
+	}
+}
+
+func (m *GlobalServerManager) recvProcess(ctx api.StreamContext, endpoint string, c *websocket.Conn, cancel context.CancelFunc, wg *sync.WaitGroup) {
+	defer func() {
+		cancel()
+		c.Close()
 		wg.Done()
+		conf.Log.Infof("websocket recv endpoint %v stop recvProcess", endpoint)
 	}()
 	conf.Log.Infof("websocket endpoint %v start recvProcess", endpoint)
-	topic := fmt.Sprintf("recv/%s/%s", WebsocketTopicPrefix, endpoint)
+	topic := recvTopic(endpoint)
 	for {
 		select {
 		case <-ctx.Done():
@@ -81,12 +131,13 @@ func (m *GlobalServerManager) recvProcess(ctx api.StreamContext, endpoint string
 	}
 }
 
-func (m *GlobalServerManager) RegisterWebSocketEndpoint(ctx api.StreamContext, endpoint string) (string, error) {
+func (m *GlobalServerManager) RegisterWebSocketEndpoint(ctx api.StreamContext, endpoint string) (string, string, error) {
 	conf.Log.Infof("websocket endpoint %v register", endpoint)
 	m.Lock()
 	defer m.Unlock()
-	recvTopic := fmt.Sprintf("recv/%s/%s", WebsocketTopicPrefix, endpoint)
-	pubsub.CreatePub(recvTopic)
+	rTopic := recvTopic(endpoint)
+	sTopic := sendTopic(endpoint)
+	pubsub.CreatePub(rTopic)
 	m.router.HandleFunc(endpoint, func(w http.ResponseWriter, r *http.Request) {
 		c, err := m.upgrader.Upgrade(w, r, nil)
 		if err != nil {
@@ -95,16 +146,16 @@ func (m *GlobalServerManager) RegisterWebSocketEndpoint(ctx api.StreamContext, e
 		}
 		subCtx, cancel := ctx.WithCancel()
 		wg := m.AddEndpointConnection(endpoint, c, cancel)
+		go m.handleProcess(subCtx, endpoint, m.FetchInstanceID(), c, cancel, wg)
 		conf.Log.Infof("websocket endpint %v create connection", endpoint)
-		wg.Add(1)
-		go m.recvProcess(subCtx, endpoint, c, wg)
 	})
 	conf.Log.Infof("websocker endpoint %v registered success", endpoint)
-	return recvTopic, nil
+	return rTopic, sTopic, nil
 }
 
 func (m *GlobalServerManager) UnRegisterWebSocketEndpoint(endpoint string) *websocketEndpointContext {
 	conf.Log.Infof("websocket endpoint %v unregister", endpoint)
+	pubsub.RemovePub(recvTopic(endpoint))
 	m.Lock()
 	defer m.Unlock()
 	wctx, ok := m.websocketEndpoint[endpoint]
@@ -140,6 +191,7 @@ func (m *GlobalServerManager) AddEndpointConnection(endpoint string, c *websocke
 		return wctx.wg
 	}
 	wg := &sync.WaitGroup{}
+	wg.Add(1)
 	m.websocketEndpoint[endpoint] = &websocketEndpointContext{
 		wg: wg,
 		conns: map[*websocket.Conn]context.CancelFunc{
@@ -149,6 +201,11 @@ func (m *GlobalServerManager) AddEndpointConnection(endpoint string, c *websocke
 	return wg
 }
 
+func (m *GlobalServerManager) FetchInstanceID() int {
+	m.instanceID++
+	return m.instanceID
+}
+
 // getEndpointConnections only for unit test
 func (m *GlobalServerManager) getEndpointConnections(endpoint string) *websocketEndpointContext {
 	m.RLock()
diff --git a/internal/io/websocket/websocket_sink.go b/internal/io/websocket/websocket_sink.go
new file mode 100644
index 00000000..613a21d5
--- /dev/null
+++ b/internal/io/websocket/websocket_sink.go
@@ -0,0 +1,83 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package websocket
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
+	"github.com/lf-edge/ekuiper/v2/internal/io/memory/pubsub"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+)
+
+type WebsocketSink struct {
+	cfg   *WebsocketConfig
+	props map[string]any
+	topic string
+}
+
+func (w *WebsocketSink) Provision(ctx api.StreamContext, configs map[string]any) error {
+	cfg := &WebsocketConfig{}
+	if err := cast.MapToStruct(configs, cfg); err != nil {
+		return err
+	}
+	if !strings.HasPrefix(cfg.Endpoint, "/") {
+		return fmt.Errorf("websocket endpoint should start with /")
+	}
+	w.cfg = cfg
+	w.props = configs
+	return nil
+}
+
+func (w *WebsocketSink) Close(ctx api.StreamContext) error {
+	pubsub.RemovePub(w.topic)
+	return connection.DetachConnection(ctx, buildWebsocketEpID(w.cfg.Endpoint), w.props)
+}
+
+func (w *WebsocketSink) Connect(ctx api.StreamContext) error {
+	conn, err := connection.FetchConnection(ctx, buildWebsocketEpID(w.cfg.Endpoint), "websocket", w.props)
+	if err != nil {
+		return err
+	}
+	c, ok := conn.(*httpserver.WebsocketConnection)
+	if !ok {
+		return fmt.Errorf("should use websocket connection")
+	}
+	w.topic = c.SendTopic
+	pubsub.CreatePub(w.topic)
+	return nil
+}
+
+func (w *WebsocketSink) Collect(ctx api.StreamContext, item api.RawTuple) error {
+	return w.collect(ctx, item.Raw())
+}
+
+func (w *WebsocketSink) collect(ctx api.StreamContext, data []byte) error {
+	pubsub.ProduceAny(ctx, w.topic, data)
+	return nil
+}
+
+func GetSink() api.Sink {
+	return &WebsocketSink{}
+}
+
+var _ api.BytesCollector = &WebsocketSink{}
+
+func buildWebsocketEpID(endpoint string) string {
+	return fmt.Sprintf("$$ws/%s", endpoint)
+}
diff --git a/internal/io/websocket/websocket_source.go b/internal/io/websocket/websocket_source.go
index e2f689cf..5dc6d2a1 100644
--- a/internal/io/websocket/websocket_source.go
+++ b/internal/io/websocket/websocket_source.go
@@ -53,11 +53,11 @@ func (w *WebsocketSource) Provision(ctx api.StreamContext, configs map[string]an
 
 func (w *WebsocketSource) Close(ctx api.StreamContext) error {
 	pubsub.CloseSourceConsumerChannel(w.topic, w.sourceID)
-	return connection.DetachConnection(ctx, w.cfg.Endpoint, w.props)
+	return connection.DetachConnection(ctx, buildWebsocketEpID(w.cfg.Endpoint), w.props)
 }
 
 func (w *WebsocketSource) Connect(ctx api.StreamContext) error {
-	conn, err := connection.FetchConnection(ctx, w.cfg.Endpoint, "websocket", w.props)
+	conn, err := connection.FetchConnection(ctx, buildWebsocketEpID(w.cfg.Endpoint), "websocket", w.props)
 	if err != nil {
 		return err
 	}


Test patch:
diff --git a/internal/io/http/httpserver/websocket_server_test.go b/internal/io/http/httpserver/websocket_server_test.go
index 5117350f..02f47afd 100644
--- a/internal/io/http/httpserver/websocket_server_test.go
+++ b/internal/io/http/httpserver/websocket_server_test.go
@@ -16,6 +16,7 @@ package httpserver
 
 import (
 	"testing"
+	"time"
 
 	"github.com/gorilla/websocket"
 	"github.com/stretchr/testify/require"
@@ -32,14 +33,16 @@ func TestWebsocketServerRecvData(t *testing.T) {
 	defer ShutDown()
 	ctx := mockContext.NewMockContext("1", "2")
 	endpint := "/e1"
-	recvTopic, err := RegisterWebSocketEndpoint(ctx, endpint)
+	rTopic, _, err := RegisterWebSocketEndpoint(ctx, endpint)
 	require.NoError(t, err)
-	subCh := pubsub.CreateSub(recvTopic, nil, "test", 1024)
-	defer pubsub.CloseSourceConsumerChannel(recvTopic, "test")
+	subCh := pubsub.CreateSub(rTopic, nil, "test", 1024)
+	defer pubsub.CloseSourceConsumerChannel(rTopic, "test")
 	conn, err := testx.CreateWebsocketClient(ip, port, endpint)
 	require.NoError(t, err)
 	defer conn.Close()
 	data := []byte("123")
+	// wait goroutine process started
+	time.Sleep(10 * time.Millisecond)
 	require.NoError(t, conn.WriteMessage(websocket.TextMessage, data))
 	recvData := <-subCh
 	require.Equal(t, data, recvData.([]byte))
@@ -53,7 +56,7 @@ func TestWebsocketServerRecvDataCancel(t *testing.T) {
 	defer ShutDown()
 	ctx := mockContext.NewMockContext("1", "2")
 	endpint := "/e1"
-	_, err := RegisterWebSocketEndpoint(ctx, endpint)
+	_, _, err := RegisterWebSocketEndpoint(ctx, endpint)
 	require.NoError(t, err)
 	UnRegisterWebSocketEndpoint(endpint)
 }
@@ -65,13 +68,47 @@ func TestWebsocketServerRecvDataOther(t *testing.T) {
 	defer ShutDown()
 	ctx := mockContext.NewMockContext("1", "2")
 	endpint := "/e1"
-	_, err := RegisterWebSocketEndpoint(ctx, endpint)
+	_, _, err := RegisterWebSocketEndpoint(ctx, endpint)
 	require.NoError(t, err)
 	conn, err := testx.CreateWebsocketClient(ip, port, endpint)
 	require.NoError(t, err)
+	defer conn.Close()
+	// wait goroutine process started
+	time.Sleep(10 * time.Millisecond)
 	require.NoError(t, conn.WriteMessage(websocket.PingMessage, []byte("123")))
 	require.NoError(t, conn.WriteMessage(websocket.CloseMessage, []byte("123")))
 	wctx := manager.getEndpointConnections(endpint)
 	wctx.wg.Wait()
 	require.Equal(t, 0, len(wctx.conns))
 }
+
+func TestWebsocketServerSendData(t *testing.T) {
+	endpoint := "/e1"
+	topic := sendTopic(endpoint)
+	pubsub.CreatePub(topic)
+	ip := "127.0.0.1"
+	port := 10085
+	InitGlobalServerManager(ip, port, nil)
+	defer ShutDown()
+	ctx := mockContext.NewMockContext("1", "2")
+	_, sTopic, err := RegisterWebSocketEndpoint(ctx, endpoint)
+	require.NoError(t, err)
+	require.Equal(t, topic, sTopic)
+	conn, err := testx.CreateWebsocketClient(ip, port, endpoint)
+	require.NoError(t, err)
+	defer conn.Close()
+	// wait goroutine process started
+	time.Sleep(10 * time.Millisecond)
+	assertCh := make(chan struct{})
+	go func() {
+		msgTyp, data, err := conn.ReadMessage()
+		require.NoError(t, err)
+		require.Equal(t, websocket.TextMessage, msgTyp)
+		require.Equal(t, []byte("123"), data)
+		assertCh <- struct{}{}
+	}()
+	time.Sleep(10 * time.Millisecond)
+	pubsub.ProduceAny(ctx, topic, []byte("123"))
+	<-assertCh
+	UnRegisterWebSocketEndpoint(endpoint)
+}
diff --git a/internal/io/websocket/websocket_sink_test.go b/internal/io/websocket/websocket_sink_test.go
new file mode 100644
index 00000000..55b9cd5e
--- /dev/null
+++ b/internal/io/websocket/websocket_sink_test.go
@@ -0,0 +1,68 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package websocket
+
+import (
+	"testing"
+	"time"
+
+	"github.com/gorilla/websocket"
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
+	"github.com/lf-edge/ekuiper/v2/internal/testx"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
+func init() {
+	modules.RegisterConnection("websocket", httpserver.CreateWebsocketConnection)
+}
+
+func TestWebsocketSink(t *testing.T) {
+	connection.InitConnectionManager4Test()
+	ip := "127.0.0.1"
+	port := 10081
+	endpoint := "/e1"
+	httpserver.InitGlobalServerManager(ip, port, nil)
+	defer httpserver.ShutDown()
+	ctx := mockContext.NewMockContext("1", "2")
+	ws := &WebsocketSink{}
+	props := map[string]any{
+		"datasource": endpoint,
+	}
+	require.Error(t, ws.Provision(ctx, map[string]any{
+		"datasource": "",
+	}))
+	require.NoError(t, ws.Provision(ctx, props))
+	require.NoError(t, ws.Connect(ctx))
+	expData := []byte("123")
+	assertCh := make(chan struct{})
+	conn, err := testx.CreateWebsocketClient(ip, port, endpoint)
+	require.NoError(t, err)
+	go func() {
+		msgTyp, data, err := conn.ReadMessage()
+		require.NoError(t, err)
+		require.Equal(t, websocket.TextMessage, msgTyp)
+		require.Equal(t, expData, data)
+		assertCh <- struct{}{}
+	}()
+	// wait goroutine start
+	time.Sleep(10 * time.Millisecond)
+	require.NoError(t, ws.collect(ctx, expData))
+	<-assertCh
+	ws.Close(ctx)
+}