DataWorks:Melhores práticas do RestAPI Reader

Contexto

Prática 1: Ler dados de uma API que consulta informações em um intervalo de tempo

Cenário de exemplo: Definição da API

Preparação: Criar uma tabela particionada no MaxCompute

CREATE TABLE IF NOT EXISTS ods_xiaobo_rest2
(
  `axis`  STRING
  ,`value` BIGINT
  ,`createTime` STRING
)
PARTITIONED BY
(
  ds  STRING
)
LIFECYCLE 3650;

Configurar a tarefa de sincronização

1. Adicione uma fonte de dados RestAPI.

   Adicione uma fonte de dados RestAPI no seu workspace do DataWorks. Para mais informações, consulte Adicionar uma fonte de dados RestAPI. Na caixa de diálogo Add RestAPI Data Source, especifique Data Source Name e Data Source Description, selecione Applicable Environment (desenvolvimento ou produção) e configure Default Request Header (valor padrão: {}). As configurações principais são as seguintes.

   • url: Defina como o endereço da API RESTful.

   • Authentication Method: Escolha um método de autenticação suportado pela API da sua fonte de dados e configure os parâmetros necessários.

   • Resource group connectivity: Fontes de dados RestAPI aceitam apenas grupos de recursos exclusivos para Data Integration. Selecione o grupo adequado e teste a conectividade entre a fonte de dados e o recurso.

2. Crie um nó de sincronização em lote e configure a tarefa.

   Crie um nó de sincronização em lote no Data Studio do DataWorks. Para mais detalhes, veja Criar um nó de sincronização em lote. As configurações essenciais incluem:

   • Configurações de Data Source:

     • Data Source: Selecione a fonte de dados RestAPI criada na etapa anterior.

     • Request Method: Como a API de exemplo é do tipo GET, selecione GET.

     • Return Data Structure: A resposta da API é um array JSON. Escolha Array Data.

     • The JSON path for data storage: Os dados retornados estão sob o campo data. Defina como data.

     • Request Param: Combine parâmetros de requisição com parâmetros de agendamento para sincronizar diariamente os dados do dia corrente.

       • Defina os parâmetros de requisição como startTime=${extract_day} ${start_time}&endTime=${extract_day} ${end_time}

       • Nas configurações de agendamento, adicione três parâmetros: extract_day=${yyyy-mm-dd}, start_time=00:00:00 e end_time=23:59:59.

       Se a data de execução for 2023-01-05, o valor de extract_day será 2023-01-04, resultando na concatenação: startTime=2023-01-04 00:00:00&endTime=2023-01-04 23:59:59.

   • Configurações de Data Destination:

     • Data Source e Table: Selecione a tabela particionada do MaxCompute criada anteriormente.

     • Partition Information: Utilize informações de partição combinadas com parâmetros de agendamento.

       • Defina a informação da partição como ${bizdate}.

       • Nas configurações de agendamento, adicione o parâmetro: bizdate=$bizdate.

       Para uma execução em 2023-01-05, o valor da partição será 20230104.

   • Configurações de Field Mapping: Com base na definição da API, insira os nomes das colunas da resposta RestAPI. Os nomes diferenciam maiúsculas de minúsculas. Após adicionar as colunas, utilize The same name mapping ou desenhe linhas manualmente para criar os mapeamentos.

Execução de teste

Verificação de dados

select * from ods_xiaobo_rest2 where ds='20230104' order by createtime;

Enviar, implantar e retroalimentar dados

Prática 2: Ler dados de uma RestAPI paginada

Cenário de exemplo: Definição da API

Preparação: Criar uma tabela particionada no MaxCompute

CREATE TABLE IF NOT EXISTS ods_xiaobo_rest1
(
  `id` BIGINT
  ,`name` STRING
)
PARTITIONED BY
(
  ds  STRING
)
LIFECYCLE 3650;

Configurar a tarefa de sincronização

1. Adicione uma fonte de dados RestAPI.

   Adicione a fonte de dados RestAPI no workspace do DataWorks. Consulte Adicionar uma fonte de dados RestAPI para mais detalhes. Na caixa de diálogo Add RestAPI Data Source, preencha Data Source Name e Data Source Description, escolha Applicable Environment (desenvolvimento ou produção) e configure Default Request Header (padrão: {}). As configurações chave são:

   • url: Informe o endereço da API RESTful.

   • Authentication Method: Selecione o método de autenticação compatível com sua API e configure os parâmetros exigidos.

   • Resource group connectivity: Apenas grupos de recursos exclusivos para Data Integration são suportados. Escolha o grupo e teste a conexão.

2. Crie um nó de sincronização em lote e configure a tarefa.

   No Data Studio do DataWorks, crie o nó de sincronização em lote. Veja Criar um nó de sincronização em lote para instruções completas. Configure os itens principais conforme abaixo.

   • Configurações de Data Source:

     • Data Source: Escolha a fonte RestAPI criada previamente.

     • Request Method: Sendo uma API GET, selecione GET.

     • Return Data Structure: Como a resposta é um array JSON, marque Array Data.

     • The JSON path for data storage: Os dados residem no campo data. Preencha com data.

     • Request Param: O tamanho da página é fixo. Defina como pageSize=50. Evite páginas muito grandes para não sobrecarregar o servidor RestAPI e a própria tarefa de sincronização.

     • The number of requests: Para este exemplo, selecione Multiple Requests.

       O parâmetro de paginação desta API é pageNum. Após escolher Multiple Requests, configure os campos relacionados:

       • Parameter used for multiple requests: Defina como pageNum.

       • StartIndex: Defina como 1.

       • Step: Defina como 1.

       • EndIndex: Defina como 100.

   • Configurações de Data Destination:

     • Data Source e Table: Selecione a tabela particionada do MaxCompute criada antes.

     • Partition Information: Combine a partição com parâmetros de agendamento.

       • Configure a partição como ${bizdate}.

       • Adicione nas configurações de agendamento o parâmetro: bizdate=$bizdate.

       Se executado em 2023-01-05, o valor da partição será 20230104.

   • Configurações de Field Mapping: Insira os nomes das colunas conforme a resposta da RestAPI, respeitando maiúsculas e minúsculas. Use The same name mapping ou faça o mapeamento manual ligando as colunas.

Execução de teste

Verificação de dados

select * from ods_xiaobo_rest1 where ds='20230104' order by id;

Prática 3: Ler dados de uma RestAPI do tipo POST

Cenário de exemplo: Definição da API

Preparação: Criar uma tabela particionada no MaxCompute

CREATE TABLE IF NOT EXISTS ods_xiaobo_rest3
(
  `user_id` BIGINT
  ,`name` STRING
  ,`axis`  STRING
  ,`value` BIGINT
  ,`create_time` STRING
)
PARTITIONED BY
(
  ds  STRING
)
LIFECYCLE 3650;

Configurar a tarefa de sincronização

1. Adicione uma fonte de dados RestAPI.

   Registre a fonte de dados RestAPI no workspace do DataWorks. Siga as instruções em Adicionar uma fonte de dados RestAPI. Preencha Data Source Name, Data Source Description, selecione Applicable Environment e ajuste Default Request Header (valor padrão: {}). Pontos importantes:

   • url: Endereço da API RESTful.

   • Authentication Method: Método de autenticação suportado pela API e respectivos parâmetros.

   • Resource group connectivity: Use grupos de recursos exclusivos para Data Integration e valide a conectividade.

2. Crie um nó de sincronização em lote e configure a tarefa.

   Crie o nó no Data Studio do DataWorks conforme descrito em Criar um nó de sincronização em lote. Atenção aos seguintes ajustes:

   • Configurações de Data Source:

     • Data Source: Fonte RestAPI criada anteriormente.

     • Request Method: Selecione POST.

     • Return Data Structure: Marque Array Data pois a resposta é um array JSON.

     • The JSON path for data storage: Defina como data.

     • Header: Esta API POST aceita JSON no corpo. Configure como {"Content-Type":"application/json"}.

     • Request Param: Use parâmetros de requisição aliados aos de agendamento para sincronizar dados diários.

       • Configure os parâmetros de requisição como:

         {
         
             "userId":16,
         
             "startTime":"${extract_day} 00:00:00",
         
             "endTime":"${extract_day} 23:59:59"
         
         }

       • Adicione nas configurações de agendamento: extract_day=${yyyy-mm-dd}.

   • Configurações de Data Destination:

     • Data Source e Table: Tabela particionada do MaxCompute criada previamente.

     • Partition Information: Vincule a partição aos parâmetros de agendamento.

       • Valor da partição: ${bizdate}.

       • Parâmetro de agendamento: bizdate=$bizdate.

       Executando em 2023-01-05, a partição assumirá o valor 20230104.

   • Configurações de Field Mapping: Mapeie as colunas da resposta RestAPI conforme a definição da API. Use pontos (.) para separar colunas aninhadas. Respeite maiúsculas e minúsculas. Utilize The same name mapping ou mapeamento manual.

Execução de teste

Verificação de dados

select * from ods_xiaobo_rest3 where ds='20230105' order by create_time;

Prática 4: Iterar sobre um parâmetro de requisição para ler de uma RestAPI

Cenário de exemplo: Definição da API

Nesta prática, você lê dados iterativamente de uma API RESTful e os grava em uma tabela particionada do MaxCompute. A API de exemplo é uma GET de teste auto-hospedada que retorna dados meteorológicos com base em data, província e cidade.

Adapte as configurações à sua API real durante a implementação. Este exemplo visa apenas demonstrar o fluxo de trabalho.

• Solicitação de exemplo:

  http://TestAPIAddress:Port/rest/test5?date=2023-01-04&province=zhejiang&city=hangzhou

• Resposta de exemplo:

  {
    "province": "P1",
    "city": "hz",
    "date": "2023-01-04",
    "minTemperature": "-14",
    "maxTemperature": "-7",
    "unit": "℃",
    "weather": "cool"
  }

• Chamada de API de exemplo em ferramenta de teste: Via Postman, envie uma requisição GET com date=2023-01-04, province=p1 e city=hz. A resposta JSON incluirá minTemperature (-14), maxTemperature (-7), unit (℃) e weather (COOL).

Preparação: Criar uma tabela de parâmetros e uma tabela particionada no MaxCompute

Os dados lidos serão sincronizados para uma tabela particionada do MaxCompute. Crie primeiro uma tabela de parâmetros para armazenar os valores de província e cidade usados na iteração e, em seguida, a tabela particionada de destino.

O uso de tabelas particionadas com overwrite evita duplicidade e facilita análises.

As instruções CREATE TABLE são:

CREATE TABLE IF NOT EXISTS `citys`
(
  `province` STRING ,
  `city` STRING
);

insert into citys
select 'shanghai','shanghai'
union all select 'zhejiang','hangzhou'
union all select 'sichuan','chengdu';

CREATE TABLE IF NOT EXISTS ods_xiaobo_rest5
(
    `minTemperature` STRING ,
    `maxTemperature` STRING ,
    `unit` STRING ,
    `weather` STRING 
)
PARTITIONED BY 
(
    `province` STRING ,
    `city` STRING ,
    `ds`  STRING
)
LIFECYCLE 3650;

Após enviar as tabelas para produção no DataWorks Standard Edition, visualize-as no Data Map.

Configurar a tarefa de sincronização

1. Adicione uma fonte de dados RestAPI.

   Registre a fonte RestAPI no workspace do DataWorks. Consulte Adicionar uma fonte de dados RestAPI. Na caixa Add RestAPI Data Source, informe Data Source Name, Data Source Description, selecione Applicable Environment e configure Default Request Header (padrão: {}). Configurações principais:

   • url: Endereço da API RESTful.

   • Authentication Method: Método de autenticação e parâmetros correspondentes.

   • Resource group connectivity: Selecione o grupo de recursos e teste a conexão.

2. Crie um nó de atribuição chamado setval_citys no DataStudio. Veja Criar um nó de atribuição.

   Configurações essenciais:

   Nº	Descrição
   ①	Linguagem de atribuição: ODPS SQL Código de atribuição: SELECT province ,city FROM citys;
   ②	Propriedade de reexecução: Defina como You can run again after successful or failed operation..

   Concluída a configuração, envie e implante o nó de atribuição.

3. Crie um nó for-each no DataStudio. Consulte Criar um nó for-each. Principais configurações:

   Nº	Descrição
   ①	Propriedade de reexecução: Defina como You can run again after successful or failed operation..
   ②	Dependência de nó upstream: Selecione o nó setval_citys criado na etapa anterior.
   ③	Parâmetro de contexto do nó: Escolha a origem do parâmetro de entrada.
   ④	Nó de sincronização em lote: Configure o nó de sincronização dentro do nó for-each conforme as etapas a seguir.

4. Crie e configure o nó de sincronização em lote. Detalhes em Criar um nó de sincronização em lote.

   Pontos de configuração:

   Nº	Descrição
   ①	Configure os parâmetros de agendamento: bizdate=$[yyyymmdd-1] bizdate_year=$[yyyy-1] bizdate_month=$[mm-1] bizdate_day=$[dd-1]
   ②	Configure os parâmetros de requisição da RestAPI. Province e city vêm do nó for-each. date=${bizdate_year}-${bizdate_month}-${bizdate_day}&province=${dag.foreach.current[0]}&city=${dag.foreach.current[1]}
   ③	Configure o parâmetro de partição do MaxCompute. O parâmetro province vem do nó for-each. province=${dag.foreach.current[0]}
   ④	Configure o parâmetro de partição do MaxCompute. O parâmetro city vem do nó for-each. city=${dag.foreach.current[1]}
   ⑤	Configure o parâmetro de partição do MaxCompute. O parâmetro ds vem dos parâmetros de agendamento. ds=${bizdate}
   ⑥	Insira os nomes das colunas da resposta RestAPI conforme a definição da API. Respeite maiúsculas e minúsculas. Use The same name mapping ou mapeamento manual.

   Finalizada a configuração, envie e implante o nó for-each.

Execução de teste

1. Após enviar e implantar com sucesso os nós de atribuição e for-each, acesse as tarefas agendadas no Operation Center e execute uma operação de retroalimentação de dados no nó de atribuição. Consulte Retroalimentar dados.

2. Selecione a data comercial para a tarefa de retroalimentação e os nós downstream conforme seu cenário.

3. Após a execução, verifique os parâmetros nos detalhes da instância de retroalimentação e analise o log de execução para garantir a correção.

   Neste exemplo, os dados são gravados na partição province=shanghai,city=shanghai,ds=20231215 da tabela do MaxCompute.

Verificação de dados

Execute uma consulta ad hoc no Data Studio para validar a sincronização no MaxCompute. Exemplo de consulta:

No exemplo, ods_xiaobo_rest5 é a tabela particionada criada na etapa de Preparação.

SELECT  weather
        ,mintemperature
        ,maxtemperature
        ,unit
        ,province
        ,city
        ,ds
FROM    ods_xiaobo_rest5
WHERE   ds != 1
ORDER BY ds,province,city;

Após a consulta, confirme se os dados foram sincronizados corretamente para o MaxCompute.