MaxCompute:Execute jobs MapReduce no runtime SQL

Contexto

O MaxCompute disponibiliza uma API MapReduce para crie programas Java que processam dados no MaxCompute.

A versão mais recente do MaxCompute permite que você execute jobs MapReduce no modo de execução do runtime SQL. Nesse modo, os jobs MapReduce podem usar o compilador SQL do MaxCompute, o otimizador baseado em custo (CBO) e o mecanismo de execução vetorizada, beneficiando-se das melhorias contínuas feitas no mecanismo SQL em termos de recursos, desempenho e estabilidade.

Os jobs MapReduce executados no modo de execução do runtime SQL podem usar os novos recursos do SQL do MaxCompute para realizar operações que antes não eram suportadas. Em comparação com o mecanismo MapReduce clássico, o modo de execução do runtime SQL adiciona os seguintes recursos:

• Suporte a views como source de entrada.

• Suporte a tabelas externas como source de entrada.

• Suporte a operações de leitura e gravação em um sistema de arquivos distribuído.

• Suporte a operações de leitura e gravação em tabelas agrupadas por hash ou intervalo.

Esse modo também oferece os seguintes benefícios:

• Aproveite as otimizações contínuas de desempenho do otimizador baseado em custo (CBO) e do mecanismo de execução vetorizada do SQL.

• Use os novos mecanismos de compressão de formato de armazenamento.

• Ajuste dinamicamente o paralelismo para melhorar o desempenho em cenários que unem tabelas muito grandes, como tabelas agrupadas por hash.

• Aproveite a estabilidade comprovada do mecanismo SQL, estabelecida por meio de extensa execução de jobs e testes de estresse. Isso melhora a confiabilidade de mecanismos como failover e persistent volume claim (PVC).

• Obtenha insights mais profundos ao usar o MaxCompute Studio e o LogView. Essas ferramentas fornecem informações detalhadas de execução para jobs SQL, como o plano de execução, informações de compilação e configurações do job. Essa visão detalhada da E/S em cada estágio e do fluxo de trabalho geral ajuda você a identificar problemas, otimizar o desempenho e melhorar a eficiência de desenvolvimento e O&M.

Observações

• Este recurso não requer alterações nas suas APIs ou na lógica do job existentes. Apenas especifique o modo de execução.

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

• Os jobs MapReduce executados no modo de execução do runtime SQL ainda são cobrados de acordo com as regras de faturamento padrão do MapReduce. Para mais informações, consulte MapReduce pay-as-you-go.

Instruções

1. Defina o modo de execução.

   A propriedade odps.mr.run.mode controla o modo de execução. Os valores válidos são:

   • lot (Padrão): O job é executado no mecanismo de execução MapReduce.

   • sql: O job é executado no mecanismo de execução SQL. Se o job falhar, um erro será retornado.

   • hybrid: O sistema primeiro execute o job no mecanismo de execução SQL. Se a tentativa falhar, o job passa a ser executado no mecanismo de execução MapReduce.

   Selecione e defina o modo de execução de uma das seguintes maneiras:

   Controle no nível do projeto

   Como administrador do projeto, ative esse modo para todos os jobs execute o seguinte comando:

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

   Controle no nível da sessão

   Ative esse modo apenas para o job atual utilizando 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 seu job usando um objeto JobConf, conforme o exemplo a seguir:

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

   Nota

   Para cenários especiais como StreamJob e SecondarySort, defina os seguintes sinalizadores:

   • 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 mais informações sobre o uso do LogView, consulte Use LogView V2.0 to view job information.

   • LogView XML

     Abra o LogView e acesse a aba Source XML para que você visualize as informações XML enviadas pelo cliente. Essas informações mostram o equivalente em 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 ;

   • LogView Summary

     Na aba Summary do LogView, é possível verificar que o job foi executado com o execution engine SQL. O código abaixo apresenta um exemplo:

     Nota

     Para jobs MapReduce que não são executados no modo de execução do runtime SQL, nenhuma informação do mecanismo de execução é exibida. Para jobs extended MapReduce (MR2) do MaxCompute que não são executados no modo de execução do runtime SQL, o mecanismo de execução é o cganjiang.

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

   • LogView JSON Summary

     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. Já o JSON Summary de um job baseado em SQL traz informações detalhadas 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 código abaixo apresenta um exemplo:

      "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}]])
     ]