ApsaraMQ for Kafka:Crie um conector sink do OSS

Pré-requisitos

Notas de uso

• O conector exporta dados de um tópico de origem em uma instância do ApsaraMQ for Kafka para o Function Compute e, em seguida, do Function Compute para o Object Storage Service (OSS). A instância do ApsaraMQ for Kafka, o service Function Compute e o bucket do OSS devem estar na mesma região. Para obter mais informações sobre as limitações do conector, consulte Limitations.

• Este recurso é baseado no service Function Compute. O Function Compute oferece uma cota gratuita. Se você exceder essa cota, será cobrado pelos recursos adicionais utilizados. Para obter mais informações sobre faturamento, consulte Billing overview.

• O Function Compute oferece suporte a consultas de log para invocações de função. Para obter mais informações, consulte Configure logs.

• Ao exportar mensagens, o ApsaraMQ for Kafka as serializa usando a codificação de string UTF-8. O sistema não oferece suporte a formatos de dados binários.

Crie e implante um conector sink do OSS

1. Faça logon no console do ApsaraMQ for Kafka.

2. Na página Overview, selecione uma região na seção Resource Distribution.

3. No painel de navegação à esquerda, clique em Connectors.

4. Na página Connectors, selecione a instância à qual o conector pertence na lista suspensa Select Instance e clique em Create Connector.

5. No assistente Create Connector, execute as seguintes etapas:

   1. Na etapa Configure Basic Information, configure os parâmetros e clique em Next. A tabela a seguir descreve os parâmetros.

      Importante

      O ApsaraMQ for Kafka seleciona automaticamente a opção Authorize to Create Service Linked Role.

      • Se uma função vinculada ao service ainda não tiver sido criada, o ApsaraMQ for Kafka criará uma automaticamente para você. Essa função permite que o ApsaraMQ for Kafka exporte dados para o OSS.

      • Se uma função vinculada ao service já existir, o ApsaraMQ for Kafka não criará uma nova.

      Para obter mais informações sobre essa função vinculada ao service, consulte Service-linked roles.

      Parâmetro	Descrição	Exemplo
      Name	O nome do conector. Deve atender aos seguintes requisitos: O nome deve ter no máximo 48 caracteres e pode conter apenas dígitos, letras minúsculas e hifens (-). Não pode começar com um hífen (-). O nome deve ser exclusivo na mesma instância do ApsaraMQ for Kafka. Uma tarefa de sincronização de dados para um Connector deve usar um Group chamado connect-nome da tarefa. Se você não crie o Group manualmente, o sistema o criará automaticamente.	kafka-oss-sink
      Instance	Por padrão, o nome e o ID da instância são exibidos.	demo alikafka_post-cn-st21p8vj****

   2. Na etapa Configure Source Service, selecione Message Queue for Apache Kafka como o service de origem, configure os parâmetros e clique em Next. A tabela a seguir descreve os parâmetros.

      Parâmetro	Descrição	Exemplo
      Data Source Topic	O tópico do qual os dados serão exportados.	oss-test-input
      Consumer Thread Concurrency	O número de threads de consumo simultâneas para o tópico de origem. Valor padrão: 6. Valores válidos: 1 2 3 6 12	6
      Consumer Offset	O offset a partir do qual o consumo começa. Valores válidos: Earliest Offset: o consumo começa a partir da primeira mensagem disponível na partição. Latest Offset: o consumo começa a partir da próxima mensagem produzida após a inicialização do conector.	Earliest Offset
      VPC ID	A VPC na qual a tarefa de sincronização de dados é executada. clique em Configure Runtime Environment para exibir este parâmetro. Por padrão, ele é definido como a VPC da instância do ApsaraMQ for Kafka e não requer configuração.	vpc-bp1xpdnd3l***
      vSwitch ID	O vSwitch para a tarefa de sincronização de dados. clique em Configure Runtime Environment para exibir este parâmetro. O vSwitch deve estar na mesma VPC da instância do ApsaraMQ for Kafka. Por padrão, é o vSwitch especificado quando a instância do ApsaraMQ for Kafka foi implantada.	vsw-bp1d2jgg81***
      Failure Handling Policy	Especifica a política para lidar com falhas na entrega de mensagens em uma partição de tópico. clique em Configure Runtime Environment para exibir este parâmetro. Valores válidos: Continue Subscription: mantém a assinatura da partição de tópico onde o erro ocorreu e registra um log de erro. Stop Subscription: cancela a assinatura da partição de tópico onde o erro ocorreu e registra um log de erro. Nota Para saber como visualizar logs, consulte Connector-related operations. Para saber como encontrar soluções com base em códigos de erro, consulte Error codes.	Continue Subscription
      Resource Creation Method	O método de criação dos tópicos e do grupo de consumidores exigidos pelo conector. clique em Configure Runtime Environment para exibir este parâmetro. Auto Manual	Auto
      Connector Consumer Group	O grupo de consumidores usado pelo conector. clique em Configure Runtime Environment para exibir este parâmetro. Recomendamos que o nome deste grupo de consumidores comece com connect-cluster.	connect-cluster-kafka-oss-sink
      Task Offset Topic	O tópico para armazenar offsets de consumo. clique em Configure Runtime Environment para exibir este parâmetro. Tópico: recomendamos que o nome do tópico comece com connect-offset. Número de partições: deve ser maior que 1. Motor de armazenamento: o motor de armazenamento do tópico deve ser Local storage. Nota Apenas instâncias da Professional Edition oferecem suporte à seleção de Local storage na criação de um tópico. As instâncias da Standard Edition não oferecem suporte a esse recurso. cleanup.policy: a política de limpeza de log para o tópico deve ser compact.	connect-offset-kafka-oss-sink
      Task Configuration Topic	O tópico para armazenar configurações de tarefas. clique em Configure Runtime Environment para exibir este parâmetro. Tópico: recomendamos que o nome do tópico comece com connect-config. Número de partições: deve ser 1. Motor de armazenamento: o motor de armazenamento do tópico deve ser Local storage. Nota Apenas instâncias da Professional Edition oferecem suporte à seleção de Local storage na criação de um tópico. As instâncias da Standard Edition não oferecem suporte a esse recurso. cleanup.policy: a política de limpeza de log para o tópico deve ser compact.	connect-config-kafka-oss-sink
      Task Status Topic	O tópico para armazenar o status da tarefa. clique em Configure Runtime Environment para exibir este parâmetro. Tópico: recomendamos que o nome do tópico comece com connect-status. Número de partições: recomendamos defina como 6. Motor de armazenamento: o motor de armazenamento do tópico deve ser Local storage. Nota Apenas instâncias da Professional Edition oferecem suporte à seleção de Local storage na criação de um tópico. As instâncias da Standard Edition não oferecem suporte a esse recurso. cleanup.policy: a política de limpeza de log para o tópico deve ser compact.	connect-status-kafka-oss-sink
      Dead-letter Queue Topic	O tópico para armazenar dados de exceção do framework Kafka Connect. clique em Configure Runtime Environment para exibir este parâmetro. Este tópico pode ser o mesmo que o Error Data Topic para economizar recursos de tópico. Tópico: recomendamos que o nome do tópico comece com connect-error. Número de partições: recomendamos defina como 6. Motor de armazenamento: o motor de armazenamento do tópico pode ser Local storage ou Cloud storage. Nota Apenas instâncias da Professional Edition oferecem suporte à seleção de Local storage na criação de um tópico. As instâncias da Standard Edition não oferecem suporte a esse recurso.	connect-error-kafka-oss-sink
      Error Data Topic	O tópico para armazenar dados de exceção do sink. clique em Configure Runtime Environment para exibir este parâmetro. Este tópico pode ser o mesmo que o Dead-letter Queue Topic para economizar recursos de tópico. Tópico: recomendamos que o nome do tópico comece com connect-error. Número de partições: recomendamos defina como 6. Motor de armazenamento: o motor de armazenamento do tópico pode ser Local storage ou Cloud storage. Nota Apenas instâncias da Professional Edition oferecem suporte à seleção de Local storage na criação de um tópico. As instâncias da Standard Edition não oferecem suporte a esse recurso.	connect-error-kafka-oss-sink

   3. Na etapa Configure Destination Service, selecione Object Storage Service como o service de destino, defina os parâmetros e clique em Create. A tabela a seguir descreve os parâmetros.

      Parâmetro	Descrição	Exemplo
      Bucket Name	O nome do bucket do OSS.	bucket_test
      AccessKey ID	O AccessKey ID da sua conta Alibaba Cloud.	yourAccessKeyID
      AccessKey Secret	O AccessKey Secret da sua conta Alibaba Cloud.	yourAccessKeySecret

      Certifique-se de que a sua conta Alibaba Cloud tenha as seguintes permissões concedidas de acordo com o princípio do menor privilégio:

      {
          "Version": "1",
          "Statement": [
              {
                  "Action": [
                      "oss:GetObject",
                      "oss:PutObject"
                  ],
                  "Resource": "*",
                  "Effect": "Allow"
              }
          ]
      }

      Nota

      O ApsaraMQ for Kafka passa o AccessKey ID e o AccessKey Secret para o OSS como variáveis de ambiente ao crie a tarefa. Após a criação, o ApsaraMQ for Kafka não armazena essas chaves.

      Após a criação do conector, você pode visualizá-lo na página Connectors.

6. acesse a página Connectors, encontre o conector que você criou e clique em Deploy na coluna Actions.

Enviar mensagens

Envie uma mensagem para o tópico de origem no ApsaraMQ for Kafka para verifique se os dados são exportados para o OSS.

1. Na página Connectors, encontre o conector de destino e clique em Test na coluna Actions.

2. No painel Send Message, envie uma mensagem de teste.

   • defina Sending Method como Console.

     1. Na caixa de texto Message Key, insira a chave da mensagem. Por exemplo, demo.

     2. Na caixa de texto Message Content, insira o conteúdo da mensagem de teste. Por exemplo, {"key": "test"}.

     3. defina Send to Specified Partition para especifique se a mensagem será enviada para uma partição específica.

        • clique em Yes e insira o ID da partição na caixa de texto Partition ID. Por exemplo, 0. Para consultar o ID da partição, consulte Visualize o status da partição.

        • clique em No para não especifique uma partição.

   • defina Sending Method como Docker. execute o comando Docker na seção Run the Docker container to produce a sample message para enviar a mensagem.

   • defina Sending Method como SDK. Selecione um SDK para a linguagem ou framework necessário e um tipo de conexão para enviar mensagens.

Verifique os resultados

Após enviar uma mensagem para o tópico de origem no ApsaraMQ for Kafka, verifique o gerenciador de arquivos do OSS para confirmar que os dados foram exportados. Para obter mais informações, consulte File overview.

Se novos objetos forem gerados no bucket do OSS, os dados serão sincronizados com o OSS.

[
    {
        "key":"123",
        "offset":4,
        "overflowFlag":true,
        "partition":0,
        "timestamp":1603779578478,
        "topic":"Test",
        "value":"1",
        "valueSize":272687
    }
]

Operações relacionadas

Configure os recursos do Function Compute exigidos pelo conector sink do OSS de acordo com suas necessidades.