MaxCompute:Desenvolvimento e depuração

Procedimento:

1. Crie um projeto Java chamado graph_examples.

2. Adicione os pacotes JAR do diretório lib do cliente MaxCompute ao Java Build Path do projeto Eclipse. Na aba Libraries da página Java Build Path, selecione mapreduce-api.jar, expanda-o e clique duas vezes em Javadoc location. Selecione Javadoc URL e insira http://odps.alibaba-inc.com/doc/prdoc/odps_graph/api/.

3. Desenvolva o programa MaxCompute Graph.

   Uma prática comum é copiar e modificar um exemplo existente, como o algoritmo Single Source Shortest Path. Neste exemplo, apenas o caminho do pacote foi alterado para package com.aliyun.odps.graph.example.

4. Compile e empacote o programa.

   No Eclipse, clique com o botão direito no diretório de source (diretório src na figura) e selecione Export > Java > JAR file para gerar um pacote JAR. Escolha um caminho de destino para o pacote JAR, como D:\\odps\\clt\\odps-graph-example-sssp.jar.

5. Use o cliente MaxCompute para executar o job SSSP. Para mais informações, consulte Run a Graph job.

Procedimento:

1. Baixe o pacote Maven odps-graph-local.

2. No projeto Eclipse, clique com o botão direito no arquivo do programa principal do job Graph (arquivo que contém a função main) e selecione Run As > Run Configurations.

3. Na aba Arguments, defina Program arguments como 1 sssp_in sssp_out.

4. Ainda na aba Arguments, configure os VM arguments conforme abaixo.

   -Dodps.runner.mode=local
   -Dodps.project.name=<project.name>
   -Dodps.end.point=<end.point>
   -Dodps.access.id=<access.id> 
   -Dodps.access.key=<access.key>

   Defina também os parâmetros do programa em Program arguments. Por exemplo, use 1 sssp_in sssp_out, que representam, respectivamente, o nó inicial, o nome da tabela de entrada e o nome da tabela de saída.

5. No modo local, em que o parâmetro odps.end.point não é especificado, crie as tabelas sssp_in e sssp_out no diretório warehouse e adicione dados à tabela de entrada sssp_in. O código a seguir mostra um exemplo de dados de entrada.

   1,"2:2,3:1,4:4"
   2,"1:2,3:2,4:1"
   3,"1:1,2:2,5:1"
   4,"1:4,2:1,5:1"
   5,"3:1,4:1"

   Para mais informações sobre o diretório warehouse, consulte Run jobs locally.

6. Clique em Run para executar o job SSSP localmente.

   Nota

   Para configurações de parâmetros, consulte o arquivo conf/odps_config.ini no cliente MaxCompute. Os parâmetros anteriores são de uso comum. A lista a seguir descreve esses parâmetros:

   • odps.runner.mode: Defina o valor como local. Este parâmetro é obrigatório para depuração local.

   • odps.project.name: Especifica o projeto atual. Parâmetro obrigatório.

   • odps.end.point: Especifica o endpoint do service MaxCompute. Parâmetro opcional. Se omitido, tabelas e recursos serão lidos do diretório warehouse local. Uma exceção será lançada se os dados ou metadados não existirem. Caso especifique este parâmetro, o sistema tentará primeiro ler do warehouse local. Se os dados ou metadados solicitados não existirem, o sistema recorrerá à leitura diretamente do MaxCompute.

   • odps.access.id: AccessKey ID para conexão ao service MaxCompute. Válido apenas quando odps.end.point for especificado.

   • odps.access.key: AccessKey Secret para conexão ao service MaxCompute. Válido apenas quando odps.end.point for especificado.

   • odps.cache.resources: Especifica a lista de recursos a serem utilizados. Equivale à opção -resources no comando JAR.

   • odps.local.warehouse: Caminho para o diretório warehouse local. Se não especificado, o caminho padrão será ./warehouse.

   Abaixo está a saída de depuração da execução local do job SSSP no Eclipse.

   Counters: 3
            com.aliyun.odps.graph.local.COUNTER
                    TASK_INPUT_BYTE=211
                    TASK_INPUT_RECORD=5
                    TASK_OUTPUT_BYTE=161
                    TASK_OUTPUT_RECORD=5
    graph task finish

   Nota

   No exemplo anterior, o diretório warehouse local deve conter as tabelas sssp_in e sssp_out. Para mais informações sobre as tabelas sssp_in e sssp_out, consulte Write a Graph program.

Sempre que uma sessão de depuração local é executada, um diretório temporário é criado no diretório do projeto Eclipse. Um exemplo de nome de diretório temporário é graph_20130816154834_240_5772. Ele contém os seguintes subdiretórios e arquivos: counters (informações de contadores), inputs (dados de entrada, como zhemin_test1.sssp_in), outputs (resultados de saída, incluindo o subdiretório _default_), resources (arquivos de recursos, incluindo o subdiretório centers), superSteps (informações de superstep) e o arquivo de configuração de job job.xml.

O diretório temporário de um job Graph executado localmente inclui os seguintes diretórios e arquivos:

• counters: Contém informações de contadores geradas durante a execução do job.

• inputs: Armazena os dados de entrada do job. O sistema tenta primeiramente recuperar dados do warehouse local. Se os dados não forem encontrados e o parâmetro odps.end.point estiver definido, o sistema usa o SDK do MaxCompute para ler os dados do servidor. Por padrão, são lidos no máximo 10 registros para cada input. Altere esse limite usando o parâmetro -Dodps.mapred.local.record.limit, mas o valor não pode exceder 10.000.

• outputswarehouse: Contém os dados de saída do job. Após a conclusão do job, os dados resultantes deste diretório sobrescrevem a tabela correspondente no warehouse local.

• resources: Armazena os recursos utilizados pelo job. Assim como nas entradas, o sistema tenta primeiro recuperar recursos do warehouse local. Se os recursos não forem encontrados, o sistema usa o SDK do MaxCompute para lê-los do servidor, desde que o parâmetro odps.end.point esteja definido.

• job.xml: Contém a configuração do job.

• superstep: Armazena informações de persistência para cada iteração.

Procedimento:

1. Configure o cliente MaxCompute.

2. Use o comando add jar /path/work.jar -f; para atualizar o pacote JAR.

3. Execute o job usando o comando JAR e verifique o log de execução e os dados resultantes.

Os seguintes itens de configuração de job Graph afetam o desempenho:

• setSplitSize(long): Tamanho da divisão para a tabela de entrada. Unidade: MB. O valor deve ser maior que 0. Valor padrão: 64.

• setNumWorkers(int): Número de workers para o job. Valores válidos: [1, 1000]. Valor padrão: 1. O número ideal de workers depende do tamanho de entrada do job em bytes e do splitSize.

• setWorkerCPU(int): Recursos de CPU para cada worker. Um valor de 100 representa um núcleo de CPU. Valores válidos: [50, 800]. Valor padrão: 200.

• setWorkerMemory(int): Recursos de memória para cada worker. Unidade: MB. Valores válidos: [256, 12288]. Valor padrão: 4096.

• setMaxIteration(int): Número máximo de iterações. Valor padrão: -1. Um valor menor ou igual a 0 indica que o número máximo de iterações não é uma condição de término do job.

• setJobPriority(int): Prioridade do job. Valores válidos: [0, 9]. Valor padrão: 9. Um valor maior indica uma prioridade menor.

Recomendações gerais para otimização:

• Considere aumentar o número de workers usando o método setNumWorkers.

• Avalie reduzir o tamanho da divisão com o método setSplitSize para acelerar o carregamento de dados.

• Incremente os recursos de CPU ou memória para cada worker.

• Defina o número máximo de iterações. Para aplicações que não exigem alta precisão, reduza o número de iterações para encerrar o job mais rapidamente.

As APIs setNumWorkers e setSplitSize podem ser usadas em conjunto para melhorar a velocidade de carregamento de dados. Suponha que setNumWorkers esteja definido como workerNum, setSplitSize esteja definido como splitSize e o tamanho total de entrada em bytes seja inputSize. O número de divisões de entrada é calculado como splitNum=inputSize/splitSize. A relação entre workerNum e splitNum é a seguinte:

• Cenário 1: Se splitNum for igual a workerNum, cada worker carregará uma divisão.

• Cenário 2: Se splitNum for maior que workerNum, cada worker carregará uma ou mais divisões.

• Cenário 3: Se splitNum for menor que workerNum, cada worker carregará zero ou uma divisão.

Outro problema comum de desempenho é o data skew (desequilíbrio de dados). Na saída dos contadores, isso aparece como alguns workers processando significativamente mais vértices ou arestas do que outros. O data skew geralmente ocorre quando poucas chaves correspondem a um número desproporcionalmente grande de vértices, arestas ou mensagens. Essas chaves são atribuídas a um pequeno número de workers, fazendo com que esses workers tenham tempos de execução mais longos. Para resolver esse problema, use um dos métodos a seguir:

• Use um Combiner para agregar mensagens para essas chaves localmente, o que reduz o número de mensagens enviadas.

  Um Combiner reduz o uso de memória para armazenamento de mensagens e diminui o tráfego de rede, encurtando o tempo total de execução do job.

• Otimize sua lógica de negócios.

  Para grandes conjuntos de dados, a E/S de disco pode dominar o tempo de processamento. Reduzir o volume de dados melhora o throughput e o desempenho:

  • Reduza o volume de dados de entrada: Para certas aplicações de tomada de decisão, processar um subconjunto amostrado dos dados pode afetar apenas a precisão do resultado, e não sua exatidão geral. Nesses casos, considere amostrar os dados antes de importá-los para a tabela de entrada.

  • Evite ler campos desnecessários: A classe TableInfo no framework MaxCompute Graph permite ler colunas específicas, passadas como um array de nomes de colunas, em vez de toda a tabela ou partição. Isso também reduz o volume de dados de entrada e melhora o desempenho do job.

Os seguintes pacotes JAR são carregados na JVM por padrão quando você executa um programa Graph. Não é necessário enviar esses recursos ou incluí-los com a opção -libjars no comando.

• commons-codec-1,3.jar

• commons-io-2.0.1.jar

• commons-lang-2,5.jar

• commons-logging-1.0.4.jar

• commons-logging-api-1.0.4.jar

• guava-14,0.jar

• json.jar

• log4j-1.2.15.jar

• slf4j-api-1,4,3.jar

• slf4j-log4j12-1,4,3.jar

• xmlenc-0,52.jar