MaxCompute:Executar jobs MapReduce no runtime SQL

Contexto

O MaxCompute oferece uma API MapReduce para escrever programas Java que processam dados na plataforma.

A versão mais recente do MaxCompute permite executar jobs MapReduce no modo de execução do runtime sql. Nesse modo, os jobs utilizam o compilador SQL do MaxCompute, o otimizador baseado em custo (CBO) e o mecanismo de execução vetorizada. Assim, beneficiam-se das melhorias contínuas do mecanismo SQL em recursos, desempenho e estabilidade.

Jobs MapReduce executados no modo de runtime sql aproveitam novos recursos do SQL do MaxCompute para realizar operações antes incompatíveis. Em comparação com o mecanismo clássico do MapReduce, o modo de execução do runtime sql adiciona as seguintes capacidades:

• Suporte a views como source de entrada.

• Compatibilidade com tabelas externas como source de entrada.

• Operações de leitura e gravação em sistemas de arquivos distribuídos.

• Leitura e escrita em tabelas com cluster hash ou range.

Este modo também proporciona os seguintes benefícios:

• Aproveitamento das otimizações contínuas de desempenho do CBO e do mecanismo de execução vetorizada do SQL.

• Uso de novos mecanismos de compressão de formato de armazenamento.

• Ajuste dinâmico de paralelismo para melhorar o desempenho em cenários com junção de tabelas muito grandes, como tabelas com cluster hash.

• Utilização da estabilidade comprovada do mecanismo SQL, validada por extensa execução de jobs e testes de estresse. Isso aumenta a confiabilidade de mecanismos como failover e persistent volume claim (PVC).

• Obtenção de insights mais profundos com o MaxCompute Studio e o LogView. Essas ferramentas fornecem informações detalhadas sobre a execução de jobs SQL, incluindo plano de execução, dados de compilação e configurações do job. A visão detalhada da E/S em cada etapa e do fluxo de trabalho geral ajuda a identificar problemas, otimizar o desempenho e aumentar a eficiência de desenvolvimento e O&M.

Observações

• Não é necessário alterar APIs existentes ou a lógica dos jobs. Basta especificar o modo de execução.

• Esse recurso suporta apenas jobs MapReduce escritos com a API MapReduce. Para mais informações, consulte SDK Overview.

• Jobs MapReduce executados no modo de runtime sql seguem as regras padrão de faturamento do MapReduce. Para mais detalhes, consulte MapReduce pay-as-you-go.

Instruções

1. Defina o modo de execução.

   Controle o modo de execução por meio da propriedade odps.mr.run.mode. Os valores válidos são:

   • lot (Padrão): O job executa no mecanismo MapReduce.

   • sql: O job executa no mecanismo SQL. Se falhar, o sistema retorna um erro.

   • hybrid: O sistema tenta primeiro executar o job no mecanismo SQL. Caso a tentativa falhe, o job reverte automaticamente para o mecanismo MapReduce.

   Configure o modo de execução de uma das seguintes formas:

   Controle no nível do projeto

   Para ativar esse modo em todos os jobs de um projeto, o administrador deve executar o seguinte comando:

   setproject odps.mr.run.mode=<lot/sql/hybrid>;

   Controle no nível da sessão

   Para ative esse modo apenas no job atual, utilize um dos métodos abaixo:

   • Adicione a instrução set odps.mr.run.mode=<lot/sql/hybrid> antes do comando JAR.

   • Defina a propriedade no código do job usando um objeto JobConf, conforme o exemplo a seguir:

     JobConf job = new JobConf();
     job.set("odps.mr.run.mode","hybrid")

   Nota

   Em cenários específicos, como StreamJob e SecondarySort, defina as seguintes flags:

   • StreamJob: set odps.mr.sql.stream.enable=true;

   • SecondarySort: set odps.mr.sql.group.enable=true;

2. Visualize os detalhes do job.

   Use ferramentas como LogView e MaxCompute Studio para inspecionar as expressões SQL geradas no cliente e visualize os detalhes de execução do job. Para saber mais sobre o uso do LogView, consulte Use LogView V2.0 to view job information.

   • XML do LogView

     Abra o LogView e acesse a aba Source XML para visualize as informações XML enviadas pelo cliente. Esses dados mostram o equivalente SQL do job MapReduce. O código abaixo apresenta um exemplo:

     create temporary function mr2sql_mapper_152955927079392291755 as   'com.aliyun.odps.mapred.bridge.LotMapperUDTF' using ; 
     create temporary function mr2sql_reducer_152955927079392291755 as 'com.aliyun.odps.mapred.bridge.LotReducerUDTF' using ; 
     
     @sub_query_mapper :=
     SELECT k_id,v_gmt_create,v_gmt_modified,v_product_id,v_admin_seq,v_sku_attr,v_sku_price,v_sku_stock,v_sku_code,v_sku_image,v_delivery_time,v_sku_bulk_order,v_sku_bulk_discount,v_sku_image_version,v_currency_code
     FROM(
       SELECT mr2sql_mapper_152955927079392291755(id,gmt_create,gmt_modified,product_id,admin_seq,sku_attr,sku_price,sku_stock,sku_code,sku_image,delivery_time,sku_bulk_order,sku_bulk_discount,sku_image_version,currency_code ) as (k_id,v_gmt_create,v_gmt_modified,v_product_id,v_admin_seq,v_sku_attr,v_sku_price,v_sku_stock,v_sku_code,v_sku_image,v_delivery_time,v_sku_bulk_order,v_sku_bulk_discount,v_sku_image_version,v_currency_code)
       FROM ae_antispam.product_sku_tt_inc
       WHERE ds = "20180615"  AND hh = "21"                     
       UNION ALL
       SELECT mr2sql_mapper_152955927079392291755(id,gmt_create,gmt_modified,product_id,admin_seq,sku_attr,sku_price,sku_stock,sku_code,sku_image,delivery_time,sku_bulk_order,sku_bulk_discount,sku_image_version,currency_code ) as (k_id,v_gmt_create,v_gmt_modified,v_product_id,v_admin_seq,v_sku_attr,v_sku_price,v_sku_stock,v_sku_code,v_sku_image,v_delivery_time,v_sku_bulk_order,v_sku_bulk_discount,v_sku_image_version,v_currency_code)
       FROM ae_antispam.product_sku
     ) open_mr_alias1
     DISTRIBUTE BY k_id SORT BY k_id ASC;
     
     @sub_query_reducer := 
     SELECT mr2sql_reducer_152955927079392291755(k_id,v_gmt_create,v_gmt_modified,v_product_id,v_admin_seq,v_sku_attr,v_sku_price,v_sku_stock,v_sku_code,v_sku_image,v_delivery_time,v_sku_bulk_order,v_sku_bulk_discount,v_sku_image_version,v_currency_code) as (id,gmt_create,gmt_modified,product_id,admin_seq,sku_attr,sku_price,sku_stock,sku_code,sku_image,delivery_time,sku_bulk_order,sku_bulk_discount,sku_image_version,currency_code)
     FROM @sub_query_mapper;
     FROM @sub_query_reducer	
     INSERT OVERWRITE TABLE ae_antispam.product_sku
     SELECT id,gmt_create,gmt_modified,product_id,admin_seq,sku_attr,sku_price,sku_stock,sku_code,sku_image,delivery_time,sku_bulk_order,sku_bulk_discount,sku_image_version,currency_code ;

   • Resumo do LogView

     Na aba Summary do LogView, confirme que o job execute usando o execution engine SQL. Veja um exemplo no código a seguir:

     Nota

     Para jobs MapReduce fora do modo de execução do runtime sql, nenhuma informação sobre o mecanismo de execução é exibida. Já nos jobs extended MapReduce (MR2) do MaxCompute fora do modo de runtime sql, o mecanismo de execução aparece como cganjiang.

      Job run mode: fuxi job
      Job run engine: execution engine

   • JSON Summary do LogView

     O JSON Summary de um job MapReduce padrão contém apenas informações básicas de entrada e saída das tarefas Map e Reduce. Por outro lado, o JSON Summary de um job baseado em SQL traz dados detalhados de cada fase de execução, incluindo todos os parâmetros, o plano lógico, o plano físico e os detalhes da execução. O trecho abaixo ilustra esse formato:

      "midlots" : 
      [
      "LogicalTableSink(table=[[odps_flighting.flt_20180621104445_step1_ad_quality_tech_qp_algo_antifake_wordbag_filter_bag_change_result_lv2_20, auctionid,word,match_word(3) {0, 1, 2}]])
     OdpsLogicalProject(auctionid=[$0], word=[$1], match_word=[$2])
     OdpsLogicalProject(auctionid=[$0], word=[$1], match_word=[$2])
     OdpsLogicalProject(auctionid=[$0], word=[$1], match_word=[$2])
     OdpsLogicalProject(auctionid=[$2], word=[$3], match_word=[$4])
     OdpsLogicalTableFunctionScan(invocation=[[MR2SQL_MAPPER_152955294118813063732($0, $1)]()], rowType=[RecordType(VARCHAR(2147483647) item_id, VARCHAR(2147483647) text, VARCHAR(2147483647) __tf_0_0, VARCHAR(2147483647) __tf_0_1, VARCHAR(2147483647) __tf_0_2)])
     OdpsLogicalTableScan(table=[[ad_quality_tech.qp_algo_antifake_wordbag_filter_bag_change_lv2_20, item_id,text(2) {0, 1}]])
     ]