Todos os produtos
Search

Este produto

    ApsaraMQ for Kafka:Crie um conector sink do OSS

    Central de documentação

    ApsaraMQ for Kafka:Crie um conector sink do OSS

    Última atualização: Sep 21, 2026

    Crie um conector sink do OSS para exportar dados de um tópico de origem no ApsaraMQ for Kafka para um objeto no Object Storage Service (OSS).

    Pré-requisitos

    Para mais informações, consulte Prerequisites.

    Observações de uso

    1. O conector particiona os dados com base no tempo de processamento do evento, e não no tempo de criação. Ao usar o particionamento por tempo, dados próximos a um limite de tempo podem ser entregues no diretório da próxima janela de tempo.

    2. Tratamento de dados inválidos: ao configurar uma expressão JSONPath para o caminho de partição personalizado ou para o conteúdo do arquivo, se os dados recebidos não corresponderem à regra, o conector entregará esses dados inválidos no diretório invalidRuleData/ do bucket, de acordo com a política de lotes. Caso encontre esse diretório no seu bucket, verifique a expressão JSONPath e certifique-se de que o consumidor não esteja perdendo dados.

    3. A latência de ponta a ponta pode variar de segundos a minutos.

    4. Caso a regra JSONPath configurada para o caminho de partição personalizado ou para o conteúdo do arquivo precise extrair dados do corpo da mensagem da origem Kafka, codifique o conteúdo no formato JSON na origem.

    5. O conector grava dados no OSS em tempo real acrescentando-os a um objeto. Em um mesmo caminho de partição, o objeto visível mais recente ainda pode estar recebendo gravações e não está em seu estado final. Tenha cautela ao consumir esses dados.

    Faturamento

    As tarefas do conector são executadas no Function Compute. O processamento e a transmissão de dados são cobrados de acordo com as tarifas do Function Compute. Para mais informações, consulte Billing overview.

    Etapa 1: Crie os recursos do serviço de destino

    Crie um bucket no console do OSS. Para mais informações, consulte Create a bucket in the console.

    Este exemplo usa um bucket chamado oss-sink-connector-bucket.

    Etapa 2: Crie e inicie o conector sink do OSS

    1. Acesse o console do ApsaraMQ for Kafka. Na página Overview, selecione uma região na seção Resource Distribution.

    2. No painel de navegação à esquerda, escolha Connector Ecosystem Integration > Tasks.

    3. Na página Tasks, clique em Create Task.

      • Criação de tarefa

        1. Na etapa Source, selecione Message Queue for Apache Kafka como o Data Provider. Configure os parâmetros a seguir e clique em Next.

          Parâmetro

          Descrição

          Exemplo

          Região

          A região da instância de origem do Message Queue for Apache Kafka.

          China (Beijing)

          Instância do Kafka

          A instância de origem do Message Queue for Apache Kafka.

          alikafka_post-cn-jte3****

          Tópico

          O tópico a partir do qual as mensagens são consumidas.

          demo-topic

          Group ID

          O grupo de consumidores da instância de origem.

          • Quick Create (recomendado): um Group ID no formato GID_EVENTBRIDGE_xxx é criado automaticamente.

          • Use Existing: selecione um Group ID existente. Não compartilhe este Group ID com outros serviços para evitar interrupções no consumo de mensagens.

          Quick Create

          Offset do consumidor

          O offset a partir do qual o consumo de mensagens é iniciado.

          • Latest offset (latest)

          • Earliest offset (earliest)

          Latest offset (latest)

          Configuração de rede

          O tipo de rede para roteamento de mensagens.

          • Basic Network

          • Self-managed Internet

          Basic Network

          VPC

          Necessário apenas se a Configuração de rede estiver definida como Self-managed Internet.

          vpc-bp17fapfdj0dwzjkd****

          vSwitch

          Necessário apenas se a Configuração de rede estiver definida como Self-managed Internet.

          vsw-bp1gbjhj53hdjdkg****

          Grupo de segurança

          Necessário apenas se a Configuração de rede estiver definida como Self-managed Internet.

          alikafka_pre-cn-7mz2****

          Data Format

          O formato de codificação para o conteúdo da mensagem. Recomendamos Json caso não haja requisitos específicos de codificação.

          • Json: codifica dados binários como um objeto JSON no payload usando UTF-8.

          • Text: codifica dados binários como uma string UTF-8 no payload. Este é o formato padrão.

          • Binary: codifica dados binários como uma string codificada em Base64 no payload.

          Json

          Messages

          Parâmetro de Advanced configuration. O número máximo de mensagens por lote. O sistema envia uma solicitação quando a contagem acumulada de mensagens atinge esse valor. Valores válidos: 1 a 10.000.

          100

          Interval (Unit: Seconds)

          Parâmetro de Advanced configuration. O intervalo, em segundos, no qual as mensagens são agregadas e enviadas para o sink. Valores válidos: 0 a 15. O valor 0 indica a entrega imediata das mensagens.

          3

        2. Na etapa Filtering, defina o Pattern Content para filtrar eventos. Para mais informações, consulte event pattern.

        3. Na etapa Transformation, configure a transformação de dados para executar operações como divisão, mapeamento, enriquecimento e roteamento dinâmico. Para mais informações, consulte Use Function Compute to clean message data.

        4. Na etapa Sink, selecione Object Storage Service como o Service Type, configure os parâmetros a seguir e clique em Save.

          Parâmetro

          Descrição

          Exemplo

          Bucket do OSS

          O bucket do OSS criado.

          Importante

          • Certifique-se de que o bucket especificado exista e não seja excluído durante a execução da tarefa.

          • A classe de armazenamento do bucket deve ser Standard ou Infrequent Access (IA). Não há suporte para armazenamento de arquivamento.

          • Após a criação de uma tarefa de conector sink do OSS, o sistema gera um caminho de arquivo de sistema .tmp/ no diretório raiz do bucket. Não exclua esse caminho nem use os objetos do OSS contidos nele.

          oss-sink-connector-bucket

          Storage Path

          a/b/c/a.txta/b/c/a.txt{millisecond Unix timestamp}_{8-character random string}A chave de um objeto do OSS é composta por um caminho e um nome. Por exemplo, se a chave do objeto for a/b/c/a.txt, o caminho é a/b/c/ e o nome é a.txt. É possível personalizar o caminho de partição. O nome é gerado automaticamente pelo conector no formato: {millisecond Unix timestamp}_{8-character random string}.

          • Ao deixar este campo em branco ou defini-lo como /, o particionamento não é aplicado e os dados são salvos no diretório raiz do bucket.

          • Há suporte a variáveis de tempo: {yyyy}, {MM}, {dd} e {HH} representam ano, mês, dia e hora, respectivamente. Essas variáveis diferenciam letras maiúsculas de minúsculas.

          • Há suporte a expressões JSONPath para personalizar o caminho, por exemplo, {$.data.topic} e {$.data.partition}. As variáveis JSONPath devem ser expressões JSONPath válidas. Devido às restrições de caminho do OSS, os valores extraídos com JSONPath devem ser do tipo int ou string. Esses valores devem conter apenas caracteres UTF-8 padrão e não podem incluir espaços, .., emojis, / ou \. Caso contrário, podem ocorrer exceções de gravação de dados.

          • Há suporte a constantes.

            Nota

            O particionamento ajuda a agrupar dados de forma lógica e evita problemas de desempenho causados pelo excesso de objetos pequenos em um único caminho.

            O throughput do conector escala de acordo com o número de partições. Poucas partições podem causar baixo throughput e acúmulo de dados. Muitas partições podem levar à fragmentação de dados, ao aumento das operações de gravação e ao excesso de objetos pequenos. Considere as seguintes recomendações:

            • Kafka Source: há suporte a particionamento por tempo e por partição. Para aumentar o throughput, adicione mais partições do Kafka. Por exemplo: prefix/{yyyy}/{MM}/{dd}/{HH}/{$.data.partition}/

            • Agrupamento de negócios: particiona os dados por um campo de negócio específico. O throughput é determinado pelo número de valores exclusivos desse campo. Por exemplo: prefixV2/{$.data.body.field}/

            Use prefixos constantes diferentes para tarefas distintas a fim de evitar que várias tarefas gravem no mesmo diretório, o que pode causar confusão de dados.

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

          Batch aggregation object size

          O tamanho de destino para a agregação de objetos, em MB. Valores válidos: 1 a 1.024.

          Nota

          • O conector grava dados em lotes de até 16 MB. Como resultado, o tamanho final do objeto pode exceder o valor configurado em até 16 MB.

          • Para cenários de alto tráfego, recomendamos definir o tamanho do objeto de agregação de lotes na casa das centenas de megabytes (por exemplo, 128 MB ou 512 MB) e a janela de tempo no nível de horas (por exemplo, 60 ou 120 minutos).

          5

          Batch aggregation time window

          A janela de tempo para agregação, em minutos. Valores válidos: 1 a 1.440.

          1

          File Compression

          • No Compression Required: gera um objeto do OSS sem extensão de arquivo.

          • GZIP: gera um objeto com a extensão .gz.

          • Snappy: gera um objeto com a extensão .snappy.

          • Zstd: gera um objeto com a extensão .zstd.

          Ao selecionar uma opção de compactação, o conector agrupa os dados em lotes com base no tamanho pré-compactação. Assim, o tamanho do objeto no OSS será menor que o Batch aggregation object size configurado. Após a descompactação, o tamanho se aproxima do valor configurado.

          No Compression Required

          File Content

          • Complete Data: o conector encapsula a mensagem original no protocolo CloudEvents. No exemplo a seguir, o campo data contém os dados da mensagem, e os demais campos são metadados adicionados pelo protocolo CloudEvents.

            {
      "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: entrega apenas uma parte dos dados extraídos usando uma expressão JSONPath. Por exemplo, ao especificar $.data, apenas o valor do campo data é entregue ao OSS.

          Para reduzir os custos de armazenamento e melhorar a eficiência, use Data Extraction com a expressão $.data para entregar apenas a mensagem de origem original ao OSS, sem o encapsulamento do CloudEvents.

          Data Extraction

          $.data

      • Propriedades da tarefa

        Configure a política de nova tentativa para entregas de eventos com falha e a forma de lidar com erros. Para mais informações, consulte Retries and dead-letter queues.

    4. Retorne à página Tasks. Encontre a tarefa criada e clique em Enable na coluna Actions.

    5. Na caixa de diálogo Note, leia a mensagem e clique em OK.

      A tarefa leva de 30 a 60 segundos para iniciar. Monitore o progresso na coluna Status da página Tasks.

    Etapa 3: Teste o conector sink do OSS

    1. Na página Tasks, clique no tópico de origem na coluna Event Source da tarefa.

    2. Na página de detalhes do tópico, clique em Send Test Message.

    3. No painel Start to Send and Consume Message, configure o conteúdo da mensagem da seguinte forma e clique em OK.

      Selecione a guia Console. No campo Message Key, insira oss-sink-k2. No campo Message Content, insira oss-sink-v2. Para Send to Specified Partition, selecione No.

    4. Na página Tasks, clique no bucket de destino na coluna Event Target da tarefa.

    5. Na página do bucket, no painel de navegação à esquerda, escolha Object Management > Objects.

      • Diretório tmp: caminho de sistema do qual o conector depende. Não exclua nem use objetos do OSS neste caminho.

      • Diretório de arquivos de dados: os subdiretórios são gerados com base na regra de caminho de partição da tarefa, e os objetos de dados são carregados no diretório mais profundo.

      Neste exemplo, o caminho é /alikafka_<topic>/<partition>/2023/04/18/02/. Isso mostra que os subdiretórios são criados com base no nome do tópico, no número da partição e em uma estrutura de ano/mês/dia/hora. O diretório mais profundo contém objetos de dados como .oss_meta_file e partition_3_of....

    6. Na coluna Actions à direita do objeto, escolha 图标 > Download.

    7. Abra o arquivo baixado para visualizar o conteúdo da mensagem.

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

      A saída contém várias mensagens, cada uma formatada como um objeto JSON em uma linha distinta.