DocsICP FilingConsole
官方文档
首页

Stream data to OSS

更新时间：
复制 MD 格式
产品详情
我的收藏

This topic describes how to create an OSS sink connector to stream data from a source topic in ApsaraMQ for Kafka to objects in Object Storage Service (OSS).

Prerequisites

For more information, see Prerequisites.

Usage notes

  1. The connector uses the event processing time for time-based partitioning, not the event creation time. For time-based partitioning, data near a time boundary may be delivered to the subdirectory for the next time partition.

  2. Dirty data handling: If you configure a JsonPath expression for a custom partition or for the file content, but a message does not match the expression, the connector routes this data to the invalidRuleData/ path in the bucket based on the batching policy. If you find this directory in your bucket, check the accuracy of your JsonPath expression and ensure that consumers do not miss any data.

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

  4. If the JsonPath expression configured for a custom partition or file content needs to extract data from the Kafka source message, you must encode the message content in JSON format on the Kafka source side.

  5. The connector appends upstream data to OSS in real time. Therefore, the latest file in a partition path is usually being written to and is not in a final state. Consume this data with caution.

Billing

Connector tasks run on the Alibaba Cloud Function Compute platform. You pay for the computing resources consumed for data processing and transmission based on the unit price of Function Compute. For more information, see Billing overview.

Step 1: Create a destination bucket

Create a bucket in the Object Storage Service console. For detailed instructions, see Create buckets.

This topic uses a bucket named oss-sink-connector-bucket as an example.

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.

    • Create task

      1. In the Source step, set Data Provider to ApsaraMQ for Kafka, configure the following parameters, and then click Next.

        Parameter

        Description

        Example

        Region

        The region where the source ApsaraMQ for Kafka instance is located.

        China (Beijing)

        ApsaraMQ for Kafka Instance

        The source ApsaraMQ for Kafka instance that produces messages.

        alikafka_post-cn-jte3****

        Topic

        Select the topic for producing ApsaraMQ for Kafka messages.

        topic

        Group ID

        The consumer group ID for the source ApsaraMQ for Kafka instance.

        • Quick Create: Creates a new group ID.

        • Use Existing: Select an existing group ID.

        GID_http_1

        Consumer Offset

        The position from which to start consuming messages.

        Latest

        Network Configuration

        The network type used to route messages.

        Classic Network

        VPC

        The ID of the VPC. This parameter is required only if Network Configuration is set to VPC.

        vpc-bp17fapfdj0dwzjkd****

        vSwitches

        The ID of the vSwitch. This parameter is required only if Network Configuration is set to VPC.

        vsw-bp1gbjhj53hdjdkg****

        Security Group

        The security group. This parameter is required only if Network Configuration is set to VPC.

        alikafka_pre-cn-7mz2****

        Messages

        The maximum number of messages to send per function invocation. A request is sent only when the number of accumulated messages reaches this value. Valid values: 1 to 10,000.

        100

        Interval (Unit: Seconds)

        The interval between function invocations. The system aggregates and sends messages to Function Compute at this interval. Valid values: 0 to 15 seconds. A value of 0 indicates that messages are delivered immediately.

        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, set Service Type to OSS, and configure the following parameters.

        Parameter

        Description

        Example

        OSS bucket

        The destination OSS bucket.

        Important

        • Make sure the specified bucket exists and is not deleted while the task is running.

        • You can deliver data only to buckets in the Standard or Infrequent Access (IA) storage classes. You cannot deliver data to buckets in the Archive storage class.

        • After you create an OSS sink connector task, the platform creates a .tmp/ system directory in the root of the OSS bucket. Do not delete this directory or use the objects within it.

        oss-sink-connector-bucket

        Storage Path

        An OSS object key consists of a path and a name. For example, if the object key is a/b/c/a.txt, the path is a/b/c/ and the name is a.txt. You can customize the path for partitioning. The name is automatically generated by the connector with the format: {unix_timestamp_in_milliseconds}_{8_character_random_string}. Example: 1705576353794_elJmxu3v.

        • If this parameter is not configured or is set to /, data is not partitioned and is saved to the root directory of the bucket.

        • Time variables are supported: {yyyy}, {MM}, {dd}, and {HH} represent the year, month, day, and hour, respectively. These variables are case-sensitive.

        • JsonPath expressions are supported to customize path parameters, such as {$.data.topic} and {$.data.partition}. The expressions must be valid. Due to OSS path restrictions, we recommend extracting values as an int or string. The extracted values must be standard UTF-8 characters and must not contain spaces, .., emojis, /, or \ to avoid data write errors.

        • Constants are supported.

          Note

          Partitions allow you to group data logically and prevent issues caused by an excessive number of small files in a single path.

          Connector throughput scales with the number of partitions. Too few partitions, or no partitions, can reduce connector throughput and cause upstream backlogs. Too many partitions can lead to data fragmentation, increased write operations, and numerous small files. Therefore, your partitioning strategy is crucial. Consider the following recommendations:

          • Kafka Source: Partition by both time and Kafka partition ID. If performance is insufficient, increase the number of Kafka partitions to improve connector throughput. Example: prefix/{yyyy}/{MM}/{dd}/{HH}/{$.data.partition}/

          • Business-based grouping: Partition by a business field in the data. Throughput depends on the number of distinct values for that field. Example: prefixV2/{$.data.body.field}/

          We recommend using different constant prefixes for different tasks. This prevents multiple tasks from sharing partitions and writing to the same directory, which can mix data and complicate management.

        alikafka_post-cn-9dhsaassdd****/guide-oss-sink-topic/YYYY/MM/dd/HH

        Time Zone

        The default time zone is UTC+8:00. This setting applies only to time-based partitioning.

        UTC+8:00

        Batch aggregation object size

        The target size for aggregating data into a single object. Valid values: 1 to 1,024 MB.

        Note

        • The connector writes data to an OSS object in batches. The size of each batch is in the range of (0 MB, 16 MB]. Therefore, the final object size may be slightly larger than the configured value, by up to 16 MB.

        • In high-traffic scenarios, we recommend setting the aggregation object size to hundreds of megabytes (for example, 128 MB or 512 MB) and the aggregation time window to the hour level (for example, 60 min or 120 min).

        5

        Batch aggregation time window

        The time window for aggregating data. Valid values: 1 to 1,440 minutes.

        1

        File Compression

        • None: The generated OSS object has no file extension.

        • GZIP: The generated object has a .gz extension.

        • Snappy: The generated object has a .snappy extension.

        • Zstd: The generated object has a .zstd extension.

        When compression is enabled, the connector batches data based on the pre-compression size. As a result, the object size displayed in OSS is smaller than the configured batch size, and the decompressed size is close to the batch size.

        None

        File Content

        • Full Event: The connector wraps the original message in the CloudEvents envelope. The resulting data includes the CloudEvents metadata. In the following example, the data field contains the original message, and the other fields are metadata from the CloudEvents envelope.

          {
  "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"
  }
}

        • Partial Event: Delivers only part of the data, extracted by a JsonPath expression. For example, if you configure $.data, only the value of the data field is delivered to OSS.

        To exclude CloudEvents metadata, select Partial Event and use the $.data expression. This delivers only the original source message to OSS, which saves storage costs and improves transmission efficiency.

        Partial Event

        $.data

    • Task properties

      Configure the retry policy for failed event deliveries and the error handling method. For more information, see Retry policies and dead-letter queues.

  4. Return to the Tasks page, find your task, and click Enable in the Actions column.

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

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

Step 3: Verify the OSS sink connector

  1. On the Message Outflow page, in the row for your OSS sink connector task, click the source topic name in the Event Sources column.

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

  3. In the Quickly experience message sending and consumption panel, configure the message as follows and click OK.

    Set Message key to oss-sink-k2, set Message content to oss-sink-v2, and for Send to specified partition, select No.

  4. On the Message Outflow page, in the row for your OSS sink connector task, click the destination bucket name in the Event Targets column.

  5. On the Bucket page, in the left navigation bar, select File Management > File List.

    • tmp/ directory: A system directory required by the connector. Do not delete or use the objects in this directory.

    • Data directories: Subdirectories are generated based on the partition path rule configured for the task. Data files are uploaded to the deepest-level directory.

    The path structure of the data directory is similar to alikafka_p[TopicName]/[PartitionID]/yyyy/MM/dd/HH/. In the deepest-level directory, you can find a metadata file (such as .oss_meta_file...) and partition data files (such as partition_3_of...). Finding these files confirms that the OSS sink connector has successfully written messages to OSS.

  6. In the Actions column for 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"}

    Messages are separated by line breaks.

Previous：Route events to Simple Log ServiceNext：Route events to Tablestore