RocketMQ sink connector

Support Apache RocketMQ Version

  • 4.9.0 (Or a newer version, for reference)

Support These Engines

Spark
Flink
SeaTunnel Zeta

Key Features

By default, we will use 2pc to guarantee the message is sent to RocketMQ exactly once.

Description

Write Rows to a Apache RocketMQ topic.

Sink Options

Name Type Required Default Description
topic string yes - RocketMQ topic name.
name.srv.addr string yes - RocketMQ name server cluster address.
acl.enabled Boolean no false false
access.key String no When ACL_ENABLED is true, access key cannot be empty
secret.key String no When ACL_ENABLED is true, secret key cannot be empty
producer.group String no SeaTunnel-producer-Group SeaTunnel-producer-Group
partition.key.fields array no - -
format String no json Data format. The default format is json. Optional text format. The default field separator is “,”.If you customize the delimiter, add the “field_delimiter” option.
field.delimiter String no , Customize the field delimiter for data format.
producer.send.sync Boolean no false If true, the message will be sync sent.
common-options config no - Sink plugin common parameters, please refer to Sink Common Options for details.

partition.key.fields [array]

Configure which fields are used as the key of the RocketMQ message.

For example, if you want to use value of fields from upstream data as key, you can assign field names to this property.

Upstream data is the following:

name age data
Jack 16 data-example1
Mary 23 data-example2

If name is set as the key, then the hash value of the name column will determine which partition the message is sent to.

Task Example

Fake to Rocketmq Simple

The data is randomly generated and asynchronously sent to the test topic

  1. env {
  2.   parallelism = 1
  3. }
  5. source {
  6.   FakeSource {
  7.     schema = {
  8.       fields {
  9.         c_map = "map<string, string>"
  10.         c_array = "array<int>"
  11.         c_string = string
  12.         c_boolean = boolean
  13.         c_tinyint = tinyint
  14.         c_smallint = smallint
  15.         c_int = int
  16.         c_bigint = bigint
  17.         c_float = float
  18.         c_double = double
  19.         c_decimal = "decimal(30, 8)"
  20.         c_bytes = bytes
  21.         c_date = date
  22.         c_timestamp = timestamp
  23.       }
  24.     }
  25.   }
  26. }
  28. transform {
  29.   # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
  30.   # please go to https://seatunnel.apache.org/docs/category/transform
  31. }
  33. sink {
  34.   Rocketmq {
  35.     name.srv.addr = "localhost:9876"
  36.     topic = "test_topic"
  37.   }
  38. }

Rocketmq To Rocketmq Simple

Consuming Rocketmq writes to c_int field Hash number of partitions written to different partitions This is the default asynchronous way to write

  1. env {
  2.   parallelism = 1
  3. }
  5. source {
  6.   Rocketmq {
  7.     name.srv.addr = "localhost:9876"
  8.     topics = "test_topic"
  9.     result_table_name = "rocketmq_table"
  10.     schema = {
  11.       fields {
  12.         c_map = "map<string, string>"
  13.         c_array = "array<int>"
  14.         c_string = string
  15.         c_boolean = boolean
  16.         c_tinyint = tinyint
  17.         c_smallint = smallint
  18.         c_int = int
  19.         c_bigint = bigint
  20.         c_float = float
  21.         c_double = double
  22.         c_decimal = "decimal(30, 8)"
  23.         c_bytes = bytes
  24.         c_date = date
  25.         c_timestamp = timestamp
  26.       }
  27.     }
  28.   }
  29. }
  31. sink {
  32.   Rocketmq {
  33.     name.srv.addr = "localhost:9876"
  34.     topic = "test_topic_sink"
  35.     partition.key.fields = ["c_int"]
  36.   }
  37. }

Timestamp consumption write Simple

This is a stream consumption specified time stamp consumption, when there are new partitions added the program will refresh the perception and consumption at intervals, and write to another topic type

  2. env {
  3.   parallelism = 1
  4.   job.mode = "STREAMING"
  5. }
  7. source {
  8.   Rocketmq {
  9.     name.srv.addr = "localhost:9876"
  10.     topics = "test_topic"
  11.     result_table_name = "rocketmq_table"
  12.     start.mode = "CONSUME_FROM_FIRST_OFFSET"
  13.     batch.size = "400"
  14.     consumer.group = "test_topic_group"
  15.     format = "json"
  16.     format = json
  17.     schema = {
  18.       fields {
  19.         c_map = "map<string, string>"
  20.         c_array = "array<int>"
  21.         c_string = string
  22.         c_boolean = boolean
  23.         c_tinyint = tinyint
  24.         c_smallint = smallint
  25.         c_int = int
  26.         c_bigint = bigint
  27.         c_float = float
  28.         c_double = double
  29.         c_decimal = "decimal(30, 8)"
  30.         c_bytes = bytes
  31.         c_date = date
  32.         c_timestamp = timestamp
  33.       }
  34.     }
  35.   }
  36. }
  38. transform {
  39.   # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
  40.   # please go to https://seatunnel.apache.org/docs/category/transform
  41. }
  42. sink {
  43.   Rocketmq {
  44.     name.srv.addr = "localhost:9876"
  45.     topic = "test_topic"
  46.     partition.key.fields = ["c_int"]
  47.     producer.send.sync = true
  48.   }
  49. }