ApsaraMQ for Kafka:Create an OSS sink connector

Usage notes

1. The connector partitions data based on the event processing time, not the event creation time. If you use time-based partitioning, data near a time boundary may be delivered to the directory for the next time window.

2. Dirty data handling: If you configure a JSONPath expression for the custom partition path or file content, but the incoming data does not match the rule, the connector delivers this dirty data to the invalidRuleData/ directory in the bucket according to the batching policy. If you find this directory in your bucket, verify your JSONPath expression and ensure that your consumer is not missing data.

3. End-to-end latency can range from seconds to minutes.

4. If the JSONPath rule configured for the custom partition path or file content needs to extract data from the message body of the Kafka source, you must encode the content into JSON format at the source.

5. The connector writes data from the upstream source to OSS in real time by appending to an object. Therefore, in a single partition path, the latest visible object is usually still being written, not in its final state. Consume this data with caution.

Billing

Connector tasks run on Function Compute. The compute resources consumed for data processing and transmission are billed based on Function Compute pricing. For more information, see Billing overview.

Step 1: Create target service resources

Create a bucket in the OSS console. For more information, see Create a bucket in the console.

This example uses a bucket named oss-sink-connector-bucket.

Step 2: Create and start the OSS sink connector

1. Log on to the ApsaraMQ for Kafka console. In the Resource Distribution section of the Overview page, select the region where the ApsaraMQ for Kafka instance that you want to manage resides.

2. In the left-side navigation pane, choose Connector Ecosystem Integration > Tasks.

3. On the Tasks page, click Create Task.

   • Task Creation

     1. In the Source step, select Message Queue for Apache Kafka as the Data Provider. Configure the following parameters, and then click Next.

        Parameter	Description	Example
        Region	The region of the source Message Queue for Apache Kafka instance.	China (Beijing)
        Kafka instance	The source Message Queue for Apache Kafka instance.	alikafka_post-cn-jte3****
        Topic	Select the topic for producing Message Queue for Apache Kafka messages.	demo-topic
        Group ID	The consumer group of the source instance. Quick Create: This is the recommended option. A group ID that is named in the GID_EVENTBRIDGE_xxx format is automatically created. Use Existing: Select an independent group ID. Do not use a group ID that is shared with other services to avoid disrupting existing message consumption.	Quick Create
        Consumer offset	The offset from which message consumption starts. Latest offset (latest) Earliest offset (earliest)	Latest offset (latest)
        Network configuration	The network type for routing messages. Basic Network Self-managed Internet	Basic Network
        VPC	Required only if Network configuration is set to Self-managed Internet.	vpc-bp17fapfdj0dwzjkd****
        vSwitch	Required only if Network configuration is set to Self-managed Internet.	vsw-bp1gbjhj53hdjdkg****
        Security group	Required only if Network configuration is set to Self-managed Internet.	alikafka_pre-cn-7mz2****
        Data Format	The encoding format for the message content. For data sources that support binary transmission, select one of the following formats. We recommend Json if you do not have specific encoding requirements. Json: Encodes binary data as a JSON object into the payload using UTF-8. Text: Encodes binary data as a UTF-8 string into the payload. This is the default format. Binary: Encodes binary data as a Base64-encoded string into the payload.	Json
        Messages	Advanced configuration parameter. The maximum number of messages to send per batch. A request is sent only when the number of accumulated messages reaches the specified value. Valid values: 1 to 10,000.	100
        Interval (Unit: Seconds)	Advanced configuration parameter. The interval for invoking a function. The system aggregates messages and sends them to the sink at each interval. Valid values: 0 to 15. A value of 0 specifies that messages are delivered immediately without a delay.	3

     2. In the Filtering step, set the Pattern Content to filter events. For more information, see event pattern.

     3. In the Transformation step, configure data transformation to perform operations such as splitting, mapping, enrichment, and dynamic routing. For more information, see Use Function Compute to clean message data.

     4. In the Sink step, select Object Storage Service as the Service Type, configure the following parameters, and then click Save.

        Parameter	Description	Example
        OSS bucket	The OSS bucket that you created. Important Ensure the specified bucket exists and is not deleted while the task is running. The bucket's storage class must be Standard or Infrequent Access (IA). Archive storage is not supported. After you create an OSS sink connector task, the system generates a .tmp/ system file path in the root directory of the bucket. Do not delete this path or use the OSS objects within it.	oss-sink-connector-bucket
        Storage Path	a/b/c/a.txta/b/c/a.txt{millisecond Unix timestamp}_{8-character random string}An OSS object key consists of a path and a name. For example, if the object key is , the path is and the name is . You can customize the partition path. The name is automatically generated by the connector in the format: , for example, . If you leave this blank or set it to /, no partitioning is applied, and data is saved to the root directory of the bucket. Supports time variables: {yyyy}, {MM}, {dd}, and {HH} represent year, month, day, and hour, respectively. These variables are case-sensitive. Supports JSONPath expressions to customize the path, for example, {$.data.topic} and {$.data.partition}. The JSONPath variables must be valid JSONPath expressions. Due to OSS path constraints, values extracted using JSONPath should be of type int or string. The values must contain only standard UTF-8 characters and must not include spaces, .., emojis, /, or \. Otherwise, data write exceptions may occur. Supports constants. Note Partitioning helps group data logically and prevents performance issues caused by an excessive number of small objects in a single path. The connector's throughput is positively correlated with the number of partitions. Low or no partitioning can result in low throughput and cause data backlogs in the source. Too many partitions can lead to data fragmentation, increased write operations, and an excessive number of small objects. Therefore, a proper partitioning strategy is crucial. Consider the following recommendations: Kafka Source: Supports partitioning by both time and partition. If performance is insufficient, you can increase the number of Kafka partitions to indirectly increase the connector's throughput. For example: prefix/{yyyy}/{MM}/{dd}/{HH}/{$.data.partition}/ Business Grouping: Partitions data by a specific business field. The throughput rate is then determined by the number of unique values of this field. For example: prefixV2/{$.data.body.field}/ We recommend using different constant prefixes for different tasks to avoid multiple tasks writing to the same directory, which can cause data confusion.	alikafka_post-cn-9dhsaassdd****/guide-oss-sink-topic/yyyy/MM/dd/HH
        Batch aggregation object size	The target size for object aggregation. The value is in MB. Valid values: 1 to 1,024. Note The connector writes data in batches of up to 16 MB. As a result, the final object size may exceed the configured value by up to 16 MB. For high-traffic scenarios, we recommend setting the batch aggregation object size to hundreds of megabytes (for example, 128 MB or 512 MB) and the time window to the hourly level (for example, 60 minutes or 120 minutes).	5
        Batch aggregation time window	The time window for aggregation. The value is in minutes. Valid values: 1 to 1,440.	1
        File Compression	No Compression Required: Generates an OSS object with no file extension. GZIP: Generates an object with the .gz extension. Snappy: Generates an object with the .snappy extension. Zstd: Generates an object with the .zstd extension. If you select a compression option, the connector batches data based on the pre-compression size. As a result, the object size in OSS will be smaller than the configured Batch aggregation object size. After decompression, the size will be close to the configured value.	No Compression Required
        File Content	Complete Data: The connector wraps the original message in the CloudEvents protocol. Complete data includes the data with the CloudEvents protocol wrapper. In the following example, the data field contains the message data, and other fields are metadata added by the CloudEvents protocol. { "specversion": "1.0", "id": "8e215af8-ca18-4249-8645-f96c1026****", "source": "acs:alikafka", "type": "alikafka:Topic:Message", "subject": "acs:alikafka:alikafka_pre-cn-i7m2msb9****:topic:****", "datacontenttype": "application/json; charset=utf-8", "time": "2022-06-23T02:49:51.589Z", "aliyunaccountid": "182572506381****", "data": { "topic": "****", "partition": 7, "offset": 25, "timestamp": 1655952591589, "headers": { "headers": [], "isReadOnly": false }, "key": "keytest", "value": "hello kafka msg" } } Data Extraction: Delivers only a part of the data extracted using a JSONPath expression. For example, if you specify $.data, only the value of the data field is delivered to OSS. To save storage costs and improve efficiency, use Data Extraction with the $.data expression. This delivers only the original source message to OSS, omitting the CloudEvents wrapper.	Data Extraction $.data

   • Task properties

     Configure the retry policy for failed event deliveries and how to handle errors. For more information, see Retries and dead-letter queues.

4. Return to the Tasks page. Find the task you created and click Enable in the Actions column.

5. In the Note dialog box, read the message and then click OK.

   The task takes 30 to 60 seconds to start. You can monitor the progress in the Status column on the Tasks page.

Step 3: Test the OSS sink connector

1. On the Tasks page, click the source topic in the Event Source column for your task.

2. On the topic details page, click Send Test Message.

3. In the Start to Send and Consume Message panel, configure the message content as follows, and then click OK.

   Select the Console tab. In the Message Key field, enter oss-sink-k2. In the Message Content field, enter oss-sink-v2. For Send to Specified Partition, select No.

4. On the Tasks page, click the target bucket in the Event Target column for your task.

5. On the bucket page, in the left-side navigation pane, choose Object Management > Objects.

   • tmp directory: This is a system path that the connector depends on. Do not delete or use OSS objects in this path.

   • Data file directory: Subdirectories are generated based on the task's partition path rule, and data objects are uploaded to the deepest directory.

   In this example, the breadcrumb path is /alikafka_<topic>/<partition>/2023/04/18/02/. This shows that subdirectories are created based on the topic name, partition number, and a year/month/day/hour structure. The deepest directory contains data objects such as .oss_meta_file and partition_3_of....

6. In the Actions column to the right of the object, choose > Download.

7. Open the downloaded file to view the message content.

   {"topic":"guide-oss-sink-topic","partition":0,"offset":0,"timestamp":1681378474218,"headers":{"headers":[],"isReadOnly":false},"key":"oss-sink-k2","value":"oss-sink-v2"}
   {"topic":"guide-oss-sink-topic","partition":0,"offset":1,"timestamp":1681378491498,"headers":{"headers":[],"isReadOnly":false},"key":"oss-sink-k2","value":"oss-sink-v2"}
   {"topic":"guide-oss-sink-topic","partition":0,"offset":2,"timestamp":1681378492515,"headers":{"headers":[],"isReadOnly":false},"key":"oss-sink-k2","value":"oss-sink-v2"}

   The output contains multiple messages, with each message formatted as a JSON object on a new line.