MaxCompute:Use o Kettle para agendar jobs do MaxCompute

Informações básicas

O Kettle é uma ferramenta de ETL open source escrita em Java. Ele roda em Windows, Unix e Linux, oferecendo uma GUI para construir pipelines de dados. O Kettle suporta uma ampla variedade de fontes de dados, incluindo bancos como Oracle, MySQL e DB2, além de sistemas de big data como HDFS, HBase, Cassandra e MongoDB.

Crie um job no Kettle para se conectar a um projeto do MaxCompute e, em seguida, agende esse job como parte de um fluxo de trabalho de ETL.

Pré-requisitos

Procedimento

1. Etapa 1: Colocar o driver JDBC do MaxCompute

   Coloque o driver JDBC do MaxCompute no diretório de drivers do Kettle.

2. Etapa 2: Conectar-se a um projeto do MaxCompute

   Conecte o Kettle ao seu projeto do MaxCompute.

3. Etapa 3: Criar um fluxo de trabalho de agendamento

   Crie um fluxo de trabalho de agendamento de jobs na interface Spoon.

4. Etapa 4: Executar o fluxo de trabalho de agendamento

   Execute o fluxo de trabalho de agendamento de jobs.

5. Etapa 5: Visualizar os resultados do job

   Visualize os resultados do job.

Etapa 1: Colocar o driver JDBC do MaxCompute

Coloque o arquivo JAR do driver JDBC do MaxCompute, como odps-jdbc-3.2.9-jar-with-dependencies.jar, no diretório data-integration/lib da sua instalação do Kettle.

Etapa 2: Conectar-se a um projeto do MaxCompute

1. No diretório data-integration da sua instalação do Kettle, clique duas vezes em Spoon.bat (no Windows) ou Spoon (no macOS) para iniciar o Spoon.

2. Na barra de menu superior, selecione File > New > Job para criar um job do Kettle destinado ao fluxo de trabalho de agendamento.

3. Na aba View, clique em com o botão direito em Database connections na árvore de navegação e selecione Create.

4. Na caixa de diálogo Database Connection, na aba General, configure os parâmetros conforme descrito na tabela a seguir.

   Parâmetro	Descrição
   Network Connection Name	Nome personalizado para a conexão de banco de dados. Por exemplo, MaxCompute.
   Connection type	Selecione Generic database.
   Connection Method	Selecione Native (JDBC).
   Dialect	Selecione Hadoop Hive 2.
   Custom Connection URL	URL de conexão para o projeto do MaxCompute. O formato é jdbc:odps:<maxcompute_endpoint>?project=<MaxCompute_project_name>. Remova os símbolos <> ao configurar a URL. Parâmetros: <MaxCompute_endpoint>: Obrigatório. O endpoint da região onde seu projeto do MaxCompute está localizado. Para obter uma lista de endpoints, consulte Endpoints. <MaxCompute_project_name>: Obrigatório. O nome do projeto do MaxCompute de destino. Este é o nome do seu projeto do MaxCompute, não de um workspace. Faça login no MaxCompute console, alterne a região no canto superior esquerdo e localize o nome do projeto na página Projects.
   Custom Driver Class Name	Classe do driver JDBC. Defina como com.aliyun.odps.jdbc.OdpsDriver.
   User Name	Seu AccessKey ID para o projeto do MaxCompute. Acesse a página AccessKey Management para obter o AccessKey ID.
   Password	Seu AccessKey secret.

5. Clique em Test. Após estabelecer a conexão com sucesso, clique em OK na caixa de diálogo de confirmação. Em seguida, clique em Confirm na caixa de diálogo Confirm para salvar a conexão.

Etapa 3: Criar um fluxo de trabalho de agendamento

Na aba Design do Spoon, construa um fluxo de trabalho de agendamento de jobs criando e vinculando objetos principais.

O exemplo a seguir detalha um processo de ETL que utiliza o comando LOAD para carregar dados do OSS em uma tabela interna do MaxCompute. Para os dados de amostra, consulte LOAD. Os jobs neste processo são divididos por tipo de objeto principal conforme descrito abaixo.

1. Na interface do Spoon, clique em na aba Design.

2. Com base na divisão de objetos anterior, arraste os objetos principais do painel de navegação à esquerda para a tela de jobs à direita. Conecte os objetos principais seguindo a estrutura abaixo.

   Para conectar objetos principais, selecione um objeto de origem, pressione e segure Shift e clique em no objeto de destino para criar um salto de conexão.

   Selecione a aba Core Objects no painel esquerdo e expanda General e outras categorias para obter componentes. A estrutura de conexão é: Start → Create table → Load from OSS → Processing → Success (caminho de sucesso); configure também caminhos de falha de Create table, Load from OSS e Processing para o nó Abort job para tratamento de erros.

3. Clique em com o botão direito em um objeto principal baseado em script e selecione Edit job entry. Na caixa de diálogo SQL, configure os parâmetros listados na tabela a seguir e clique em OK. Repita esse processo para configurar todos os objetos principais baseados em script.

   Parâmetro	Descrição
   Job entry name	Nome da entrada do job. Por exemplo, Create table, Load from OSS ou Processing.
   Connection	Nome da conexão de banco de dados a ser usada. Refere-se à conexão criada na Etapa 2. Por exemplo, MaxCompute.
   Send SQL as single statement?	Desmarque esta opção.
   SQL Script	Script SQL para a entrada do job. Os scripts SQL para os objetos principais baseados em script neste exemplo são: Create table CREATE TABLE ambulance_data_csv_load ( vehicleId INT, recordId INT, patientId INT, calls INT, locationLatitute DOUBLE, locationLongtitue DOUBLE, recordTime STRING, direction STRING); Load from OSS LOAD OVERWRITE TABLE ambulance_data_csv_load FROM LOCATION 'oss://oss-cn-hangzhou-internal.aliyuncs.com/mc-test/data_location/' STORED BY 'com.aliyun.odps.CsvStorageHandler' WITH serdeproperties ( 'odps.properties.rolearn'='acs:ram::xxxxx:role/aliyunodpsdefaultrole', -- The ARN of AliyunODPSDefaultRole. You can obtain it from the RAM console. 'odps.text.option.delimiter'=',' ); Processing INSERT OVERWRITE TABLE ambulance_data_csv SELECT * FROM ambulance_data_csv_load;

Etapa 4: Executar o fluxo de trabalho de agendamento

1. Na interface do fluxo de trabalho de agendamento de jobs, clique em no ícone de execução no canto superior esquerdo. Na caixa de diálogo Run Options, clique em Execute.

   Na caixa de diálogo, defina Run configuration como Pentaho local, Log level como Basic logging, mantenha Clear log before execution marcado por padrão e deixe as demais opções com seus valores padrão.

2. Opcional: Se for solicitado salvar o job, clique em Yes e nomeie o fluxo de trabalho de agendamento de jobs (por exemplo, mc).

3. Monitore o status de execução no diagrama DAG na interface do fluxo de trabalho ou na seção Execution Results. O fluxo de trabalho de agendamento de jobs estará concluído quando uma marca de verificação verde aparecer em cada objeto.

   Ao término do fluxo de agendamento de jobs, o DAG exibirá marcas de verificação verdes nos nós Create table, Load from OSS, Processing e Success. A área Execution Results mostrará o resultado de cada entrada de job como verdadeiro, com a saída final Job has ended.

Etapa 5: Visualizar os resultados do job

Após a conclusão do fluxo de trabalho de agendamento de jobs, execute uma consulta SQL para verificar se os dados foram gravados na tabela de destino conforme esperado.

1. Na interface do Spoon, clique em na aba View. Sob o job do Kettle criado (por exemplo, mc), expanda Database connections.

2. Clique em com o botão direito na conexão de banco de dados criada (por exemplo, MaxCompute) e selecione SQL Editor.

3. Na caixa de diálogo Simple SQL editor, insira uma consulta SQL e clique em Execute. Os resultados aparecerão na caixa de diálogo Examine preview data.

   A consulta SQL é a seguinte:

   SELECT * FROM ambulance_data_csv;