Object Storage Service:Transmita dados do EMR Kafka para o OSS-HDFS com o Flume

Pré-requisitos

• O serviço OSS-HDFS está ativado e você tem as permissões de acesso necessárias. Para mais informações, consulte Ativar o serviço OSS-HDFS.

• Você criou um cluster DataLake e selecionou o serviço Flume. Para mais informações, consulte Criar um cluster.

• Você criou um cluster DataFlow e selecionou o serviço Kafka. Para mais informações, consulte Criar um cluster.

Procedimento

1. Configure o Flume.

   1. Acesse a página de configuração do Flume.

      1. Faça login no console do E-MapReduce.

      2. Na barra de navegação superior, selecione uma região e um grupo de recursos.

      3. Na página EMR on ECS, localize o cluster desejado e clique em Services na coluna Actions.

      4. Na aba Services, clique em Configure na área do serviço FLUME.

   2. Defina o tamanho máximo do heap da Java Virtual Machine (JVM) (Xmx).

      Gravar dados do Flume no serviço OSS-HDFS consome muita memória. Recomendamos aumentar o valor de Xmx do agente do Flume.

      1. Clique em flume-env.sh.

         Este tópico usa a configuração global. Para configurar um nó específico, selecione Independent Node Configuration na lista suspensa da página Configure do serviço FLUME.

      2. Modifique o valor do parâmetro JAVA_OPTS.

         Por exemplo, para definir o tamanho máximo do heap da JVM como 1 GB, altere o valor do parâmetro para -Xmx1g.

      3. Clique em Salve.

   3. Modifique a configuração do arquivo flume-conf.properties.

      1. Clique em flume-conf.properties.

         Este tópico usa a configuração global. Para configurar um nó específico, selecione Independent Node Configuration na lista suspensa da página Configure do serviço FLUME.

      2. No editor do arquivo flume-conf.properties, insira a seguinte configuração.

         Nota

         O valor de default-agent no exemplo abaixo deve corresponder ao valor do parâmetro agent_name na página Configure do serviço FLUME.

         default-agent.sources = source1
         default-agent.sinks = k1
         default-agent.channels = c1
         
         default-agent.sources.source1.type = org.apache.flume.source.kafka.KafkaSource
         default-agent.sources.source1.channels = c1
         default-agent.sources.source1.kafka.bootstrap.servers = <kafka-host1:port1,kafka-host2:port2...>
         default-agent.sources.source1.kafka.topics = flume-test
         default-agent.sources.source1.kafka.consumer.group.id = flume-test-group
         
         default-agent.sinks.k1.type = hdfs
         default-agent.sinks.k1.hdfs.path = oss://{yourBucketName}.{yourBucketRegion}.oss-dls.aliyuncs.com/{path}
         default-agent.sinks.k1.hdfs.fileType=DataStream
         
         # Use a memory channel to buffer events
         default-agent.channels.c1.type = memory
         default-agent.channels.c1.capacity = 100
         default-agent.channels.c1.transactionCapacity = 100
         
         # Bind the source and sink to the channel
         default-agent.sources.source1.channels = c1
         default-agent.sinks.k1.channel = c1

         Parâmetro	Descrição
         default-agent.sources.source1.kafka.bootstrap.servers	Nomes de host e números de porta dos brokers no cluster Kafka.
         default-agent.sinks.k1.hdfs.path	O caminho do OSS-HDFS deve seguir o formato oss://{yourBucketName}.{yourBucketRegion}.oss-dls.aliyuncs.com/{path}. Exemplo: oss://flume-test.cn-hangzhou.oss-dls.aliyuncs.com/result. Componentes do caminho: {yourBucketName}: Insira o nome do bucket com o serviço OSS-HDFS ativado. {yourBucketRegion}: Insira o ID da região do bucket. {path}: Insira o nome do diretório do serviço OSS-HDFS.
         default-agent.channels.c1.capacity	Número máximo de eventos que o canal pode armazenar. Ajuste esse valor conforme o ambiente.
         default-agent.channels.c1.transactionCapacity	Número máximo de eventos que uma source pode gravar ou um sink pode recuperar do canal por transação. Ajuste esse valor conforme o ambiente.

      3. Clique em Salve.

2. Teste a sincronização de dados.

   1. Conecte-se ao cluster EMR Dataflow via SSH. Para mais informações, consulte Fazer login em um cluster.

   2. Crie um tópico chamado flume-test.

      kafka-topics.sh --partitions 10 --replication-factor 2 --zookeeper master-1-1:2181/emr-kafka --topic flume-test --create

   3. Gere dados de teste.

      kafka-console-producer.sh --topic flume-test --broker-list master-1-1:9092

      Por exemplo, insira abc e pressione Enter.

      Um arquivo é gerado no caminho oss://flume-test.cn-hangzhou.oss-dls.aliyuncs.com/result. O nome do arquivo segue o formato FlumeData.xxxx, com o timestamp atual em milissegundos como sufixo.