Migrar RDS para MaxCompute com particionamento dinâmico - MaxCompute: Central de Documentação do Alibaba Cloud

Utilize o DataWorks Data Integration para criar partições automaticamente durante a migração de dados do ApsaraDB RDS para o MaxCompute.

Pré-requisitos

Configure um ambiente no DataWorks.
1. Activate MaxCompute.
2. Crie um fluxo de trabalho no DataWorks. Este tutorial utiliza o modo básico do DataWorks. Para mais informações, consulte Create a workflow.
Adicione as fontes de dados.
- Adicione uma fonte de dados MySQL como source. Para mais informações, consulte Configure a MySQL data source.
- Adicione uma fonte de dados MaxCompute como destino para receber os dados do RDS. Para mais informações, consulte Configure MaxCompute data source.

Criar partições automaticamente

Agende uma tarefa diária para sincronizar dados do ApsaraDB RDS com o MaxCompute e criar partições baseadas em data automaticamente. Para mais detalhes sobre tarefas de sincronização de dados, consulte Data Development and O&M.

Nota

Este exemplo utiliza o modo simples do DataWorks. Ao criar um workspace, a opção Join the public preview of Data Development (Data Studio) vem desativada por padrão, e workspaces em preview público não se aplicam a este exemplo.

Faça login no console do DataWorks.
Crie uma tabela de destino no MaxCompute.
1. No painel de navegação à esquerda, clique em Workspace.
2. Na coluna Actions do seu workspace, clique em Quick Access > Data Development.
3. Clique com o botão direito no fluxo de trabalho criado e escolha New table > MaxCompute > Table.
4. Na página Create Table, selecione uma instância de engine, um schema e um caminho. Insira um Table Name e clique em Create.
5. Na página de edição da tabela, clique no ícone para alternar para o modo DDL.
6. Na caixa de diálogo DDL, insira a seguinte instrução e clique em Generate Table Schema. Na caixa de diálogo Confirm operation exibida, clique em OK.
```
CREATE TABLE IF NOT EXISTS ods_user_info_d (
uid STRING COMMENT 'User ID',
gender STRING COMMENT 'Gender',
age_range STRING COMMENT 'Age range',
zodiac STRING COMMENT 'Zodiac sign'
)
PARTITIONED BY (
dt STRING
);                           
```
7. Clique em Submit to Production Environment.
Crie um nó de sincronização offline.
1. Acesse a página de análise de dados. Clique com o botão direito no fluxo de trabalho especificado e escolha Create Node > Data Integration > Offline synchronization.
2. Na caixa de diálogo Create Node, insira o Name e clique em Confirm.
3. Selecione a fonte de dados, o grupo de recursos e o destino, e então teste a conexão.
  - Source: A fonte de dados MySQL que você criou.
  - Resource Group: Selecione um grupo de recursos exclusivo para Data Integration.
  - Destination: A fonte de dados MaxCompute que você criou.
4. Clique em Next para configurar a tarefa. Para o source, selecione a tabela ods_user_info_d e defina a chave de divisão como uid. Para o destino, selecione a tabela ods_user_info_d e insira dt=${bizdate} nas informações de partição. Defina o grupo de recursos do Tunnel como public transport resource e o schema como default. No modo de escrita, selecione Insert Overwrite (Clear existing data before writing) e, para gravar strings vazias como nulo, selecione No.
Configure os parâmetros de agendamento.
1. No painel de navegação à direita, clique em Properties.
2. Na seção Scheduling Parameter, o parâmetro padrão é ${bizdate}, que utiliza o formato aaaammdd.
  Nota
  O valor padrão do parâmetro corresponde ao Destination configurado nas Partition Information. O sistema substitui automaticamente o valor da partição pela data de negócio, que corresponde ao dia anterior à execução da tarefa, pois jobs ETL geralmente processam dados do dia precedente. Para utilizar a data atual de execução, personalize o parâmetro.
  É possível personalizar formatos de data com as seguintes expressões:
  - N anos depois: $[add_months(yyyymmdd,12*N)]
  - N anos antes: $[add_months(yyyymmdd,-12*N)]
  - N meses antes: $[add_months(yyyymmdd,-N)]
  - N semanas depois: $[yyyymmdd+7*N]
  - N meses depois: $[add_months(yyyymmdd,N)]
  - N semanas antes: $[yyyymmdd-7*N]
  - N dias depois: $[yyyymmdd+N]
  - N dias antes: $[yyyymmdd-N]
  - N horas depois: $[hh24miss+N/24]
  - N horas antes: $[hh24miss-N/24]
  - N minutos depois: $[hh24miss+N/24/60]
  - N minutos antes: $[hh24miss-N/24/60]
  Nota
  
  Utilize colchetes ([]) para definir a fórmula de cálculo de uma variável personalizada. Por exemplo: key1=$[yyyy-mm-dd].
  
  Por padrão, a unidade de cálculo para variáveis personalizadas é dias. Por exemplo, $[hh24miss-N/24/60] representa o resultado de (yyyymmddhh24miss - (N/24/60 * 1 day)), formatado como hh24miss.
  
  A unidade de cálculo da função add_months é meses. Por exemplo, $[add_months(yyyymmdd,12*N)-M/24/60] representa o resultado de (yyyymmddhh24miss - (12 * N * 1 month) - (M/24/60 * 1 day)), formatado como yyyymmdd.
Clique no ícone para executar o código.
Visualize os resultados no Runtime Log.

Reprocessar dados históricos

Para sincronizar dados históricos anteriores à tarefa agendada, utilize o recurso Complement data no Operation Center do DataWorks. Isso permite sincronizar os dados e criar partições automaticamente.

Filtre os dados históricos da fonte ApsaraDB RDS por data.

Na seção Source do nó de sincronização, defina a condição de Data Filter como ${bizdate}.
Execute o reprocessamento de dados. Para mais informações, consulte Manage data backfill instances.

Verifique os resultados de extração dos dados do ApsaraDB RDS no log de execução.

O log de execução mostra que o MaxCompute criou a partição automaticamente. A cláusula where na configuração do leitor (where=[20180913]) corresponde à partition na configuração do gravador (partition=[dt=20180913]

Alibaba DI Console, Build 201805310000 .
Copyright 2018 Alibaba Group, All rights reserved .
Start Job[16961870], traceId [283789484710656#79023#None#None#228255635341196741#None#None#rds_sync], running in Pipeline[basecommon_
89484710656]
The Job[16961870] will run in PhysicsPipeline [basecommon_group_283789484710656_oxs] with requestId [4f44180d-300c-47c3-8ea3-805d2
2018-12-02 03:31:25 : ---
Reader: mysql
                column=["uid","gender","age_range","zodiac"]
                connection=[{"datasource":"xxx","table":["`ods_user_info_d`"]}]]
                where=[20180913                          ]
                splitPk=[uid                             ]
Writer: odps
                isCompress=[false                        ]
                partition=[dt=20180913                    ]
                truncate=[true                           ]
                datasource=[odps_first                   ]
                column=["uid","gender","age_range","zodiac"]
                emptyAsNull=[false                       ]
                table=[ods_user_info_d                    ]
Setting:
                errorLimit=[{"record":""}                 ]
                speed=[{"concurrent":1,"dmu":1,"mbps":"10","throttle":true}]
2018-12-02 03:31:26 : State: 1(SUBMIT) | Total: 0R 0B | Speed: 0R/s 0B/s | Error: 0R 0B | Stage: 0.0%
2018-12-02 03:31:36 : State: 3(RUN) | Total: 0R 0B | Speed: 0R/s 0B/s | Error: 0R 0B | Stage: 0.0%

Valide o resultado. Execute o comando abaixo no cliente MaxCompute para verificar se os dados foram gravados.
```
SELECT count(*) from ods_user_info_d where dt = 20180913;
```

Particionar campos não relacionados a data usando hash

Caso precise particionar dados por um campo que não seja data, como província, o Data Integration não consegue realizar o particionamento automático. Como alternativa, aplique hash em um campo do RDS para armazenar registros com o mesmo valor desse campo na partição correspondente do MaxCompute.

Sincronize todos os dados para uma tabela temporária no MaxCompute e crie um nó de script SQL. Execute o seguinte comando.

drop table if exists ods_user_t;
CREATE TABLE ods_user_t ( 
        dt STRING,
        uid STRING,
        gender STRING,
        age_range STRING,
        zodiac STRING);
--Store the data from the MaxCompute table in the temporary table.
insert overwrite table ods_user_t select dt,uid,gender,age_range,zodiac from ods_user_info_d;

Crie um nó de tarefa de sincronização chamado mysql_to_odps. Trata-se de uma tarefa simples para sincronizar todos os dados do RDS com o MaxCompute, sem necessidade de partições.

Utilize uma instrução SQL para realizar o particionamento dinâmico na tabela de destino. O comando é o seguinte.

drop table if exists ods_user_d;
//Create a partitioned ODPS table. This is the final destination table.
    CREATE TABLE ods_user_d (
    uid STRING,
        gender STRING,
        age_range STRING,
        zodiac STRING
)
PARTITIONED BY (
    dt STRING
);
//Run the dynamic partitioning SQL statement. It automatically partitions data based on the dt field of the temporary table. Records with the same value in the dt field are placed in a partition created for that value.
//For example, if some records have the value 20181025 in the dt field, a partition dt=20181025 is automatically created in the ODPS partitioned table.
//The dynamic partitioning SQL is as follows.
//Note that the dt field is included in the SELECT statement. This specifies that partitions are automatically created based on this field.
insert overwrite table ods_user_d partition(dt)select dt,uid,gender,age_range,zodiac from ods_user_t;
//After the import is complete, delete the temporary table to save storage costs.
drop table if exists ods_user_t;

No MaxCompute, é possível usar instruções SQL para sincronizar dados.

Configure os três nós em um fluxo de trabalho para execução sequencial.

A ordem de execução é tabela temporária, mysql_to_odps e tabela de destino.

Acompanhe o processo de execução. Concentre-se no processo de particionamento dinâmico do último nó.

=20181203065434115g3a2eqsa
Log view:
http://logview.odps.aliyun.com/logview/?h=http://service.odps.aliyun.com/api&p=DataWorks_DOC&i=20181203065434115g3a2eqsa&token=V0NaNDFxUmpn...
Summary:
resource cost: cpu 0.00 Core * Min, memory 0.00 GB * Min
inputs:
  dataworks_doc.ods_user_t: 20028 (119496 bytes)
outputs:
  dataworks_doc.ods_use xxx 20028 (119176 bytes)
Job run time: 0.000
Job run mode: service job
Job run engine: execution engine
M1:
  instance count: 1
  run time: 0.000
  instance time:
    min: 0.000, max: 0.000, avg: 0.000
  input records:
    ...

Valide o resultado. Execute o comando abaixo no cliente MaxCompute para verificar os dados gravados.
```
SELECT count(*) from ods_user_d where dt = 20180913;
```