Realtime Compute for Apache Flink:White paper de desempenho (benchmark Nexmark)

Etapa 1: Baixe a ferramenta de teste

Baixe e extraia o pacote da ferramenta de teste Nexmark nexmark-flink.tar.gz.

A estrutura de diretórios após a extração é a seguinte:

nexmark-flink/
├── run_nexmark.sh          # Test entry script
├── nexmark_env.sh          # Environment variable configuration file (requires editing)
├── bin/                    # Runtime scripts
├── conf/                   # Flink job configurations
├── lib/                    # JAR files (to be uploaded to the console)
└── queries-vvp/            # Nexmark Query SQL files

Etapa 2: Fazer upload do JAR Nexmark

1. Faça login no console do Realtime Compute for Apache Flink.

2. Clique em no espaço de projeto desejado. No painel de navegação à esquerda, selecione File Management > Upload Resource.

3. Selecione e faça upload do arquivo nexmark-flink-0.2-SNAPSHOT.jar. O arquivo está localizado no diretório nexmark-flink/lib da ferramenta de teste.

4. Após concluir o upload, clique em no nome do arquivo para copiar seu endereço OSS. Você precisará desse endereço em uma etapa posterior de configuração. O formato do caminho do arquivo varia conforme o tipo de armazenamento:

   • Armazenamento em OSS Bucket: oss://<nome do OSS Bucket>/artifacts/namespaces/<nome do espaço de projeto>/<nome do arquivo>

     Exemplo: oss://oss-test/artifacts/namespaces/flink-default/nexmark-flink-0.2-SNAPSHOT.jar

   • Armazenamento totalmente gerenciado: oss://flink-fullymanaged-<ID do workspace>/artifacts/namespaces/<nome do espaço de projeto>/<nome do arquivo>

     Exemplo: oss://flink-fullymanaged-e6a123456789/artifacts/namespaces/flink-default/nexmark-flink-0.2-SNAPSHOT.jar

   Para visualizar o tipo de armazenamento do seu workspace, acesse o console de gerenciamento do Realtime Compute for Apache Flink, localize o workspace desejado e clique em Details na coluna Actions.

Etapa 3: Configure parâmetros de execução

Edite o arquivo nexmark-flink/nexmark_env.sh e defina os seguintes parâmetros.

Etapa 4: Execute o teste

1. No diretório nexmark-flink, execute o seguinte comando.

   ./run_nexmark.sh

2. A ferramenta de teste cria e executa automaticamente os jobs Nexmark usando a OpenAPI.

3. Ao final do teste, a duração de cada consulta é exibida em milissegundos. O exemplo a seguir mostra uma saída de amostra:

   INFO  com.github.nexmark.flink.vvp.Nexmark - q0 13078
   ============================================================================
   ✓ Benchmark execution completed successfully
   ============================================================================

Preparação do ambiente

Crie um cluster Flink usando EMR on ECS com a seguinte configuração:

• Versão do EMR: EMR-5.21.0

• Especificação de hardware: Três instâncias ecs.g6a.xlarge (4 vCPU / 16 GiB), incluindo um nó mestre e dois nós principais.

• Ative os serviços Hadoop e HDFS.

• Configure o login sem senha entre todos os nós. Por exemplo, faça upload do seu arquivo de chave privada (como key.pem) para o nó mestre. Em seguida, adicione a seguinte configuração ao arquivo ~/.ssh/config no nó mestre. Substitua os endereços IP e o caminho do arquivo pelos seus valores reais.

  Host 192.168.0.0
      HostName 192.168.0.0
      User root
      IdentityFile /path/to/key.pem
      StrictHostKeyChecking no
  Host 192.168.0.1
      HostName 192.168.0.1
      User root
      IdentityFile /path/to/key.pem
      StrictHostKeyChecking no
  Host 192.168.0.2
      HostName 192.168.0.2
      User root
      IdentityFile /path/to/key.pem
      StrictHostKeyChecking no

  Use ssh para verificar se o login sem senha entre os nós está funcionando corretamente. Se ocorrer um erro de "bad permissions", execute chmod 600 /path/to/key.pem para corrigir as permissões.

Preparação do software

1. Baixe o pacote Flink desejado (Apache Flink Downloads) e o pacote de teste Nexmark (nexmark-flink.tgz). Faça upload dos pacotes para o nó mestre e extraia-os.

   tar -zxvf flink-1.20.4-bin-scala_2.12.tgz
   tar -zxvf nexmark-flink.tgz
   mv flink-1.20.4 flink
   mv nexmark-flink nexmark

1. Copie os arquivos JAR do diretório nexmark/lib para flink/lib. Esses JARs contêm o gerador de dados Nexmark.

   cp nexmark/lib/* flink/lib/

1. Defina as variáveis de ambiente. Edite ~/.bashrc, adicione a seguinte configuração e execute source ~/.bashrc para que as alterações entrem em vigor.

   Configure os caminhos com base no seu ambiente real.

   export JAVA_HOME=/etc/alternatives/java_sdk_11
   export PATH=$JAVA_HOME/bin:$PATHexport FLINK_HOME=/mnt/disk1/flink
   export HADOOP_CLASSPATH=$(/opt/apps/HADOOP-COMMON/hadoop-common-current/bin/hadoop classpath)

Configuração e inicialização do cluster

1. Configure os workers do Flink. Este teste usa oito TaskManagers, implantados da seguinte forma: dois no nó mestre e três em cada nó principal.

   Edite flink/conf/workers, certificando-se de substituir o endereço IP pelo valor real.

   192.168.0.0
   192.168.0.0
   192.168.0.1
   192.168.0.1
   192.168.0.1
   192.168.0.2
   192.168.0.2
   192.168.0.2

1. Substitua flink/conf/config.yaml por nexmark/conf/config.yaml e atualize os seguintes itens de configuração:

   • jobmanager.rpc.address: O endereço IP do nó mestre, como 192.168.0.0.

   • state.checkpoints.dir: O caminho HDFS, como hdfs:///checkpoints

   • taskmanager.memory.process.size: 4G

1. Edite nexmark/conf/nexmark.yaml e defina nexmark.metric.reporter.host como o endereço IP do nó mestre.

2. Distribua os diretórios flink e nexmark, bem como as configurações de variáveis de ambiente, para cada nó principal.

   Substitua os endereços IP pelos seus valores reais.

   scp -r flink 192.168.0.1:/mnt/disk1/
   scp -r flink 192.168.0.2:/mnt/disk1/
   scp -r nexmark 192.168.0.1:/mnt/disk1/
   scp -r nexmark 192.168.0.2:/mnt/disk1/
   scp ~/.bashrc 192.168.0.1:~/
   scp ~/.bashrc 192.168.0.2:~/

   Após concluir a distribuição, execute source ~/.bashrc em cada nó principal para que as variáveis de ambiente entrem em vigor.

1. No nó mestre, inicie o cluster Flink.

   flink/bin/start-cluster.sh

1. Inicialize o ambiente de teste Nexmark. Este script configura o Metric Reporter necessário em cada nó.

   nexmark/bin/setup_cluster.sh

Defina limites de recursos

Após a inicialização do cluster Flink, use cgroups para limitar o uso de CPU de cada processo TaskManager a 75%. Isso evita que os TaskManagers percam conexão devido a timeouts de heartbeat causados por contenção de recursos.

Execute os seguintes comandos em todos os nós onde os TaskManagers estão em execução, incluindo o nó mestre:

yum install -y libcgroup libcgroup-tools
cgcreate -t root:root -a root:root -g cpu,memory:mygroup
echo 100000 > /sys/fs/cgroup/cpu/mygroup/cpu.cfs_period_us
echo 300000 > /sys/fs/cgroup/cpu/mygroup/cpu.cfs_quota_us
echo $((12 * 1024 * 1024 * 1024)) > /sys/fs/cgroup/memory/mygroup/memory.limit_in_bytes
jps | grep TaskManagerRunner | awk '{print $1}' | xargs cgclassify -g cpu,memory:mygroup

Aqui, cpu.cfs_quota_us / cpu.cfs_period_us = 300000 / 100000 = 3, o que significa que o cgroup pode usar no máximo 3 núcleos de CPU (75% de 4 vCPUs).

Execute o Nexmark

No nó mestre, execute o seguinte comando e aguarde a conclusão para visualizar os resultados:

nexmark/bin/run_query.sh q0,q1,q2,q3,q4,q5,q7,q8,q9,q10,q11,q12,q14,q15,q16,q17,q18,q19,q20,q21,q22